feat(doc-storage): init

.eslintrc.js

@@ -51,7 +51,7 @@ const allPackages = [
   'packages/common/debug',
   'packages/common/env',
   'packages/common/infra',
-  'packages/common/theme',
+  'packages/common/workspace-storage',
   'tools/cli',
 ];
 

packages/common/workspace-storage/package.json

@@ -0,0 +1,17 @@
+{
+  "name": "@affine/workspace-storage",
+  "type": "module",
+  "version": "0.15.0",
+  "private": true,
+  "sideEffects": false,
+  "exports": {
+    ".": "./src/index.ts"
+  },
+  "dependencies": {
+    "lodash-es": "^4.17.21",
+    "yjs": "patch:yjs@npm%3A13.6.18#~/.yarn/patches/yjs-npm-13.6.18-ad0d5f7c43.patch"
+  },
+  "devDependencies": {
+    "@types/lodash-es": "^4.17.12"
+  }
+}

packages/common/workspace-storage/src/blob.ts

@@ -0,0 +1,16 @@
+import { Connection } from './connection';
+
+export interface WorkspaceBlobStorageOptions {}
+
+export interface Blob {
+  key: string;
+  bin: Uint8Array;
+  mimeType: string;
+}
+
+export abstract class WorkspaceBlobStorageAdapter extends Connection {
+  abstract getBlob(workspaceId: string, key: string): Promise<Blob | null>;
+  abstract setBlob(workspaceId: string, blob: Blob): Promise<string>;
+  abstract deleteBlob(workspaceId: string, key: string): Promise<boolean>;
+  abstract listBlobs(workspaceId: string): Promise<Blob>;
+}

packages/common/workspace-storage/src/connection.ts

@@ -0,0 +1,11 @@
+export class Connection {
+  connected: boolean = false;
+  connect(): Promise<void> {
+    this.connected = true;
+    return Promise.resolve();
+  }
+  disconnect(): Promise<void> {
+    this.connected = false;
+    return Promise.resolve();
+  }
+}

packages/common/workspace-storage/src/doc.ts

@@ -0,0 +1,145 @@
+import { mergeUpdates } from 'yjs';
+
+import { Connection } from './connection';
+import { type Lock, SingletonLocker } from './lock';
+
+export interface DocRecord {
+  workspaceId: string;
+  docId: string;
+  bin: Uint8Array;
+  version: number;
+}
+
+export interface DocUpdate {
+  bin: Uint8Array;
+  version: number;
+}
+
+export interface WorkspaceDocStorageOptions {
+  mergeUpdates?: (updates: Uint8Array[]) => Uint8Array;
+}
+
+export abstract class WorkspaceDocStorageAdapter extends Connection {
+  private readonly locker = new SingletonLocker();
+
+  constructor(protected readonly options: WorkspaceDocStorageOptions) {
+    super();
+  }
+
+  // open apis
+  async getDoc(workspaceId: string, docId: string): Promise<DocRecord | null> {
+    using _lock = await this.lockDocForUpdate(workspaceId, docId);
+
+    let snapshot = await this.getDocSnapshot(workspaceId, docId);
+    const updates = await this.getDocPendingUpdates(workspaceId, docId);
+
+    if (updates.length) {
+      if (snapshot) {
+        updates.unshift(snapshot);
+      }
+
+      snapshot = this.squash(updates);
+
+      await this.setDocSnapshot(snapshot);
+      await this.markUpdatesMerged(workspaceId, docId, snapshot.version);
+    }
+
+    return snapshot;
+  }
+
+  abstract pushDocUpdates(
+    workspaceId: string,
+    docId: string,
+    updates: DocUpdate[]
+  ): Promise<number>;
+  abstract deleteDoc(workspaceId: string, docId: string): Promise<boolean>;
+  abstract deleteWorkspace(workspaceId: string): Promise<void>;
+  async rollbackDoc(
+    workspaceId: string,
+    docId: string,
+    version: number
+  ): Promise<void> {
+    // using _lock = await this.lockDocForUpdate(workspaceId, docId);
+    const toSnapshot = await this.getHistoryDocSnapshot(
+      workspaceId,
+      docId,
+      version
+    );
+    if (!toSnapshot) {
+      throw new Error('Can not find the version to rollback to.');
+    }
+
+    const fromSnapshot = await this.getDocSnapshot(workspaceId, docId);
+
+    if (!fromSnapshot) {
+      throw new Error('Can not find the current version of the doc.');
+    }
+
+    // recover
+    await this.setDocSnapshot({
+      ...toSnapshot,
+      // deal with the version, maybe we need to lock snapshot for writing before starting rollback
+      // version: fromSnapshot.version + 1,
+      // bin: newBin
+    });
+  }
+
+  abstract getDocVersions(
+    workspaceId: string
+  ): Promise<Record<string, number> | null>;
+  abstract listDocHistories(
+    workspaceId: string,
+    docId: string,
+    query: { skip?: number; limit?: number }
+  ): Promise<DocRecord[]>;
+  abstract getHistoryDocSnapshot(
+    workspaceId: string,
+    docId: string,
+    version: number
+  ): Promise<DocRecord | null>;
+
+  // api for internal usage
+  protected abstract getDocSnapshot(
+    workspaceId: string,
+    docId: string
+  ): Promise<DocRecord | null>;
+  protected abstract setDocSnapshot(snapshot: DocRecord): Promise<void>;
+  abstract getDocPendingUpdates(
+    workspaceId: string,
+    docId: string
+  ): Promise<DocRecord[]>;
+  abstract markUpdatesMerged(
+    workspaceId: string,
+    docId: string,
+    version: number
+  ): Promise<number>;
+
+  protected squash(updates: DocRecord[]): DocRecord {
+    const merge = this.options?.mergeUpdates ?? mergeUpdates;
+    const lastUpdate = updates.at(-1);
+    if (!lastUpdate) {
+      throw new Error('No updates to be squashed.');
+    }
+
+    // fast return
+    if (updates.length === 1) {
+      return lastUpdate;
+    }
+
+    const finalUpdate = merge(updates.map(u => u.bin));
+
+    return {
+      version: lastUpdate.version,
+      workspaceId: lastUpdate.workspaceId,
+      docId: lastUpdate.docId,
+      bin: finalUpdate,
+    };
+  }
+
+  protected async lockDocForUpdate(
+    workspaceId: string,
+    docId: string
+  ): Promise<Lock> {
+    return this.locker.lock(`workspace:${workspaceId}:update`, docId);
+  }
+}

packages/common/workspace-storage/src/index.ts

@@ -0,0 +1,22 @@
+import type { WorkspaceBlobStorageAdapter } from './blob';
+import { Connection } from './connection';
+import type { WorkspaceDocStorageAdapter } from './doc';
+
+export class WorkspaceSyncProvider extends Connection {
+  constructor(
+    public readonly doc: WorkspaceDocStorageAdapter,
+    public readonly blob: WorkspaceBlobStorageAdapter
+  ) {
+    super();
+  }
+
+  override async connect() {
+    await this.doc.connect();
+    await this.blob.connect();
+  }
+
+  override async disconnect() {
+    await this.doc.disconnect();
+    await this.blob.disconnect();
+  }
+}

packages/common/workspace-storage/src/lock.ts

@@ -0,0 +1,41 @@
+export interface Locker {
+  lock(domain: string, resource: string): Promise<Lock>;
+}
+
+export class SingletonLocker implements Locker {
+  lockedResource = new Map<string, Lock>();
+  constructor() {}
+
+  async lock(domain: string, resource: string) {
+    let lock = this.lockedResource.get(`${domain}:${resource}`);
+
+    if (!lock) {
+      lock = new Lock();
+    }
+
+    await lock.acquire();
+
+    return lock;
+  }
+}
+
+export class Lock {
+  private inner: Promise<void> = Promise.resolve();
+  private release: () => void = () => {};
+
+  async acquire() {
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    let release: () => void = null!;
+    const nextLock = new Promise<void>(resolve => {
+      release = resolve;
+    });
+
+    await this.inner;
+    this.inner = nextLock;
+    this.release = release;
+  }
+
+  [Symbol.dispose]() {
+    this.release();
+  }
+}

packages/common/workspace-storage/tsconfig.json

@@ -0,0 +1,14 @@
+{
+  "extends": "../../../tsconfig.json",
+  "include": ["./src"],
+  "compilerOptions": {
+    "composite": true,
+    "noEmit": false,
+    "outDir": "lib"
+  },
+  "references": [
+    {
+      "path": "./tsconfig.node.json"
+    }
+  ]
+}

packages/common/workspace-storage/tsconfig.node.json

@@ -0,0 +1,11 @@
+{
+  "extends": "../../../tsconfig.json",
+  "compilerOptions": {
+    "composite": true,
+    "module": "ESNext",
+    "moduleResolution": "Node",
+    "allowSyntheticDefaultImports": true,
+    "outDir": "lib",
+    "noEmit": false
+  }
+}

tsconfig.json

@@ -120,6 +120,9 @@
     {
       "path": "./packages/common/infra"
     },
+    {
+      "path": "./packages/common/workspace-storage"
+    },
     // Tools
     {
       "path": "./tools/cli"

yarn.lock

@@ -874,6 +874,16 @@ __metadata:
   languageName: unknown
   linkType: soft
 
+"@affine/workspace-storage@workspace:packages/common/workspace-storage":
+  version: 0.0.0-use.local
+  resolution: "@affine/workspace-storage@workspace:packages/common/workspace-storage"
+  dependencies:
+    "@types/lodash-es": "npm:^4.17.12"
+    lodash-es: "npm:^4.17.21"
+    yjs: "patch:yjs@npm%3A13.6.18#~/.yarn/patches/yjs-npm-13.6.18-ad0d5f7c43.patch"
+  languageName: unknown
+  linkType: soft
+
 "@alloc/quick-lru@npm:^5.2.0":
   version: 5.2.0
   resolution: "@alloc/quick-lru@npm:5.2.0"