* Refactored server to use spaces
* Other cleanup
This commit is contained in:
Zef Hemel
@@ -0,0 +1,142 @@
|
||||
import { mkdir, readdir, readFile, stat, unlink, utimes, writeFile } from "fs/promises";
|
||||
import * as path from "path";
|
||||
import { PageMeta } from "../types";
|
||||
import { SpacePrimitives } from "./space_primitives";
|
||||
import { Plug } from "../../plugos/plug";
|
||||
|
||||
export class DiskSpacePrimitives implements SpacePrimitives {
|
||||
rootPath: string;
|
||||
plugPrefix: string;
|
||||
|
||||
constructor(rootPath: string, plugPrefix: string = "_plug/") {
|
||||
this.rootPath = rootPath;
|
||||
this.plugPrefix = plugPrefix;
|
||||
}
|
||||
|
||||
pageNameToPath(pageName: string) {
|
||||
if (pageName.startsWith(this.plugPrefix)) {
|
||||
return path.join(this.rootPath, pageName + ".plug.json");
|
||||
}
|
||||
return path.join(this.rootPath, pageName + ".md");
|
||||
}
|
||||
|
||||
pathToPageName(fullPath: string): string {
|
||||
let extLength = fullPath.endsWith(".plug.json")
|
||||
? ".plug.json".length
|
||||
: ".md".length;
|
||||
return fullPath.substring(
|
||||
this.rootPath.length + 1,
|
||||
fullPath.length - extLength
|
||||
);
|
||||
}
|
||||
|
||||
async readPage(pageName: string): Promise<{ text: string; meta: PageMeta }> {
|
||||
const localPath = this.pageNameToPath(pageName);
|
||||
try {
|
||||
const s = await stat(localPath);
|
||||
return {
|
||||
text: await readFile(localPath, "utf8"),
|
||||
meta: {
|
||||
name: pageName,
|
||||
lastModified: s.mtime.getTime(),
|
||||
},
|
||||
};
|
||||
} catch (e) {
|
||||
// console.error("Error while reading page", pageName, e);
|
||||
throw Error(`Could not read page ${pageName}`);
|
||||
}
|
||||
}
|
||||
|
||||
async writePage(
|
||||
pageName: string,
|
||||
text: string,
|
||||
selfUpdate: boolean,
|
||||
lastModified?: number
|
||||
): Promise<PageMeta> {
|
||||
let localPath = this.pageNameToPath(pageName);
|
||||
try {
|
||||
// Ensure parent folder exists
|
||||
await mkdir(path.dirname(localPath), { recursive: true });
|
||||
|
||||
// Actually write the file
|
||||
await writeFile(localPath, text);
|
||||
|
||||
if (lastModified) {
|
||||
let d = new Date(lastModified);
|
||||
console.log("Going to set the modified time", d);
|
||||
await utimes(localPath, d, d);
|
||||
}
|
||||
// Fetch new metadata
|
||||
const s = await stat(localPath);
|
||||
return {
|
||||
name: pageName,
|
||||
lastModified: s.mtime.getTime(),
|
||||
};
|
||||
} catch (e) {
|
||||
console.error("Error while writing page", pageName, e);
|
||||
throw Error(`Could not write ${pageName}`);
|
||||
}
|
||||
}
|
||||
|
||||
async getPageMeta(pageName: string): Promise<PageMeta> {
|
||||
let localPath = this.pageNameToPath(pageName);
|
||||
try {
|
||||
const s = await stat(localPath);
|
||||
return {
|
||||
name: pageName,
|
||||
lastModified: s.mtime.getTime(),
|
||||
};
|
||||
} catch (e) {
|
||||
console.error("Error while getting page meta", pageName, e);
|
||||
throw Error(`Could not get meta for ${pageName}`);
|
||||
}
|
||||
}
|
||||
|
||||
async deletePage(pageName: string): Promise<void> {
|
||||
let localPath = this.pageNameToPath(pageName);
|
||||
await unlink(localPath);
|
||||
}
|
||||
|
||||
async fetchPageList(): Promise<{
|
||||
pages: Set<PageMeta>;
|
||||
nowTimestamp: number;
|
||||
}> {
|
||||
let pages = new Set<PageMeta>();
|
||||
|
||||
const walkPath = async (dir: string) => {
|
||||
let files = await readdir(dir);
|
||||
for (let file of files) {
|
||||
const fullPath = path.join(dir, file);
|
||||
let s = await stat(fullPath);
|
||||
if (s.isDirectory()) {
|
||||
await walkPath(fullPath);
|
||||
} else {
|
||||
if (file.endsWith(".md") || file.endsWith(".json")) {
|
||||
pages.add({
|
||||
name: this.pathToPageName(fullPath),
|
||||
lastModified: s.mtime.getTime(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
await walkPath(this.rootPath);
|
||||
return {
|
||||
pages: pages,
|
||||
nowTimestamp: Date.now(),
|
||||
};
|
||||
}
|
||||
|
||||
invokeFunction(
|
||||
plug: Plug<any>,
|
||||
env: string,
|
||||
name: string,
|
||||
args: any[]
|
||||
): Promise<any> {
|
||||
return plug.invoke(name, args);
|
||||
}
|
||||
|
||||
proxySyscall(plug: Plug<any>, name: string, args: any[]): Promise<any> {
|
||||
return plug.syscall(name, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
import { SpacePrimitives } from "./space_primitives";
|
||||
import { EventHook } from "../../plugos/hooks/event";
|
||||
import { PageMeta } from "../types";
|
||||
import { Plug } from "../../plugos/plug";
|
||||
|
||||
export class EventedSpacePrimitives implements SpacePrimitives {
|
||||
constructor(private wrapped: SpacePrimitives, private eventHook: EventHook) {}
|
||||
|
||||
fetchPageList(): Promise<{ pages: Set<PageMeta>; nowTimestamp: number }> {
|
||||
return this.wrapped.fetchPageList();
|
||||
}
|
||||
|
||||
proxySyscall(plug: Plug<any>, name: string, args: any[]): Promise<any> {
|
||||
return this.wrapped.proxySyscall(plug, name, args);
|
||||
}
|
||||
|
||||
invokeFunction(
|
||||
plug: Plug<any>,
|
||||
env: string,
|
||||
name: string,
|
||||
args: any[]
|
||||
): Promise<any> {
|
||||
return this.wrapped.invokeFunction(plug, env, name, args);
|
||||
}
|
||||
|
||||
readPage(pageName: string): Promise<{ text: string; meta: PageMeta }> {
|
||||
return this.wrapped.readPage(pageName);
|
||||
}
|
||||
|
||||
async writePage(
|
||||
pageName: string,
|
||||
text: string,
|
||||
selfUpdate: boolean,
|
||||
lastModified?: number
|
||||
): Promise<PageMeta> {
|
||||
const newPageMeta = await this.wrapped.writePage(
|
||||
pageName,
|
||||
text,
|
||||
selfUpdate,
|
||||
lastModified
|
||||
);
|
||||
// This can happen async
|
||||
this.eventHook
|
||||
.dispatchEvent("page:saved", pageName)
|
||||
.then(() => {
|
||||
return this.eventHook.dispatchEvent("page:index", {
|
||||
name: pageName,
|
||||
text: text,
|
||||
});
|
||||
})
|
||||
.catch((e) => {
|
||||
console.error("Error dispatching page:saved event", e);
|
||||
});
|
||||
return newPageMeta;
|
||||
}
|
||||
|
||||
getPageMeta(pageName: string): Promise<PageMeta> {
|
||||
return this.wrapped.getPageMeta(pageName);
|
||||
}
|
||||
|
||||
async deletePage(pageName: string): Promise<void> {
|
||||
await this.eventHook.dispatchEvent("page:deleted", pageName);
|
||||
return this.wrapped.deletePage(pageName);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,142 @@
|
||||
import { PageMeta } from "../types";
|
||||
import { Plug } from "../../plugos/plug";
|
||||
import { SpacePrimitives } from "./space_primitives";
|
||||
|
||||
export class HttpSpacePrimitives implements SpacePrimitives {
|
||||
pageUrl: string;
|
||||
private plugUrl: string;
|
||||
|
||||
constructor(url: string) {
|
||||
this.pageUrl = url + "/fs";
|
||||
this.plugUrl = url + "/plug";
|
||||
}
|
||||
|
||||
public async fetchPageList(): Promise<{
|
||||
pages: Set<PageMeta>;
|
||||
nowTimestamp: number;
|
||||
}> {
|
||||
let req = await fetch(this.pageUrl, {
|
||||
method: "GET",
|
||||
});
|
||||
|
||||
let result = new Set<PageMeta>();
|
||||
((await req.json()) as any[]).forEach((meta: any) => {
|
||||
const pageName = meta.name;
|
||||
result.add({
|
||||
name: pageName,
|
||||
lastModified: meta.lastModified,
|
||||
});
|
||||
});
|
||||
|
||||
return {
|
||||
pages: result,
|
||||
nowTimestamp: +req.headers.get("Now-Timestamp")!,
|
||||
};
|
||||
}
|
||||
|
||||
async readPage(name: string): Promise<{ text: string; meta: PageMeta }> {
|
||||
let res = await fetch(`${this.pageUrl}/${name}`, {
|
||||
method: "GET",
|
||||
});
|
||||
if (res.headers.get("X-Status") === "404") {
|
||||
throw new Error(`Page not found`);
|
||||
}
|
||||
return {
|
||||
text: await res.text(),
|
||||
meta: this.responseToMeta(name, res),
|
||||
};
|
||||
}
|
||||
|
||||
async writePage(
|
||||
name: string,
|
||||
text: string,
|
||||
selfUpdate?: boolean,
|
||||
lastModified?: number
|
||||
): Promise<PageMeta> {
|
||||
// TODO: lastModified ignored for now
|
||||
let res = await fetch(`${this.pageUrl}/${name}`, {
|
||||
method: "PUT",
|
||||
body: text,
|
||||
headers: lastModified
|
||||
? {
|
||||
"Last-Modified": "" + lastModified,
|
||||
}
|
||||
: undefined,
|
||||
});
|
||||
const newMeta = this.responseToMeta(name, res);
|
||||
return newMeta;
|
||||
}
|
||||
|
||||
async deletePage(name: string): Promise<void> {
|
||||
let req = await fetch(`${this.pageUrl}/${name}`, {
|
||||
method: "DELETE",
|
||||
});
|
||||
if (req.status !== 200) {
|
||||
throw Error(`Failed to delete page: ${req.statusText}`);
|
||||
}
|
||||
}
|
||||
|
||||
async proxySyscall(plug: Plug<any>, name: string, args: any[]): Promise<any> {
|
||||
let req = await fetch(`${this.plugUrl}/${plug.name}/syscall/${name}`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-type": "application/json",
|
||||
},
|
||||
body: JSON.stringify(args),
|
||||
});
|
||||
if (req.status !== 200) {
|
||||
let error = await req.text();
|
||||
throw Error(error);
|
||||
}
|
||||
if (req.headers.get("Content-length") === "0") {
|
||||
return;
|
||||
}
|
||||
return await req.json();
|
||||
}
|
||||
|
||||
async invokeFunction(
|
||||
plug: Plug<any>,
|
||||
env: string,
|
||||
name: string,
|
||||
args: any[]
|
||||
): Promise<any> {
|
||||
// Invoke locally
|
||||
if (!env || env === "client") {
|
||||
return plug.invoke(name, args);
|
||||
}
|
||||
// Or dispatch to server
|
||||
let req = await fetch(`${this.plugUrl}/${plug.name}/function/${name}`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-type": "application/json",
|
||||
},
|
||||
body: JSON.stringify(args),
|
||||
});
|
||||
if (req.status !== 200) {
|
||||
let error = await req.text();
|
||||
throw Error(error);
|
||||
}
|
||||
if (req.headers.get("Content-length") === "0") {
|
||||
return;
|
||||
}
|
||||
return await req.json();
|
||||
}
|
||||
|
||||
async getPageMeta(name: string): Promise<PageMeta> {
|
||||
let res = await fetch(`${this.pageUrl}/${name}`, {
|
||||
method: "OPTIONS",
|
||||
});
|
||||
if (res.headers.get("X-Status") === "404") {
|
||||
throw new Error(`Page not found`);
|
||||
}
|
||||
return this.responseToMeta(name, res);
|
||||
}
|
||||
|
||||
private responseToMeta(name: string, res: Response): PageMeta {
|
||||
const meta = {
|
||||
name,
|
||||
lastModified: +(res.headers.get("Last-Modified") || "0"),
|
||||
};
|
||||
return meta;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
import { SpacePrimitives } from "./space_primitives";
|
||||
import { PageMeta } from "../types";
|
||||
import Dexie, { Table } from "dexie";
|
||||
import { Plug } from "../../plugos/plug";
|
||||
|
||||
type Page = {
|
||||
name: string;
|
||||
text: string;
|
||||
meta: PageMeta;
|
||||
};
|
||||
|
||||
export class IndexedDBSpacePrimitives implements SpacePrimitives {
|
||||
private pageTable: Table<Page, string>;
|
||||
|
||||
constructor(dbName: string, readonly timeSkew: number = 0) {
|
||||
const db = new Dexie(dbName);
|
||||
db.version(1).stores({
|
||||
page: "name",
|
||||
});
|
||||
this.pageTable = db.table("page");
|
||||
}
|
||||
|
||||
async deletePage(name: string): Promise<void> {
|
||||
return this.pageTable.delete(name);
|
||||
}
|
||||
|
||||
async getPageMeta(name: string): Promise<PageMeta> {
|
||||
let entry = await this.pageTable.get(name);
|
||||
if (entry) {
|
||||
return entry.meta;
|
||||
} else {
|
||||
throw Error(`Page not found`);
|
||||
}
|
||||
}
|
||||
|
||||
invokeFunction(
|
||||
plug: Plug<any>,
|
||||
env: string,
|
||||
name: string,
|
||||
args: any[]
|
||||
): Promise<any> {
|
||||
return plug.invoke(name, args);
|
||||
}
|
||||
|
||||
async fetchPageList(): Promise<{
|
||||
pages: Set<PageMeta>;
|
||||
nowTimestamp: number;
|
||||
}> {
|
||||
let allPages = await this.pageTable.toArray();
|
||||
return {
|
||||
pages: new Set(allPages.map((p) => p.meta)),
|
||||
nowTimestamp: Date.now() + this.timeSkew,
|
||||
};
|
||||
}
|
||||
|
||||
proxySyscall(plug: Plug<any>, name: string, args: any[]): Promise<any> {
|
||||
return plug.syscall(name, args);
|
||||
}
|
||||
|
||||
async readPage(name: string): Promise<{ text: string; meta: PageMeta }> {
|
||||
let page = await this.pageTable.get(name);
|
||||
if (page) {
|
||||
return page;
|
||||
} else {
|
||||
throw new Error("Page not found");
|
||||
}
|
||||
}
|
||||
|
||||
async writePage(
|
||||
name: string,
|
||||
text: string,
|
||||
selfUpdate?: boolean,
|
||||
lastModified?: number
|
||||
): Promise<PageMeta> {
|
||||
let meta = {
|
||||
name,
|
||||
lastModified: lastModified ? lastModified : Date.now() + this.timeSkew,
|
||||
};
|
||||
await this.pageTable.put({
|
||||
name,
|
||||
text,
|
||||
meta,
|
||||
});
|
||||
return meta;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,243 @@
|
||||
import { SpacePrimitives } from "./space_primitives";
|
||||
import { safeRun } from "../../webapp/util";
|
||||
import { PageMeta } from "../types";
|
||||
import { EventEmitter } from "../event";
|
||||
import { Plug } from "../../plugos/plug";
|
||||
import { Manifest } from "../manifest";
|
||||
|
||||
const pageWatchInterval = 2000;
|
||||
const trashPrefix = "_trash/";
|
||||
const plugPrefix = "_plug/";
|
||||
|
||||
export type SpaceEvents = {
|
||||
pageCreated: (meta: PageMeta) => void;
|
||||
pageChanged: (meta: PageMeta) => void;
|
||||
pageDeleted: (name: string) => void;
|
||||
pageListUpdated: (pages: Set<PageMeta>) => void;
|
||||
plugLoaded: (plugName: string, plug: Manifest) => void;
|
||||
plugUnloaded: (plugName: string) => void;
|
||||
};
|
||||
|
||||
export class Space extends EventEmitter<SpaceEvents> {
|
||||
pageMetaCache = new Map<string, PageMeta>();
|
||||
watchedPages = new Set<string>();
|
||||
private initialPageListLoad = true;
|
||||
private saving = false;
|
||||
|
||||
constructor(private space: SpacePrimitives, private trashEnabled = true) {
|
||||
super();
|
||||
this.on({
|
||||
pageCreated: async (pageMeta) => {
|
||||
if (pageMeta.name.startsWith(plugPrefix)) {
|
||||
let pageData = await this.readPage(pageMeta.name);
|
||||
this.emit(
|
||||
"plugLoaded",
|
||||
pageMeta.name.substring(plugPrefix.length),
|
||||
JSON.parse(pageData.text)
|
||||
);
|
||||
this.watchPage(pageMeta.name);
|
||||
}
|
||||
},
|
||||
pageChanged: async (pageMeta) => {
|
||||
if (pageMeta.name.startsWith(plugPrefix)) {
|
||||
let pageData = await this.readPage(pageMeta.name);
|
||||
this.emit(
|
||||
"plugLoaded",
|
||||
pageMeta.name.substring(plugPrefix.length),
|
||||
JSON.parse(pageData.text)
|
||||
);
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
public updatePageListAsync() {
|
||||
safeRun(async () => {
|
||||
let newPageList = await this.space.fetchPageList();
|
||||
let deletedPages = new Set<string>(this.pageMetaCache.keys());
|
||||
newPageList.pages.forEach((meta) => {
|
||||
const pageName = meta.name;
|
||||
const oldPageMeta = this.pageMetaCache.get(pageName);
|
||||
const newPageMeta = {
|
||||
name: pageName,
|
||||
lastModified: meta.lastModified,
|
||||
};
|
||||
if (
|
||||
!oldPageMeta &&
|
||||
(pageName.startsWith(plugPrefix) || !this.initialPageListLoad)
|
||||
) {
|
||||
this.emit("pageCreated", newPageMeta);
|
||||
} else if (
|
||||
oldPageMeta &&
|
||||
oldPageMeta.lastModified !== newPageMeta.lastModified
|
||||
) {
|
||||
this.emit("pageChanged", newPageMeta);
|
||||
}
|
||||
// Page found, not deleted
|
||||
deletedPages.delete(pageName);
|
||||
|
||||
// Update in cache
|
||||
this.pageMetaCache.set(pageName, newPageMeta);
|
||||
});
|
||||
|
||||
for (const deletedPage of deletedPages) {
|
||||
this.pageMetaCache.delete(deletedPage);
|
||||
this.emit("pageDeleted", deletedPage);
|
||||
}
|
||||
|
||||
this.emit("pageListUpdated", this.listPages());
|
||||
this.initialPageListLoad = false;
|
||||
});
|
||||
}
|
||||
|
||||
watch() {
|
||||
setInterval(() => {
|
||||
safeRun(async () => {
|
||||
if (this.saving) {
|
||||
return;
|
||||
}
|
||||
for (const pageName of this.watchedPages) {
|
||||
const oldMeta = this.pageMetaCache.get(pageName);
|
||||
if (!oldMeta) {
|
||||
// No longer in cache, meaning probably deleted let's unwatch
|
||||
this.watchedPages.delete(pageName);
|
||||
continue;
|
||||
}
|
||||
// This seems weird, but simply fetching it will compare to local cache and trigger an event if necessary
|
||||
await this.getPageMeta(pageName);
|
||||
}
|
||||
});
|
||||
}, pageWatchInterval);
|
||||
this.updatePageListAsync();
|
||||
}
|
||||
|
||||
async deletePage(name: string, deleteDate?: number): Promise<void> {
|
||||
await this.getPageMeta(name); // Check if page exists, if not throws Error
|
||||
if (this.trashEnabled) {
|
||||
let pageData = await this.readPage(name);
|
||||
// Move to trash
|
||||
await this.writePage(
|
||||
`${trashPrefix}${name}`,
|
||||
pageData.text,
|
||||
false,
|
||||
deleteDate
|
||||
);
|
||||
}
|
||||
await this.space.deletePage(name);
|
||||
|
||||
this.pageMetaCache.delete(name);
|
||||
this.emit("pageDeleted", name);
|
||||
this.emit("pageListUpdated", new Set([...this.pageMetaCache.values()]));
|
||||
}
|
||||
|
||||
async getPageMeta(name: string): Promise<PageMeta> {
|
||||
let oldMeta = this.pageMetaCache.get(name);
|
||||
let newMeta = await this.space.getPageMeta(name);
|
||||
if (oldMeta) {
|
||||
if (oldMeta.lastModified !== newMeta.lastModified) {
|
||||
// Changed on disk, trigger event
|
||||
this.emit("pageChanged", newMeta);
|
||||
}
|
||||
}
|
||||
return this.metaCacher(name, newMeta);
|
||||
}
|
||||
|
||||
invokeFunction(
|
||||
plug: Plug<any>,
|
||||
env: string,
|
||||
name: string,
|
||||
args: any[]
|
||||
): Promise<any> {
|
||||
return this.space.invokeFunction(plug, env, name, args);
|
||||
}
|
||||
|
||||
listPages(): Set<PageMeta> {
|
||||
return new Set(
|
||||
[...this.pageMetaCache.values()].filter(
|
||||
(pageMeta) =>
|
||||
!pageMeta.name.startsWith(trashPrefix) &&
|
||||
!pageMeta.name.startsWith(plugPrefix)
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
listTrash(): Set<PageMeta> {
|
||||
return new Set(
|
||||
[...this.pageMetaCache.values()]
|
||||
.filter(
|
||||
(pageMeta) =>
|
||||
pageMeta.name.startsWith(trashPrefix) &&
|
||||
!pageMeta.name.startsWith(plugPrefix)
|
||||
)
|
||||
.map((pageMeta) => ({
|
||||
...pageMeta,
|
||||
name: pageMeta.name.substring(trashPrefix.length),
|
||||
}))
|
||||
);
|
||||
}
|
||||
|
||||
listPlugs(): Set<PageMeta> {
|
||||
return new Set(
|
||||
[...this.pageMetaCache.values()].filter((pageMeta) =>
|
||||
pageMeta.name.startsWith(plugPrefix)
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
proxySyscall(plug: Plug<any>, name: string, args: any[]): Promise<any> {
|
||||
return this.space.proxySyscall(plug, name, args);
|
||||
}
|
||||
|
||||
async readPage(name: string): Promise<{ text: string; meta: PageMeta }> {
|
||||
let pageData = await this.space.readPage(name);
|
||||
let previousMeta = this.pageMetaCache.get(name);
|
||||
if (previousMeta) {
|
||||
if (previousMeta.lastModified !== pageData.meta.lastModified) {
|
||||
// Page changed since last cached metadata, trigger event
|
||||
this.emit("pageChanged", pageData.meta);
|
||||
}
|
||||
}
|
||||
this.pageMetaCache.set(name, pageData.meta);
|
||||
return pageData;
|
||||
}
|
||||
|
||||
watchPage(pageName: string) {
|
||||
this.watchedPages.add(pageName);
|
||||
}
|
||||
|
||||
unwatchPage(pageName: string) {
|
||||
this.watchedPages.delete(pageName);
|
||||
}
|
||||
|
||||
async writePage(
|
||||
name: string,
|
||||
text: string,
|
||||
selfUpdate?: boolean,
|
||||
lastModified?: number
|
||||
): Promise<PageMeta> {
|
||||
try {
|
||||
this.saving = true;
|
||||
let pageMeta = await this.space.writePage(
|
||||
name,
|
||||
text,
|
||||
selfUpdate,
|
||||
lastModified
|
||||
);
|
||||
if (!selfUpdate) {
|
||||
this.emit("pageChanged", pageMeta);
|
||||
}
|
||||
return this.metaCacher(name, pageMeta);
|
||||
} finally {
|
||||
this.saving = false;
|
||||
}
|
||||
}
|
||||
|
||||
fetchPageList(): Promise<{ pages: Set<PageMeta>; nowTimestamp: number }> {
|
||||
return this.space.fetchPageList();
|
||||
}
|
||||
|
||||
private metaCacher(name: string, pageMeta: PageMeta): PageMeta {
|
||||
this.pageMetaCache.set(name, pageMeta);
|
||||
return pageMeta;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
import { Plug } from "../../plugos/plug";
|
||||
import { PageMeta } from "../types";
|
||||
|
||||
export interface SpacePrimitives {
|
||||
// Pages
|
||||
fetchPageList(): Promise<{ pages: Set<PageMeta>; nowTimestamp: number }>;
|
||||
readPage(name: string): Promise<{ text: string; meta: PageMeta }>;
|
||||
getPageMeta(name: string): Promise<PageMeta>;
|
||||
writePage(
|
||||
name: string,
|
||||
text: string,
|
||||
selfUpdate?: boolean,
|
||||
lastModified?: number
|
||||
): Promise<PageMeta>;
|
||||
deletePage(name: string): Promise<void>;
|
||||
|
||||
// Plugs
|
||||
proxySyscall(plug: Plug<any>, name: string, args: any[]): Promise<any>;
|
||||
invokeFunction(
|
||||
plug: Plug<any>,
|
||||
env: string,
|
||||
name: string,
|
||||
args: any[]
|
||||
): Promise<any>;
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
import { expect, test } from "@jest/globals";
|
||||
import { IndexedDBSpacePrimitives } from "./indexeddb_space_primitives";
|
||||
import { SpaceSync } from "./sync";
|
||||
import { PageMeta } from "../types";
|
||||
import { Space } from "./space";
|
||||
|
||||
// For testing in node.js
|
||||
require("fake-indexeddb/auto");
|
||||
|
||||
test("Test store", async () => {
|
||||
let primary = new Space(new IndexedDBSpacePrimitives("primary"), true);
|
||||
let secondary = new Space(
|
||||
new IndexedDBSpacePrimitives("secondary", -5000),
|
||||
true
|
||||
);
|
||||
let sync = new SpaceSync(primary, secondary, 0, 0, "_trash/");
|
||||
|
||||
async function conflictResolver(pageMeta1: PageMeta, pageMeta2: PageMeta) {}
|
||||
|
||||
// Write one page to primary
|
||||
await primary.writePage("start", "Hello");
|
||||
expect((await secondary.listPages()).size).toBe(0);
|
||||
await syncPages(conflictResolver);
|
||||
expect((await secondary.listPages()).size).toBe(1);
|
||||
expect((await secondary.readPage("start")).text).toBe("Hello");
|
||||
|
||||
// Should be a no-op
|
||||
expect(await syncPages()).toBe(0);
|
||||
|
||||
// Now let's make a change on the secondary
|
||||
await secondary.writePage("start", "Hello!!");
|
||||
await secondary.writePage("test", "Test page");
|
||||
|
||||
// And sync it
|
||||
await syncPages();
|
||||
|
||||
expect(primary.listPages().size).toBe(2);
|
||||
expect(secondary.listPages().size).toBe(2);
|
||||
|
||||
expect((await primary.readPage("start")).text).toBe("Hello!!");
|
||||
|
||||
// Let's make some random edits on both ends
|
||||
await primary.writePage("start", "1");
|
||||
await primary.writePage("start2", "2");
|
||||
await secondary.writePage("start3", "3");
|
||||
await secondary.writePage("start4", "4");
|
||||
await syncPages();
|
||||
|
||||
expect((await primary.listPages()).size).toBe(5);
|
||||
expect((await secondary.listPages()).size).toBe(5);
|
||||
|
||||
expect(await syncPages()).toBe(0);
|
||||
|
||||
console.log("Deleting pages");
|
||||
// Delete some pages
|
||||
await primary.deletePage("start");
|
||||
await primary.deletePage("start3");
|
||||
|
||||
console.log("Pages", await primary.listPages());
|
||||
console.log("Trash", await primary.listTrash());
|
||||
|
||||
await syncPages();
|
||||
|
||||
expect((await primary.listPages()).size).toBe(3);
|
||||
expect((await secondary.listPages()).size).toBe(3);
|
||||
|
||||
// No-op
|
||||
expect(await syncPages()).toBe(0);
|
||||
|
||||
await secondary.deletePage("start4");
|
||||
await primary.deletePage("start2");
|
||||
|
||||
await syncPages();
|
||||
|
||||
// Just "test" left
|
||||
expect((await primary.listPages()).size).toBe(1);
|
||||
expect((await secondary.listPages()).size).toBe(1);
|
||||
|
||||
// No-op
|
||||
expect(await syncPages()).toBe(0);
|
||||
|
||||
await secondary.writePage("start", "I'm back");
|
||||
|
||||
await syncPages();
|
||||
|
||||
expect((await primary.readPage("start")).text).toBe("I'm back");
|
||||
|
||||
// Cause a conflict
|
||||
await primary.writePage("start", "Hello 1");
|
||||
await secondary.writePage("start", "Hello 2");
|
||||
|
||||
await syncPages(SpaceSync.primaryConflictResolver(primary, secondary));
|
||||
|
||||
// Sync conflicting copy back
|
||||
await syncPages();
|
||||
|
||||
// Verify that primary won
|
||||
expect((await primary.readPage("start")).text).toBe("Hello 1");
|
||||
expect((await secondary.readPage("start")).text).toBe("Hello 1");
|
||||
|
||||
// test + start + start.conflicting copy
|
||||
expect((await primary.listPages()).size).toBe(3);
|
||||
expect((await secondary.listPages()).size).toBe(3);
|
||||
|
||||
async function syncPages(
|
||||
conflictResolver?: (
|
||||
pageMeta1: PageMeta,
|
||||
pageMeta2: PageMeta
|
||||
) => Promise<void>
|
||||
): Promise<number> {
|
||||
// Awesome practice: adding sleeps to fix issues!
|
||||
await sleep(2);
|
||||
let n = await sync.syncPages(conflictResolver);
|
||||
await sleep(2);
|
||||
return n;
|
||||
}
|
||||
});
|
||||
|
||||
function sleep(ms: number = 5): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
setTimeout(resolve, ms);
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,208 @@
|
||||
import { Space } from "./space";
|
||||
import { PageMeta } from "../types";
|
||||
import { SpacePrimitives } from "./space_primitives";
|
||||
|
||||
export class SpaceSync {
|
||||
constructor(
|
||||
private primary: Space,
|
||||
private secondary: Space,
|
||||
public primaryLastSync: number,
|
||||
public secondaryLastSync: number,
|
||||
private trashPrefix: string
|
||||
) {}
|
||||
|
||||
// Strategy: Primary wins
|
||||
public static primaryConflictResolver(
|
||||
primary: Space,
|
||||
secondary: Space
|
||||
): (pageMeta1: PageMeta, pageMeta2: PageMeta) => Promise<void> {
|
||||
return async (pageMeta1, pageMeta2) => {
|
||||
const pageName = pageMeta1.name;
|
||||
const revisionPageName = `${pageName}.conflicted.${pageMeta2.lastModified}`;
|
||||
// Copy secondary to conflict copy
|
||||
let oldPageData = await secondary.readPage(pageName);
|
||||
await secondary.writePage(revisionPageName, oldPageData.text);
|
||||
|
||||
// Write replacement on top
|
||||
let newPageData = await primary.readPage(pageName);
|
||||
await secondary.writePage(
|
||||
pageName,
|
||||
newPageData.text,
|
||||
true,
|
||||
newPageData.meta.lastModified
|
||||
);
|
||||
};
|
||||
}
|
||||
|
||||
async syncablePages(
|
||||
space: Space
|
||||
): Promise<{ pages: PageMeta[]; nowTimestamp: number }> {
|
||||
let fetchResult = await space.fetchPageList();
|
||||
return {
|
||||
pages: [...fetchResult.pages].filter(
|
||||
(pageMeta) => !pageMeta.name.startsWith(this.trashPrefix)
|
||||
),
|
||||
nowTimestamp: fetchResult.nowTimestamp,
|
||||
};
|
||||
}
|
||||
|
||||
async trashPages(space: SpacePrimitives): Promise<PageMeta[]> {
|
||||
return [...(await space.fetchPageList()).pages]
|
||||
.filter((pageMeta) => pageMeta.name.startsWith(this.trashPrefix))
|
||||
.map((pageMeta) => ({
|
||||
...pageMeta,
|
||||
name: pageMeta.name.substring(this.trashPrefix.length),
|
||||
}));
|
||||
}
|
||||
|
||||
async syncPages(
|
||||
conflictResolver?: (
|
||||
pageMeta1: PageMeta,
|
||||
pageMeta2: PageMeta
|
||||
) => Promise<void>
|
||||
): Promise<number> {
|
||||
let syncOps = 0;
|
||||
|
||||
let { pages: primaryAllPagesSet, nowTimestamp: primarySyncTimestamp } =
|
||||
await this.syncablePages(this.primary);
|
||||
let allPagesPrimary = new Map(primaryAllPagesSet.map((p) => [p.name, p]));
|
||||
let { pages: secondaryAllPagesSet, nowTimestamp: secondarySyncTimestamp } =
|
||||
await this.syncablePages(this.secondary);
|
||||
let allPagesSecondary = new Map(
|
||||
secondaryAllPagesSet.map((p) => [p.name, p])
|
||||
);
|
||||
|
||||
let allTrashPrimary = new Map(
|
||||
(await this.trashPages(this.primary))
|
||||
// Filter out old trash
|
||||
.filter((p) => p.lastModified > this.primaryLastSync)
|
||||
.map((p) => [p.name, p])
|
||||
);
|
||||
let allTrashSecondary = new Map(
|
||||
(await this.trashPages(this.secondary))
|
||||
// Filter out old trash
|
||||
.filter((p) => p.lastModified > this.secondaryLastSync)
|
||||
.map((p) => [p.name, p])
|
||||
);
|
||||
|
||||
// Iterate over all pages on the primary first
|
||||
for (let [name, pageMetaPrimary] of allPagesPrimary.entries()) {
|
||||
let pageMetaSecondary = allPagesSecondary.get(pageMetaPrimary.name);
|
||||
if (!pageMetaSecondary) {
|
||||
// New page on primary
|
||||
// Let's check it's not on the deleted list
|
||||
if (allTrashSecondary.has(name)) {
|
||||
// Explicitly deleted, let's skip
|
||||
continue;
|
||||
}
|
||||
|
||||
// Push from primary to secondary
|
||||
console.log("New page on primary", name, "syncing to secondary");
|
||||
let pageData = await this.primary.readPage(name);
|
||||
await this.secondary.writePage(
|
||||
name,
|
||||
pageData.text,
|
||||
true,
|
||||
secondarySyncTimestamp // The reason for this is to not include it in the next sync cycle, we cannot blindly use the lastModified date due to time skew
|
||||
);
|
||||
syncOps++;
|
||||
} else {
|
||||
// Existing page
|
||||
if (pageMetaPrimary.lastModified > this.primaryLastSync) {
|
||||
// Primary updated since last sync
|
||||
if (pageMetaSecondary.lastModified > this.secondaryLastSync) {
|
||||
// Secondary also updated! CONFLICT
|
||||
if (conflictResolver) {
|
||||
await conflictResolver(pageMetaPrimary, pageMetaSecondary);
|
||||
} else {
|
||||
throw Error(
|
||||
`Sync conflict for ${name} with no conflict resolver specified`
|
||||
);
|
||||
}
|
||||
} else {
|
||||
// Ok, not changed on secondary, push it secondary
|
||||
console.log(
|
||||
"Changed page on primary",
|
||||
name,
|
||||
"syncing to secondary"
|
||||
);
|
||||
let pageData = await this.primary.readPage(name);
|
||||
await this.secondary.writePage(
|
||||
name,
|
||||
pageData.text,
|
||||
false,
|
||||
secondarySyncTimestamp
|
||||
);
|
||||
syncOps++;
|
||||
}
|
||||
} else if (pageMetaSecondary.lastModified > this.secondaryLastSync) {
|
||||
// Secondary updated, but not primary (checked above)
|
||||
// Push from secondary to primary
|
||||
console.log("Changed page on secondary", name, "syncing to primary");
|
||||
let pageData = await this.secondary.readPage(name);
|
||||
await this.primary.writePage(
|
||||
name,
|
||||
pageData.text,
|
||||
false,
|
||||
primarySyncTimestamp
|
||||
);
|
||||
syncOps++;
|
||||
} else {
|
||||
// Neither updated, no-op
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Now do a simplified version in reverse, only detecting new pages
|
||||
for (let [name, pageMetaSecondary] of allPagesSecondary.entries()) {
|
||||
if (!allPagesPrimary.has(pageMetaSecondary.name)) {
|
||||
// New page on secondary
|
||||
// Let's check it's not on the deleted list
|
||||
if (allTrashPrimary.has(name)) {
|
||||
// Explicitly deleted, let's skip
|
||||
continue;
|
||||
}
|
||||
// Push from secondary to primary
|
||||
console.log("New page on secondary", name, "pushing to primary");
|
||||
let pageData = await this.secondary.readPage(name);
|
||||
await this.primary.writePage(
|
||||
name,
|
||||
pageData.text,
|
||||
false,
|
||||
primarySyncTimestamp
|
||||
);
|
||||
syncOps++;
|
||||
}
|
||||
}
|
||||
|
||||
// And finally, let's trash some pages
|
||||
for (let pageToDelete of allTrashPrimary.values()) {
|
||||
console.log("Deleting", pageToDelete.name, "on secondary");
|
||||
try {
|
||||
await this.secondary.deletePage(
|
||||
pageToDelete.name,
|
||||
secondarySyncTimestamp
|
||||
);
|
||||
syncOps++;
|
||||
} catch (e: any) {
|
||||
console.log("Page already gone", e.message);
|
||||
}
|
||||
}
|
||||
|
||||
for (let pageToDelete of allTrashSecondary.values()) {
|
||||
console.log("Deleting", pageToDelete.name, "on primary");
|
||||
try {
|
||||
await this.primary.deletePage(pageToDelete.name, primarySyncTimestamp);
|
||||
syncOps++;
|
||||
} catch (e: any) {
|
||||
console.log("Page already gone", e.message);
|
||||
}
|
||||
}
|
||||
|
||||
// Setting last sync time to the timestamps we got back when fetching the page lists on each end
|
||||
this.primaryLastSync = primarySyncTimestamp;
|
||||
this.secondaryLastSync = secondarySyncTimestamp;
|
||||
|
||||
return syncOps;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user