Refactor and renaming

packages/common/spaces/constants.ts

@@ -0,0 +1,2 @@
+export const trashPrefix = "_trash/";
+export const plugPrefix = "_plug/";

packages/common/spaces/disk_space_primitives.ts

@@ -0,0 +1,150 @@
+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/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);
+  }
+}

packages/common/spaces/evented_space_primitives.ts

@@ -0,0 +1,68 @@
+import { SpacePrimitives } from "./space_primitives";
+import { EventHook } from "@plugos/plugos/hooks/event";
+import { PageMeta } from "../types";
+import { Plug } from "@plugos/plugos/plug";
+import { trashPrefix } from "./constants";
+
+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
+    if (!pageName.startsWith(trashPrefix)) {
+      this.eventHook
+        .dispatchEvent("page:saved", pageName)
+        .then(() => {
+          return this.eventHook.dispatchEvent("page:index_text", {
+            name: pageName,
+            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);
+  }
+}

packages/common/spaces/http_space_primitives.ts

@@ -0,0 +1,142 @@
+import { PageMeta } from "../types";
+import { Plug } from "@plugos/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;
+  }
+}

packages/common/spaces/indexeddb_space_primitives.ts

@@ -0,0 +1,86 @@
+import { SpacePrimitives } from "./space_primitives";
+import { PageMeta } from "../types";
+import Dexie, { Table } from "dexie";
+import { Plug } from "@plugos/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;
+  }
+}

packages/common/spaces/space.ts

@@ -0,0 +1,244 @@
+import { SpacePrimitives } from "./space_primitives";
+import { PageMeta } from "../types";
+import { EventEmitter } from "@plugos/plugos/event";
+import { Plug } from "@plugos/plugos/plug";
+import { Manifest } from "../manifest";
+import { plugPrefix, trashPrefix } from "./constants";
+import { safeRun } from "../util";
+
+const pageWatchInterval = 2000;
+
+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.trashEnabled ||
+            (this.trashEnabled && !pageName.startsWith(trashPrefix)))
+        ) {
+          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,
+        true,
+        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;
+  }
+}

packages/common/spaces/space_primitives.ts

@@ -0,0 +1,25 @@
+import { Plug } from "@plugos/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>;
+}

packages/common/spaces/sync.test.ts

@@ -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);
+  });
+}

packages/common/spaces/sync.ts

@@ -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;
+  }
+}