feat(core): support syncing workspaces and blobs in the background (#4057)

apps/core/src/adapters/workspace.ts

@@ -9,8 +9,9 @@ import {
   ReleaseType,
   WorkspaceFlavour,
 } from '@affine/env/workspace';
+import { CRUD as CloudCRUD } from '@affine/workspace/affine/crud';
+import { startSync, stopSync } from '@affine/workspace/affine/sync';
 
-import { CRUD as CloudCRUD } from './cloud/crud';
 import { UI as CloudUI } from './cloud/ui';
 import { LocalAdapter } from './local';
 import { UI as PublicCloudUI } from './public-cloud/ui';
@@ -40,6 +41,8 @@ export const WorkspaceAdapters = {
           return false;
         }
       },
+      'service:start': startSync,
+      'service:stop': stopSync,
     } as Partial<AppEvents>,
     CRUD: CloudCRUD,
     UI: CloudUI,

packages/env/src/workspace.ts

@@ -190,6 +190,8 @@ export interface AppEvents {
   'app:init': () => string[];
   // event if you have access to workspace adapter
   'app:access': () => Promise<boolean>;
+  'service:start': () => void;
+  'service:stop': () => void;
 }
 
 export interface WorkspaceAdapter<Flavour extends WorkspaceFlavour> {

packages/workspace/package.json

@@ -7,7 +7,9 @@
     "./type": "./src/type.ts",
     "./migration": "./src/migration/index.ts",
     "./local/crud": "./src/local/crud.ts",
+    "./affine/crud": "./src/affine/crud.ts",
     "./affine/gql": "./src/affine/gql.ts",
+    "./affine/sync": "./src/affine/sync.ts",
     "./providers": "./src/providers/index.ts"
   },
   "peerDependencies": {

packages/workspace/src/affine/crud.ts

@@ -9,8 +9,6 @@ import {
   getWorkspaceQuery,
   getWorkspacesQuery,
 } from '@affine/graphql';
-import { fetcher } from '@affine/workspace/affine/gql';
-import { getOrCreateWorkspace } from '@affine/workspace/manager';
 import { createIndexeddbStorage, Workspace } from '@blocksuite/store';
 import { migrateLocalBlobStorage } from '@toeverything/infra/blocksuite';
 import {
@@ -20,6 +18,9 @@ import {
 import { getSession } from 'next-auth/react';
 import { proxy } from 'valtio/vanilla';
 
+import { getOrCreateWorkspace } from '../manager';
+import { fetcher } from './gql';
+
 const Y = Workspace.Y;
 
 async function deleteLocalBlobStorage(id: string) {

packages/workspace/src/affine/sync.ts

@@ -0,0 +1,93 @@
+import { createIndexeddbStorage } from '@blocksuite/store';
+import { pushBinary } from '@toeverything/y-indexeddb';
+import type { Doc } from 'yjs';
+import { applyUpdate } from 'yjs';
+
+import { createCloudBlobStorage } from '../blob/cloud-blob-storage';
+import { downloadBinaryFromCloud } from '../providers/cloud';
+import { CRUD } from './crud';
+
+let abortController: AbortController | undefined;
+
+const downloadRecursive = async (
+  rootGuid: string,
+  doc: Doc,
+  signal: AbortSignal
+): Promise<void> => {
+  if (signal.aborted) {
+    return;
+  }
+  const binary = await downloadBinaryFromCloud(rootGuid, doc.guid);
+  if (typeof binary !== 'boolean') {
+    const update = new Uint8Array(binary);
+    if (rootGuid === doc.guid) {
+      // only apply the root doc
+      applyUpdate(doc, update, 'affine-cloud-service');
+    } else {
+      await pushBinary(doc.guid, update);
+    }
+  }
+  return Promise.all(
+    [...doc.subdocs.values()].map(subdoc =>
+      downloadRecursive(rootGuid, subdoc, signal)
+    )
+  ).then();
+};
+
+export async function startSync() {
+  abortController = new AbortController();
+  const signal = abortController.signal;
+  const workspaces = await CRUD.list();
+  const downloadCloudPromises = workspaces.map(workspace =>
+    downloadRecursive(workspace.id, workspace.blockSuiteWorkspace.doc, signal)
+  );
+  const syncBlobPromises = workspaces.map(async workspace => {
+    const cloudBlobStorage = createCloudBlobStorage(workspace.id);
+    const indexeddbBlobStorage = createIndexeddbStorage(workspace.id);
+    return Promise.all([
+      cloudBlobStorage.crud.list(),
+      indexeddbBlobStorage.crud.list(),
+    ]).then(([cloudKeys, indexeddbKeys]) => {
+      if (signal.aborted) {
+        return;
+      }
+      const cloudKeysSet = new Set(cloudKeys);
+      const indexeddbKeysSet = new Set(indexeddbKeys);
+      // missing in indexeddb
+      const missingLocalKeys = cloudKeys.filter(
+        key => !indexeddbKeysSet.has(key)
+      );
+      // missing in cloud
+      const missingCloudKeys = indexeddbKeys.filter(
+        key => !cloudKeysSet.has(key)
+      );
+      return Promise.all([
+        ...missingLocalKeys.map(key =>
+          cloudBlobStorage.crud.get(key).then(async value => {
+            if (signal.aborted) {
+              return;
+            }
+            if (value) {
+              await indexeddbBlobStorage.crud.set(key, value);
+            }
+          })
+        ),
+        ...missingCloudKeys.map(key =>
+          indexeddbBlobStorage.crud.get(key).then(async value => {
+            if (signal.aborted) {
+              return;
+            }
+            if (value) {
+              await cloudBlobStorage.crud.set(key, value);
+            }
+          })
+        ),
+      ]);
+    });
+  });
+  await Promise.all([...downloadCloudPromises, ...syncBlobPromises]);
+}
+
+export async function stopSync() {
+  abortController?.abort();
+}

packages/workspace/src/atom.ts

@@ -143,6 +143,7 @@ const fetchMetadata: FetchMetadata = async (get, { signal }) => {
         removed.forEach(meta => {
           metadata.splice(metadata.indexOf(meta), 1);
         });
+        Adapter.Events['service:stop']?.();
         continue;
       }
       try {
@@ -178,6 +179,7 @@ const fetchMetadata: FetchMetadata = async (get, { signal }) => {
       } catch (e) {
         console.error('list data error:', e);
       }
+      Adapter.Events['service:start']?.();
     }
   }
   const metadataMap = new Map(metadata.map(x => [x.id, x]));

packages/workspace/src/providers/cloud/index.ts

@@ -8,23 +8,30 @@ const Y = Workspace.Y;
 
 const logger = new DebugLogger('affine:cloud');
 
+const hashMap = new Map<string, ArrayBuffer>();
+
 export async function downloadBinaryFromCloud(
   rootGuid: string,
   pageGuid: string
-) {
+): Promise<boolean | ArrayBuffer> {
+  if (hashMap.has(`${rootGuid}/${pageGuid}`)) {
+    return true;
+  }
   const response = await fetchWithReport(
     runtimeConfig.serverUrlPrefix +
       `/api/workspaces/${rootGuid}/docs/${pageGuid}`
   );
   if (response.ok) {
-    return response.arrayBuffer();
+    const arrayBuffer = await response.arrayBuffer();
+    hashMap.set(`${rootGuid}/${pageGuid}`, arrayBuffer);
+    return arrayBuffer;
   }
   return false;
 }
 
 async function downloadBinary(rootGuid: string, doc: Doc) {
   const buffer = await downloadBinaryFromCloud(rootGuid, doc.guid);
-  if (buffer) {
+  if (typeof buffer !== 'boolean') {
     Y.applyUpdate(doc, new Uint8Array(buffer), 'affine-cloud');
   }
 }

packages/y-indexeddb/src/utils.ts

@@ -173,3 +173,33 @@ export async function overwriteBinary(
     ],
   });
 }
+
+export async function pushBinary(
+  guid: string,
+  update: UpdateMessage['update'],
+  dbName = DEFAULT_DB_NAME
+) {
+  const dbPromise = openDB<BlockSuiteBinaryDB>(dbName, dbVersion, {
+    upgrade: upgradeDB,
+  });
+  const db = await dbPromise;
+  const t = db.transaction('workspace', 'readwrite').objectStore('workspace');
+  const doc = await t.get(guid);
+  if (!doc) {
+    await t.put({
+      id: guid,
+      updates: [
+        {
+          timestamp: Date.now(),
+          update,
+        },
+      ],
+    });
+  } else {
+    doc.updates.push({
+      timestamp: Date.now(),
+      update,
+    });
+    await t.put(doc);
+  }
+}