Merge branch 'canary' into stable

packages/frontend/workspace/package.json

@@ -4,15 +4,13 @@
   "exports": {
     "./atom": "./src/atom.ts",
     "./manager": "./src/manager/index.ts",
-    "./type": "./src/type.ts",
-    "./migration": "./src/migration/index.ts",
     "./local/crud": "./src/local/crud.ts",
-    "./affine": "./src/affine/index.ts",
     "./affine/*": "./src/affine/*.ts",
     "./providers": "./src/providers/index.ts"
   },
   "peerDependencies": {
     "@blocksuite/blocks": "*",
+    "@blocksuite/global": "*",
     "@blocksuite/store": "*"
   },
   "dependencies": {
@@ -23,14 +21,16 @@
     "@toeverything/hooks": "workspace:*",
     "@toeverything/y-indexeddb": "workspace:*",
     "async-call-rpc": "^6.3.1",
+    "idb": "^7.1.1",
+    "idb-keyval": "^6.2.1",
     "is-svg": "^5.0.0",
-    "jotai": "^2.4.3",
+    "jotai": "^2.5.1",
     "js-base64": "^3.7.5",
     "ky": "^1.0.1",
     "lib0": "^0.2.87",
     "lodash-es": "^4.17.21",
-    "nanoid": "^5.0.1",
-    "next-auth": "^4.23.2",
+    "nanoid": "^5.0.3",
+    "next-auth": "^4.24.5",
     "react": "18.2.0",
     "react-dom": "18.2.0",
     "socket.io-client": "^4.7.2",
@@ -38,7 +38,7 @@
     "valtio": "^1.11.2",
     "y-protocols": "^1.0.6",
     "y-provider": "workspace:*",
-    "yjs": "^13.6.8",
+    "yjs": "^13.6.10",
     "zod": "^3.22.4"
   },
   "devDependencies": {
@@ -48,5 +48,5 @@
     "vitest": "0.34.6",
     "ws": "^8.14.2"
   },
-  "version": "0.10.2"
+  "version": "0.10.3-canary.2"
 }

packages/frontend/workspace/src/affine/crud.ts

@@ -9,16 +9,10 @@ import {
   getWorkspaceQuery,
   getWorkspacesQuery,
 } from '@affine/graphql';
-import { createAffineDataSource } from '@affine/workspace/affine/index';
 import { createIndexeddbStorage, Workspace } from '@blocksuite/store';
 import { migrateLocalBlobStorage } from '@toeverything/infra/blocksuite';
-import {
-  createIndexedDBProvider,
-  DEFAULT_DB_NAME,
-} from '@toeverything/y-indexeddb';
 import { getSession } from 'next-auth/react';
 import { proxy } from 'valtio/vanilla';
-import { syncDataSourceFromDoc } from 'y-provider';
 
 import { getOrCreateWorkspace } from '../manager';
 import { fetcher } from './gql';
@@ -59,45 +53,41 @@ export const CRUD: WorkspaceCRUD<WorkspaceFlavour.AFFINE_CLOUD> = {
       WorkspaceFlavour.AFFINE_CLOUD
     );
 
-    Y.applyUpdate(
-      newBlockSuiteWorkspace.doc,
-      Y.encodeStateAsUpdate(upstreamWorkspace.doc)
-    );
+    if (environment.isDesktop) {
+      // this will clone all data from existing db to new db file, including docs and blobs
+      await window.apis.workspace.clone(
+        upstreamWorkspace.id,
+        createWorkspace.id
+      );
 
-    await Promise.all(
-      [...upstreamWorkspace.doc.subdocs].map(async subdoc => {
-        subdoc.load();
-        return subdoc.whenLoaded.then(() => {
-          newBlockSuiteWorkspace.doc.subdocs.forEach(newSubdoc => {
-            if (newSubdoc.guid === subdoc.guid) {
-              Y.applyUpdate(newSubdoc, Y.encodeStateAsUpdate(subdoc));
-            }
+      // skip apply updates in memory and we will use providers to sync data from db
+    } else {
+      Y.applyUpdate(
+        newBlockSuiteWorkspace.doc,
+        Y.encodeStateAsUpdate(upstreamWorkspace.doc)
+      );
+
+      await Promise.all(
+        [...upstreamWorkspace.doc.subdocs].map(async subdoc => {
+          subdoc.load();
+          return subdoc.whenLoaded.then(() => {
+            newBlockSuiteWorkspace.doc.subdocs.forEach(newSubdoc => {
+              if (newSubdoc.guid === subdoc.guid) {
+                Y.applyUpdate(newSubdoc, Y.encodeStateAsUpdate(subdoc));
+              }
+            });
           });
+        })
+      );
+
+      migrateLocalBlobStorage(upstreamWorkspace.id, createWorkspace.id)
+        .then(() => deleteLocalBlobStorage(upstreamWorkspace.id))
+        .catch(e => {
+          console.error('error when moving blob storage:', e);
         });
-      })
-    );
+      // todo(himself65): delete old workspace in the future
+    }
 
-    const datasource = createAffineDataSource(
-      createWorkspace.id,
-      newBlockSuiteWorkspace.doc,
-      newBlockSuiteWorkspace.awarenessStore.awareness
-    );
-
-    const disconnect = datasource.onDocUpdate(() => {});
-    await syncDataSourceFromDoc(upstreamWorkspace.doc, datasource);
-    disconnect();
-
-    const provider = createIndexedDBProvider(
-      newBlockSuiteWorkspace.doc,
-      DEFAULT_DB_NAME
-    );
-    provider.connect();
-    migrateLocalBlobStorage(upstreamWorkspace.id, createWorkspace.id)
-      .then(() => deleteLocalBlobStorage(upstreamWorkspace.id))
-      .catch(e => {
-        console.error('error when moving blob storage:', e);
-      });
-    // todo(himself65): delete old workspace in the future
     return createWorkspace.id;
   },
   delete: async workspace => {

packages/frontend/workspace/src/affine/download.ts

@@ -0,0 +1,37 @@
+import { fetchWithTraceReport } from '@affine/graphql';
+
+const hashMap = new Map<string, CloudDoc>();
+type DocPublishMode = 'edgeless' | 'page';
+
+export type CloudDoc = {
+  arrayBuffer: ArrayBuffer;
+  publishMode: DocPublishMode;
+};
+
+export async function downloadBinaryFromCloud(
+  rootGuid: string,
+  pageGuid: string
+): Promise<CloudDoc | null> {
+  const cached = hashMap.get(`${rootGuid}/${pageGuid}`);
+  if (cached) {
+    return cached;
+  }
+  const response = await fetchWithTraceReport(
+    runtimeConfig.serverUrlPrefix +
+      `/api/workspaces/${rootGuid}/docs/${pageGuid}`,
+    {
+      priority: 'high',
+    }
+  );
+  if (response.ok) {
+    const publishMode = (response.headers.get('publish-mode') ||
+      'page') as DocPublishMode;
+    const arrayBuffer = await response.arrayBuffer();
+    hashMap.set(`${rootGuid}/${pageGuid}`, { arrayBuffer, publishMode });
+
+    // return both arrayBuffer and publish mode
+    return { arrayBuffer, publishMode };
+  }
+
+  return null;
+}

packages/frontend/workspace/src/affine/index.ts

@@ -1,263 +0,0 @@
-import { DebugLogger } from '@affine/debug';
-import type { Socket } from 'socket.io-client';
-import { Manager } from 'socket.io-client';
-import {
-  applyAwarenessUpdate,
-  type Awareness,
-  encodeAwarenessUpdate,
-  removeAwarenessStates,
-} from 'y-protocols/awareness';
-import type { DocDataSource } from 'y-provider';
-import type { Doc } from 'yjs';
-
-import { MultipleBatchSyncSender } from './batch-sync-sender';
-import {
-  type AwarenessChanges,
-  base64ToUint8Array,
-  uint8ArrayToBase64,
-} from './utils';
-
-let ioManager: Manager | null = null;
-
-// use lazy initialization to avoid global side effect
-function getIoManager(): Manager {
-  if (ioManager) {
-    return ioManager;
-  }
-  ioManager = new Manager(runtimeConfig.serverUrlPrefix + '/', {
-    autoConnect: false,
-    transports: ['websocket'],
-  });
-  return ioManager;
-}
-
-const logger = new DebugLogger('affine:sync');
-
-export const createAffineDataSource = (
-  id: string,
-  rootDoc: Doc,
-  awareness: Awareness
-) => {
-  if (id !== rootDoc.guid) {
-    console.warn('important!! please use doc.guid as roomName');
-  }
-
-  logger.debug('createAffineDataSource', id, rootDoc.guid);
-  const socket = getIoManager().socket('/');
-  const syncSender = new MultipleBatchSyncSender(async (guid, updates) => {
-    const payload = await Promise.all(
-      updates.map(update => uint8ArrayToBase64(update))
-    );
-
-    return new Promise(resolve => {
-      socket.emit(
-        'client-update-v2',
-        {
-          workspaceId: rootDoc.guid,
-          guid,
-          updates: payload,
-        },
-        (response: {
-          // TODO: reuse `EventError` with server
-          error?: any;
-          data: any;
-        }) => {
-          // TODO: raise error with different code to users
-          if (response.error) {
-            logger.error('client-update-v2 error', {
-              workspaceId: rootDoc.guid,
-              guid,
-              response,
-            });
-          }
-
-          resolve({
-            accepted: !response.error,
-            // TODO: reuse `EventError` with server
-            retry: response.error?.code === 'INTERNAL',
-          });
-        }
-      );
-    });
-  });
-
-  return {
-    get socket() {
-      return socket;
-    },
-    queryDocState: async (guid, options) => {
-      const stateVector = options?.stateVector
-        ? await uint8ArrayToBase64(options.stateVector)
-        : undefined;
-
-      return new Promise((resolve, reject) => {
-        logger.debug('doc-load-v2', {
-          workspaceId: rootDoc.guid,
-          guid,
-          stateVector,
-        });
-        socket.emit(
-          'doc-load-v2',
-          {
-            workspaceId: rootDoc.guid,
-            guid,
-            stateVector,
-          },
-          (
-            response: // TODO: reuse `EventError` with server
-            { error: any } | { data: { missing: string; state: string } }
-          ) => {
-            logger.debug('doc-load callback', {
-              workspaceId: rootDoc.guid,
-              guid,
-              stateVector,
-              response,
-            });
-
-            if ('error' in response) {
-              // TODO: result `EventError` with server
-              if (response.error.code === 'DOC_NOT_FOUND') {
-                resolve(false);
-              } else {
-                reject(new Error(response.error.message));
-              }
-            } else {
-              resolve({
-                missing: base64ToUint8Array(response.data.missing),
-                state: response.data.state
-                  ? base64ToUint8Array(response.data.state)
-                  : undefined,
-              });
-            }
-          }
-        );
-      });
-    },
-    sendDocUpdate: async (guid: string, update: Uint8Array) => {
-      logger.debug('client-update-v2', {
-        workspaceId: rootDoc.guid,
-        guid,
-        update,
-      });
-
-      await syncSender.send(guid, update);
-    },
-    onDocUpdate: callback => {
-      const onUpdate = async (message: {
-        workspaceId: string;
-        guid: string;
-        updates: string[];
-      }) => {
-        if (message.workspaceId === rootDoc.guid) {
-          message.updates.forEach(update => {
-            callback(message.guid, base64ToUint8Array(update));
-          });
-        }
-      };
-      let destroyAwareness = () => {};
-      socket.on('server-updates', onUpdate);
-      socket.on('connect', () => {
-        socket.emit(
-          'client-handshake',
-          rootDoc.guid,
-          (response: { error?: any }) => {
-            if (!response.error) {
-              syncSender.start();
-              destroyAwareness = setupAffineAwareness(
-                socket,
-                rootDoc,
-                awareness
-              );
-            }
-          }
-        );
-      });
-
-      socket.connect();
-      return () => {
-        syncSender.stop();
-        socket.emit('client-leave', rootDoc.guid);
-        socket.off('server-updates', onUpdate);
-        destroyAwareness();
-        socket.disconnect();
-      };
-    },
-  } satisfies DocDataSource & { readonly socket: Socket };
-};
-
-function setupAffineAwareness(
-  conn: Socket,
-  rootDoc: Doc,
-  awareness: Awareness
-) {
-  const awarenessBroadcast = ({
-    workspaceId,
-    awarenessUpdate,
-  }: {
-    workspaceId: string;
-    awarenessUpdate: string;
-  }) => {
-    if (workspaceId !== rootDoc.guid) {
-      return;
-    }
-    applyAwarenessUpdate(
-      awareness,
-      base64ToUint8Array(awarenessUpdate),
-      'server'
-    );
-  };
-
-  const awarenessUpdate = (changes: AwarenessChanges, origin: unknown) => {
-    if (origin === 'server') {
-      return;
-    }
-
-    const changedClients = Object.values(changes).reduce((res, cur) => [
-      ...res,
-      ...cur,
-    ]);
-
-    const update = encodeAwarenessUpdate(awareness, changedClients);
-    uint8ArrayToBase64(update)
-      .then(encodedUpdate => {
-        conn.emit('awareness-update', {
-          workspaceId: rootDoc.guid,
-          awarenessUpdate: encodedUpdate,
-        });
-      })
-      .catch(err => logger.error(err));
-  };
-
-  const newClientAwarenessInitHandler = () => {
-    const awarenessUpdate = encodeAwarenessUpdate(awareness, [
-      awareness.clientID,
-    ]);
-    uint8ArrayToBase64(awarenessUpdate)
-      .then(encodedAwarenessUpdate => {
-        conn.emit('awareness-update', {
-          guid: rootDoc.guid,
-          awarenessUpdate: encodedAwarenessUpdate,
-        });
-      })
-      .catch(err => logger.error(err));
-  };
-
-  const windowBeforeUnloadHandler = () => {
-    removeAwarenessStates(awareness, [awareness.clientID], 'window unload');
-  };
-
-  conn.on('server-awareness-broadcast', awarenessBroadcast);
-  conn.on('new-client-awareness-init', newClientAwarenessInitHandler);
-  awareness.on('update', awarenessUpdate);
-
-  window.addEventListener('beforeunload', windowBeforeUnloadHandler);
-
-  conn.emit('awareness-init', rootDoc.guid);
-
-  return () => {
-    awareness.off('update', awarenessUpdate);
-    conn.off('server-awareness-broadcast', awarenessBroadcast);
-    conn.off('new-client-awareness-init', newClientAwarenessInitHandler);
-    window.removeEventListener('unload', windowBeforeUnloadHandler);
-  };
-}

packages/frontend/workspace/src/affine/sync.ts

@@ -1,120 +0,0 @@
-import { DebugLogger } from '@affine/debug';
-import { createIndexeddbStorage } from '@blocksuite/store';
-import {
-  createIndexedDBDatasource,
-  DEFAULT_DB_NAME,
-  downloadBinary,
-} from '@toeverything/y-indexeddb';
-import { syncDataSource } from 'y-provider';
-import type { Doc } from 'yjs';
-import { applyUpdate } from 'yjs';
-
-import { createCloudBlobStorage } from '../blob/cloud-blob-storage';
-import { createAffineDataSource } from '.';
-import { CRUD } from './crud';
-
-const performanceLogger = new DebugLogger('performance:sync');
-let abortController: AbortController | undefined;
-
-const downloadRootFromIndexedDB = async (
-  rootGuid: string,
-  doc: Doc,
-  signal: AbortSignal
-): Promise<void> => {
-  if (signal.aborted) {
-    return;
-  }
-  const update = await downloadBinary(rootGuid);
-  if (update !== false) {
-    applyUpdate(doc, update);
-  }
-};
-
-export async function startSync() {
-  performanceLogger.info('start');
-
-  abortController = new AbortController();
-  const signal = abortController.signal;
-  const workspaces = await CRUD.list();
-  performanceLogger.info('CRUD list');
-
-  const syncDocPromises = workspaces.map(workspace =>
-    downloadRootFromIndexedDB(
-      workspace.id,
-      workspace.blockSuiteWorkspace.doc,
-      signal
-    )
-  );
-  await Promise.all(syncDocPromises);
-  performanceLogger.info('all sync promise');
-
-  const syncPromises = workspaces.map(workspace => {
-    const remoteDataSource = createAffineDataSource(
-      workspace.id,
-      workspace.blockSuiteWorkspace.doc,
-      workspace.blockSuiteWorkspace.awarenessStore.awareness
-    );
-    const indexeddbDataSource = createIndexedDBDatasource({
-      dbName: DEFAULT_DB_NAME,
-    });
-    return syncDataSource(
-      (): string[] => [
-        workspace.blockSuiteWorkspace.doc.guid,
-        ...[...workspace.blockSuiteWorkspace.doc.subdocs].map(doc => doc.guid),
-      ],
-      remoteDataSource,
-      indexeddbDataSource
-    );
-  });
-
-  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([...syncPromises, ...syncBlobPromises]);
-  performanceLogger.info('sync done');
-}
-
-export async function stopSync() {
-  abortController?.abort();
-}

packages/frontend/workspace/src/atom.ts

@@ -74,15 +74,12 @@ const rootWorkspacesMetadataPrimitiveAtom = atom<Promise<
 
 type Getter = <Value>(atom: Atom<Value>) => Value;
 
-type FetchMetadata = (
-  get: Getter,
-  options: { signal: AbortSignal }
-) => Promise<RootWorkspaceMetadata[]>;
+type FetchMetadata = (get: Getter) => Promise<RootWorkspaceMetadata[]>;
 
 /**
  * @internal
  */
-const fetchMetadata: FetchMetadata = async (get, { signal }) => {
+const fetchMetadata: FetchMetadata = async get => {
   performanceJotaiLogger.info('fetch metadata start');
 
   const WorkspaceAdapters = get(workspaceAdaptersAtom);
@@ -111,23 +108,6 @@ const fetchMetadata: FetchMetadata = async (get, { signal }) => {
       }
       return [];
     };
-
-    const maybeMetadata = loadFromLocalStorage();
-
-    // migration step, only data in `METADATA_STORAGE_KEY` will be migrated
-    if (
-      maybeMetadata.some(meta => !('version' in meta)) &&
-      !window.$migrationDone
-    ) {
-      await new Promise<void>((resolve, reject) => {
-        signal.addEventListener('abort', () => reject(), { once: true });
-        window.addEventListener('migration-done', () => resolve(), {
-          once: true,
-        });
-      });
-      performanceJotaiLogger.info('migration done');
-    }
-
     metadata.push(...loadFromLocalStorage());
   }
   // step 2: fetch from adapters
@@ -211,14 +191,14 @@ const fetchMetadata: FetchMetadata = async (get, { signal }) => {
 
 const rootWorkspacesMetadataPromiseAtom = atom<
   Promise<RootWorkspaceMetadata[]>
->(async (get, { signal }) => {
+>(async get => {
   const primitiveMetadata = get(rootWorkspacesMetadataPrimitiveAtom);
   assertEquals(
     primitiveMetadata,
     null,
     'rootWorkspacesMetadataPrimitiveAtom should be null'
   );
-  return fetchMetadata(get, { signal });
+  return fetchMetadata(get);
 });
 
 type SetStateAction<Value> = Value | ((prev: Value) => Value);
@@ -268,7 +248,6 @@ export const rootWorkspacesMetadataAtom = atom<
       }
 
       if (newWorkspaceId) {
-        set(currentPageIdAtom, null);
         set(currentWorkspaceIdAtom, newWorkspaceId);
       }
       return metadata;
@@ -277,11 +256,7 @@ export const rootWorkspacesMetadataAtom = atom<
 );
 
 export const refreshRootMetadataAtom = atom(null, (get, set) => {
-  const abortController = new AbortController();
-  set(
-    rootWorkspacesMetadataPrimitiveAtom,
-    fetchMetadata(get, { signal: abortController.signal })
-  );
+  set(rootWorkspacesMetadataPrimitiveAtom, fetchMetadata(get));
 });
 
 // blocksuite atoms,

packages/frontend/workspace/src/blob/__tests__/util.spec.ts

@@ -0,0 +1,13 @@
+import { Buffer } from 'node:buffer';
+
+import { describe, expect, test } from 'vitest';
+
+import { isSvgBuffer } from '../util';
+
+describe('isSvgBuffer', () => {
+  test('basic', async () => {
+    expect(isSvgBuffer(Buffer.from('<svg></svg>'))).toBe(true);
+    expect(isSvgBuffer(Buffer.from(' \n\r\t<svg></svg>'))).toBe(true);
+    expect(isSvgBuffer(Buffer.from('<123>'))).toBe(false);
+  });
+});

packages/frontend/workspace/src/blob/cloud-blob-storage.ts

@@ -1,77 +0,0 @@
-import {
-  checkBlobSizesQuery,
-  deleteBlobMutation,
-  fetchWithTraceReport,
-  listBlobsQuery,
-  setBlobMutation,
-} from '@affine/graphql';
-import { fetcher } from '@affine/workspace/affine/gql';
-import type { BlobStorage } from '@blocksuite/store';
-
-import { predefinedStaticFiles } from './local-static-storage';
-import { bufferToBlob } from './util';
-
-export const createCloudBlobStorage = (workspaceId: string): BlobStorage => {
-  return {
-    crud: {
-      get: async key => {
-        const suffix = predefinedStaticFiles.includes(key)
-          ? `/static/${key}`
-          : `/api/workspaces/${workspaceId}/blobs/${key}`;
-
-        return fetchWithTraceReport(
-          runtimeConfig.serverUrlPrefix + suffix
-        ).then(async res => {
-          if (!res.ok) {
-            // status not in the range 200-299
-            return null;
-          }
-          return bufferToBlob(await res.arrayBuffer());
-        });
-      },
-      set: async (key, value) => {
-        const {
-          checkBlobSize: { size },
-        } = await fetcher({
-          query: checkBlobSizesQuery,
-          variables: {
-            workspaceId,
-            size: value.size,
-          },
-        });
-
-        if (size <= 0) {
-          throw new Error('Blob size limit exceeded');
-        }
-
-        const result = await fetcher({
-          query: setBlobMutation,
-          variables: {
-            workspaceId,
-            blob: new File([value], key),
-          },
-        });
-        console.assert(result.setBlob === key, 'Blob hash mismatch');
-        return key;
-      },
-      list: async () => {
-        const result = await fetcher({
-          query: listBlobsQuery,
-          variables: {
-            workspaceId,
-          },
-        });
-        return result.listBlobs;
-      },
-      delete: async (key: string) => {
-        await fetcher({
-          query: deleteBlobMutation,
-          variables: {
-            workspaceId,
-            hash: key,
-          },
-        });
-      },
-    },
-  };
-};

packages/frontend/workspace/src/blob/engine.ts

@@ -0,0 +1,139 @@
+import { DebugLogger } from '@affine/debug';
+import { difference } from 'lodash-es';
+
+const logger = new DebugLogger('affine:blob-engine');
+
+export class BlobEngine {
+  constructor(
+    private local: BlobStorage,
+    private remotes: BlobStorage[]
+  ) {}
+
+  get storages() {
+    return [this.local, ...this.remotes];
+  }
+
+  async sync() {
+    if (this.local.readonly) {
+      return;
+    }
+    logger.debug('start syncing blob...');
+    for (const remote of this.remotes) {
+      let localList;
+      let remoteList;
+      try {
+        localList = await this.local.list();
+        remoteList = await remote.list();
+      } catch (err) {
+        logger.error(`error when sync`, err);
+        continue;
+      }
+
+      if (!remote.readonly) {
+        const needUpload = difference(localList, remoteList);
+        for (const key of needUpload) {
+          try {
+            const data = await this.local.get(key);
+            if (data) {
+              await remote.set(key, data);
+            }
+          } catch (err) {
+            logger.error(
+              `error when sync ${key} from [${this.local.name}] to [${remote.name}]`,
+              err
+            );
+          }
+        }
+      }
+
+      const needDownload = difference(remoteList, localList);
+
+      for (const key of needDownload) {
+        try {
+          const data = await remote.get(key);
+          if (data) {
+            await this.local.set(key, data);
+          }
+        } catch (err) {
+          logger.error(
+            `error when sync ${key} from [${remote.name}] to [${this.local.name}]`,
+            err
+          );
+        }
+      }
+    }
+
+    logger.debug('finish syncing blob');
+  }
+
+  async get(key: string) {
+    logger.debug('get blob', key);
+    for (const storage of this.storages) {
+      const data = await storage.get(key);
+      if (data) {
+        return data;
+      }
+    }
+    return undefined;
+  }
+
+  async set(key: string, value: Blob) {
+    if (this.local.readonly) {
+      throw new Error('local peer is readonly');
+    }
+
+    // await upload to the local peer
+    await this.local.set(key, value);
+
+    // uploads to other peers in the background
+    Promise.allSettled(
+      this.remotes
+        .filter(r => !r.readonly)
+        .map(peer =>
+          peer.set(key, value).catch(err => {
+            logger.error('error when upload to peer', err);
+          })
+        )
+    )
+      .then(result => {
+        if (result.some(({ status }) => status === 'rejected')) {
+          logger.error(
+            `blob ${key} update finish, but some peers failed to update`
+          );
+        } else {
+          logger.debug(`blob ${key} update finish`);
+        }
+      })
+      .catch(() => {
+        // Promise.allSettled never reject
+      });
+  }
+
+  async delete(_key: string) {
+    // not supported
+  }
+
+  async list() {
+    const blobList = new Set<string>();
+
+    for (const peer of this.storages) {
+      const list = await peer.list();
+      if (list) {
+        for (const blob of list) {
+          blobList.add(blob);
+        }
+      }
+    }
+
+    return Array.from(blobList);
+  }
+}
+
+export interface BlobStorage {
+  name: string;
+  readonly: boolean;
+  get: (key: string) => Promise<Blob | undefined>;
+  set: (key: string, value: Blob) => Promise<void>;
+  delete: (key: string) => Promise<void>;
+  list: () => Promise<string[]>;
+}

packages/frontend/workspace/src/blob/index.ts

@@ -0,0 +1,37 @@
+import { BlobEngine } from './engine';
+import {
+  createAffineCloudBlobStorage,
+  createIndexeddbBlobStorage,
+  createSQLiteBlobStorage,
+  createStaticBlobStorage,
+} from './storage';
+
+export * from './engine';
+export * from './storage';
+
+export function createLocalBlobStorage(workspaceId: string) {
+  if (environment.isDesktop) {
+    return createSQLiteBlobStorage(workspaceId);
+  } else {
+    return createIndexeddbBlobStorage(workspaceId);
+  }
+}
+
+export function createLocalBlobEngine(workspaceId: string) {
+  return new BlobEngine(createLocalBlobStorage(workspaceId), [
+    createStaticBlobStorage(),
+  ]);
+}
+
+export function createAffineCloudBlobEngine(workspaceId: string) {
+  return new BlobEngine(createLocalBlobStorage(workspaceId), [
+    createStaticBlobStorage(),
+    createAffineCloudBlobStorage(workspaceId),
+  ]);
+}
+
+export function createAffinePublicBlobEngine(workspaceId: string) {
+  return new BlobEngine(createAffineCloudBlobStorage(workspaceId), [
+    createStaticBlobStorage(),
+  ]);
+}

packages/frontend/workspace/src/blob/sqlite-blob-storage.ts

@@ -1,34 +0,0 @@
-import { assertExists } from '@blocksuite/global/utils';
-import type { BlobStorage } from '@blocksuite/store';
-
-import { bufferToBlob } from './util';
-
-export const createSQLiteStorage = (workspaceId: string): BlobStorage => {
-  const apis = window.apis;
-  assertExists(apis);
-  return {
-    crud: {
-      get: async (key: string) => {
-        const buffer = await apis.db.getBlob(workspaceId, key);
-        if (buffer) {
-          return bufferToBlob(buffer);
-        }
-        return null;
-      },
-      set: async (key: string, value: Blob) => {
-        await apis.db.addBlob(
-          workspaceId,
-          key,
-          new Uint8Array(await value.arrayBuffer())
-        );
-        return key;
-      },
-      delete: async (key: string) => {
-        return apis.db.deleteBlob(workspaceId, key);
-      },
-      list: async () => {
-        return apis.db.getBlobKeys(workspaceId);
-      },
-    },
-  };
-};

packages/frontend/workspace/src/blob/storage/affine-cloud.ts

@@ -0,0 +1,76 @@
+import {
+  checkBlobSizesQuery,
+  deleteBlobMutation,
+  fetchWithTraceReport,
+  listBlobsQuery,
+  setBlobMutation,
+} from '@affine/graphql';
+import { fetcher } from '@affine/workspace/affine/gql';
+
+import type { BlobStorage } from '../engine';
+
+export const createAffineCloudBlobStorage = (
+  workspaceId: string
+): BlobStorage => {
+  return {
+    name: 'affine-cloud',
+    readonly: false,
+    get: async key => {
+      const suffix = key.startsWith('/')
+        ? key
+        : `/api/workspaces/${workspaceId}/blobs/${key}`;
+
+      return fetchWithTraceReport(runtimeConfig.serverUrlPrefix + suffix).then(
+        async res => {
+          if (!res.ok) {
+            // status not in the range 200-299
+            return undefined;
+          }
+          return await res.blob();
+        }
+      );
+    },
+    set: async (key, value) => {
+      const {
+        checkBlobSize: { size },
+      } = await fetcher({
+        query: checkBlobSizesQuery,
+        variables: {
+          workspaceId,
+          size: value.size,
+        },
+      });
+
+      if (size <= 0) {
+        throw new Error('Blob size limit exceeded');
+      }
+
+      const result = await fetcher({
+        query: setBlobMutation,
+        variables: {
+          workspaceId,
+          blob: new File([value], key),
+        },
+      });
+      console.assert(result.setBlob === key, 'Blob hash mismatch');
+    },
+    list: async () => {
+      const result = await fetcher({
+        query: listBlobsQuery,
+        variables: {
+          workspaceId,
+        },
+      });
+      return result.listBlobs;
+    },
+    delete: async (key: string) => {
+      await fetcher({
+        query: deleteBlobMutation,
+        variables: {
+          workspaceId,
+          hash: key,
+        },
+      });
+    },
+  };
+};

packages/frontend/workspace/src/blob/storage/index.ts

@@ -0,0 +1,4 @@
+export * from './affine-cloud';
+export * from './indexeddb';
+export * from './sqlite';
+export * from './static';

packages/frontend/workspace/src/blob/storage/indexeddb.ts

@@ -0,0 +1,32 @@
+import { createStore, del, get, keys, set } from 'idb-keyval';
+
+import type { BlobStorage } from '../engine';
+
+export const createIndexeddbBlobStorage = (
+  workspaceId: string
+): BlobStorage => {
+  const db = createStore(`${workspaceId}_blob`, 'blob');
+  const mimeTypeDb = createStore(`${workspaceId}_blob_mime`, 'blob_mime');
+  return {
+    name: 'indexeddb',
+    readonly: false,
+    get: async (key: string) => {
+      const res = await get<ArrayBuffer>(key, db);
+      if (res) {
+        return new Blob([res], { type: await get(key, mimeTypeDb) });
+      }
+      return undefined;
+    },
+    set: async (key: string, value: Blob) => {
+      await set(key, await value.arrayBuffer(), db);
+      await set(key, value.type, mimeTypeDb);
+    },
+    delete: async (key: string) => {
+      await del(key, db);
+      await del(key, mimeTypeDb);
+    },
+    list: async () => {
+      return keys<string>(db);
+    },
+  };
+};

packages/frontend/workspace/src/blob/storage/sqlite.ts

@@ -0,0 +1,33 @@
+import { assertExists } from '@blocksuite/global/utils';
+
+import type { BlobStorage } from '../engine';
+import { bufferToBlob } from '../util';
+
+export const createSQLiteBlobStorage = (workspaceId: string): BlobStorage => {
+  const apis = window.apis;
+  assertExists(apis);
+  return {
+    name: 'sqlite',
+    readonly: false,
+    get: async (key: string) => {
+      const buffer = await apis.db.getBlob(workspaceId, key);
+      if (buffer) {
+        return bufferToBlob(buffer);
+      }
+      return undefined;
+    },
+    set: async (key: string, value: Blob) => {
+      await apis.db.addBlob(
+        workspaceId,
+        key,
+        new Uint8Array(await value.arrayBuffer())
+      );
+    },
+    delete: async (key: string) => {
+      return apis.db.deleteBlob(workspaceId, key);
+    },
+    list: async () => {
+      return apis.db.getBlobKeys(workspaceId);
+    },
+  };
+};

packages/frontend/workspace/src/blob/storage/static.ts

@@ -1,6 +1,4 @@
-import type { BlobStorage } from '@blocksuite/store';
-
-import { bufferToBlob } from './util';
+import type { BlobStorage } from '../engine';
 
 export const predefinedStaticFiles = [
   '029uztLz2CzJezK7UUhrbGiWUdZ0J7NVs_qR6RDsvb8=',
@@ -38,38 +36,36 @@ export const predefinedStaticFiles = [
   'v2yF7lY2L5rtorTtTmYFsoMb9dBPKs5M1y9cUKxcI1M=',
 ];
 
-export const createStaticStorage = (): BlobStorage => {
+export const createStaticBlobStorage = (): BlobStorage => {
   return {
-    crud: {
-      get: async (key: string) => {
-        const isStaticResource =
-          predefinedStaticFiles.includes(key) || key.startsWith('/static/');
+    name: 'static',
+    readonly: true,
+    get: async (key: string) => {
+      const isStaticResource =
+        predefinedStaticFiles.includes(key) || key.startsWith('/static/');
 
-        if (!isStaticResource) {
-          return null;
-        }
+      if (!isStaticResource) {
+        return undefined;
+      }
 
-        const path = key.startsWith('/static/') ? key : `/static/${key}`;
-        const response = await fetch(path);
+      const path = key.startsWith('/static/') ? key : `/static/${key}`;
+      const response = await fetch(path);
 
-        if (response.ok) {
-          const buffer = await response.arrayBuffer();
-          return bufferToBlob(buffer);
-        }
+      if (response.ok) {
+        return await response.blob();
+      }
 
-        return null;
-      },
-      set: async (key: string) => {
-        // ignore
-        return key;
-      },
-      delete: async () => {
-        // ignore
-      },
-      list: async () => {
-        // ignore
-        return [];
-      },
+      return undefined;
+    },
+    set: async () => {
+      // ignore
+    },
+    delete: async () => {
+      // ignore
+    },
+    list: async () => {
+      // ignore
+      return [];
     },
   };
 };

packages/frontend/workspace/src/blob/util.ts

@@ -1,8 +1,39 @@
 import isSvg from 'is-svg';
 
+function fastCheckIsNotSvg(buffer: Uint8Array) {
+  // check first non-whitespace character is not '<svg'
+  for (let i = 0; i < buffer.length; i++) {
+    const ch = buffer[i];
+
+    // skip whitespace
+    if (
+      ch === 0x20 /* \s */ ||
+      ch === 0x09 /* \t */ ||
+      ch === 0x0b /* \v */ ||
+      ch === 0x0c /* \f */ ||
+      ch === 0x0a /* \n */ ||
+      ch === 0x0d /* \r */ ||
+      ch === 0xa0
+    ) {
+      continue;
+    }
+
+    return !(
+      buffer[i] === /* '<' */ 0x3c &&
+      buffer[i + 1] === /* 's' */ 0x73 &&
+      buffer[i + 2] === /* 'v' */ 0x76 &&
+      buffer[i + 3] === /* 'g' */ 0x67
+    );
+  }
+
+  return true;
+}
+
 // this has a overhead of converting to string for testing if it is svg.
-// is there a more performant way?
 export function isSvgBuffer(buffer: Uint8Array) {
+  if (fastCheckIsNotSvg(buffer)) {
+    return false;
+  }
   const decoder = new TextDecoder('utf-8');
   const str = decoder.decode(buffer);
   return isSvg(str);

packages/frontend/workspace/src/manager/index.ts

@@ -3,16 +3,19 @@ import type { BlockSuiteFeatureFlags } from '@affine/env/global';
 import { WorkspaceFlavour } from '@affine/env/workspace';
 import { createAffinePublicProviders } from '@affine/workspace/providers';
 import { __unstableSchemas, AffineSchemas } from '@blocksuite/blocks/models';
-import type { DocProviderCreator, StoreOptions } from '@blocksuite/store';
-import { createIndexeddbStorage, Schema, Workspace } from '@blocksuite/store';
+import type { DocProviderCreator } from '@blocksuite/store';
+import { Schema, Workspace } from '@blocksuite/store';
 import { INTERNAL_BLOCKSUITE_HASH_MAP } from '@toeverything/infra/__internal__/workspace';
 import { nanoid } from 'nanoid';
 import type { Doc } from 'yjs';
 import type { Transaction } from 'yjs';
 
-import { createCloudBlobStorage } from '../blob/cloud-blob-storage';
-import { createStaticStorage } from '../blob/local-static-storage';
-import { createSQLiteStorage } from '../blob/sqlite-blob-storage';
+import type { BlobEngine } from '../blob';
+import {
+  createAffineCloudBlobEngine,
+  createAffinePublicBlobEngine,
+  createLocalBlobEngine,
+} from '../blob';
 import { createAffineProviders, createLocalProviders } from '../providers';
 
 function setEditorFlags(workspace: Workspace) {
@@ -81,6 +84,12 @@ const createMonitor = (doc: Doc) => {
   });
 };
 
+const workspaceBlobEngineWeakMap = new WeakMap<Workspace, BlobEngine>();
+export function getBlobEngine(workspace: Workspace) {
+  // temporary solution to get blob engine from workspace
+  return workspaceBlobEngineWeakMap.get(workspace);
+}
+
 // if not exist, create a new workspace
 export function getOrCreateWorkspace(
   id: string,
@@ -91,48 +100,47 @@ export function getOrCreateWorkspace(
     return INTERNAL_BLOCKSUITE_HASH_MAP.get(id) as Workspace;
   }
 
-  const blobStorages: StoreOptions['blobStorages'] = [];
+  let blobEngine: BlobEngine;
   if (flavour === WorkspaceFlavour.AFFINE_CLOUD) {
-    if (isBrowser) {
-      blobStorages.push(createIndexeddbStorage);
-      blobStorages.push(createCloudBlobStorage);
-      if (environment.isDesktop && runtimeConfig.enableSQLiteProvider) {
-        blobStorages.push(createSQLiteStorage);
-      }
-      providerCreators.push(...createAffineProviders());
-
-      // todo(JimmFly): add support for cloud storage
-    }
+    blobEngine = createAffineCloudBlobEngine(id);
+    providerCreators.push(...createAffineProviders());
   } else if (flavour === WorkspaceFlavour.LOCAL) {
-    if (isBrowser) {
-      blobStorages.push(createIndexeddbStorage);
-      if (environment.isDesktop && runtimeConfig.enableSQLiteProvider) {
-        blobStorages.push(createSQLiteStorage);
-      }
-    }
+    blobEngine = createLocalBlobEngine(id);
     providerCreators.push(...createLocalProviders());
   } else if (flavour === WorkspaceFlavour.AFFINE_PUBLIC) {
-    if (isBrowser) {
-      blobStorages.push(createIndexeddbStorage);
-      if (environment.isDesktop && runtimeConfig.enableSQLiteProvider) {
-        blobStorages.push(createSQLiteStorage);
-      }
-    }
-    blobStorages.push(createCloudBlobStorage);
+    blobEngine = createAffinePublicBlobEngine(id);
     providerCreators.push(...createAffinePublicProviders());
   } else {
     throw new Error('unsupported flavour');
   }
-  blobStorages.push(createStaticStorage);
 
   const workspace = new Workspace({
     id,
     isSSR: !isBrowser,
     providerCreators: typeof window === 'undefined' ? [] : providerCreators,
-    blobStorages: blobStorages,
+    blobStorages: [
+      () => ({
+        crud: {
+          async get(key) {
+            return (await blobEngine.get(key)) ?? null;
+          },
+          async set(key, value) {
+            await blobEngine.set(key, value);
+            return key;
+          },
+          async delete(key) {
+            return blobEngine.delete(key);
+          },
+          async list() {
+            return blobEngine.list();
+          },
+        },
+      }),
+    ],
     idGenerator: () => nanoid(),
     schema: globalBlockSuiteSchema,
   });
+  workspaceBlobEngineWeakMap.set(workspace, blobEngine);
   createMonitor(workspace.doc);
   setEditorFlags(workspace);
   INTERNAL_BLOCKSUITE_HASH_MAP.set(id, workspace);

packages/frontend/workspace/src/providers/__tests__/socketio-provider.spec.ts

@@ -1,103 +0,0 @@
-/**
- * @vitest-environment happy-dom
- */
-import 'fake-indexeddb/auto';
-
-import type { AffineSocketIOProvider } from '@affine/env/workspace';
-import { __unstableSchemas, AffineSchemas } from '@blocksuite/blocks/models';
-import { Schema, Workspace } from '@blocksuite/store';
-import { describe, expect, test } from 'vitest';
-import * as awarenessProtocol from 'y-protocols/awareness';
-import { Doc } from 'yjs';
-
-import { createAffineSocketIOProvider } from '..';
-
-const schema = new Schema();
-
-schema.register(AffineSchemas).register(__unstableSchemas);
-
-describe('sockio provider', () => {
-  test.skip('test storage', async () => {
-    const workspaceId = 'test-storage-ws';
-    {
-      const workspace = new Workspace({
-        id: workspaceId,
-        isSSR: true,
-        schema,
-      });
-      const provider = createAffineSocketIOProvider(
-        workspace.id,
-        workspace.doc,
-        {
-          awareness: workspace.awarenessStore.awareness,
-        }
-      ) as AffineSocketIOProvider;
-      provider.connect();
-      const page = workspace.createPage({
-        id: 'page',
-      });
-
-      await page.waitForLoaded();
-      page.addBlock('affine:page', {
-        title: new page.Text('123123'),
-      });
-
-      await new Promise(resolve => setTimeout(resolve, 1000));
-    }
-
-    {
-      const workspace = new Workspace({
-        id: workspaceId,
-        isSSR: true,
-        schema,
-      });
-      const provider = createAffineSocketIOProvider(
-        workspace.id,
-        workspace.doc,
-        {
-          awareness: workspace.awarenessStore.awareness,
-        }
-      ) as AffineSocketIOProvider;
-
-      provider.connect();
-
-      await new Promise(resolve => setTimeout(resolve, 1000));
-      const page = workspace.getPage('page')!;
-      await page.waitForLoaded();
-      const block = page.getBlockByFlavour('affine:page');
-      expect(block[0].flavour).toEqual('affine:page');
-    }
-  });
-
-  test.skip('test collaboration', async () => {
-    const workspaceId = 'test-collboration-ws';
-    {
-      const doc = new Doc({ guid: workspaceId });
-      const provider = createAffineSocketIOProvider(doc.guid, doc, {
-        awareness: new awarenessProtocol.Awareness(doc),
-      }) as AffineSocketIOProvider;
-
-      const doc2 = new Doc({ guid: workspaceId });
-      const provider2 = createAffineSocketIOProvider(doc2.guid, doc2, {
-        awareness: new awarenessProtocol.Awareness(doc2),
-      }) as AffineSocketIOProvider;
-
-      provider.connect();
-      provider2.connect();
-
-      await new Promise(resolve => setTimeout(resolve, 500));
-
-      const subdoc = new Doc();
-      const folder = doc.getMap();
-      folder.set('subDoc', subdoc);
-      subdoc.getText().insert(0, 'subDoc content');
-
-      await new Promise(resolve => setTimeout(resolve, 1000));
-
-      expect(
-        (doc2.getMap().get('subDoc') as Doc).getText().toJSON(),
-        'subDoc content'
-      );
-    }
-  });
-});

packages/frontend/workspace/src/providers/__tests__/sqlite-provider.spec.ts

@@ -1,165 +0,0 @@
-import type {
-  SQLiteDBDownloadProvider,
-  SQLiteProvider,
-} from '@affine/env/workspace';
-import { __unstableSchemas, AffineSchemas } from '@blocksuite/blocks/models';
-import type { Y as YType } from '@blocksuite/store';
-import { Schema, Workspace } from '@blocksuite/store';
-import type { DBHandlerManager } from '@toeverything/infra/handler';
-import type {
-  EventMap,
-  UnwrapManagerHandlerToClientSide,
-} from '@toeverything/infra/type';
-import { nanoid } from 'nanoid';
-import { setTimeout } from 'timers/promises';
-import { beforeEach, describe, expect, test, vi } from 'vitest';
-import { getDoc } from 'y-provider';
-
-import {
-  createSQLiteDBDownloadProvider,
-  createSQLiteProvider,
-} from '../sqlite-providers';
-
-const Y = Workspace.Y;
-
-let id: string;
-let workspace: Workspace;
-let provider: SQLiteProvider;
-let downloadProvider: SQLiteDBDownloadProvider;
-
-let offlineYdoc: YType.Doc;
-
-let triggerDBUpdate:
-  | Parameters<typeof window.events.db.onExternalUpdate>[0]
-  | null = null;
-
-const mockedAddBlob = vi.fn();
-
-vi.stubGlobal('window', {
-  apis: {
-    db: {
-      getDocAsUpdates: async (_, guid) => {
-        const subdoc = guid ? getDoc(offlineYdoc, guid) : offlineYdoc;
-        if (!subdoc) {
-          return false;
-        }
-        return Y.encodeStateAsUpdate(subdoc);
-      },
-      applyDocUpdate: async (_, update, subdocId) => {
-        const subdoc = subdocId ? getDoc(offlineYdoc, subdocId) : offlineYdoc;
-        if (!subdoc) {
-          return;
-        }
-        Y.applyUpdate(subdoc, update, 'sqlite');
-      },
-      getBlobKeys: async () => {
-        // todo: may need to hack the way to get hash keys of blobs
-        return [];
-      },
-      addBlob: mockedAddBlob,
-    } satisfies Partial<UnwrapManagerHandlerToClientSide<DBHandlerManager>>,
-  },
-  events: {
-    db: {
-      onExternalUpdate: fn => {
-        triggerDBUpdate = fn;
-        return () => {
-          triggerDBUpdate = null;
-        };
-      },
-    },
-  } as Partial<EventMap>,
-});
-
-vi.stubGlobal('environment', {
-  isDesktop: true,
-});
-
-const schema = new Schema();
-
-schema.register(AffineSchemas).register(__unstableSchemas);
-
-beforeEach(() => {
-  id = nanoid();
-  workspace = new Workspace({
-    id,
-    isSSR: true,
-    schema,
-  });
-  provider = createSQLiteProvider(workspace.id, workspace.doc, {
-    awareness: workspace.awarenessStore.awareness,
-  }) as SQLiteProvider;
-  downloadProvider = createSQLiteDBDownloadProvider(
-    workspace.id,
-    workspace.doc,
-    {
-      awareness: workspace.awarenessStore.awareness,
-    }
-  ) as SQLiteDBDownloadProvider;
-  offlineYdoc = new Y.Doc();
-  offlineYdoc.getText('text').insert(0, 'sqlite-hello');
-});
-
-describe('SQLite download provider', () => {
-  test('sync updates', async () => {
-    // on connect, the updates from sqlite should be sync'ed to the existing ydoc
-    workspace.doc.getText('text').insert(0, 'mem-hello');
-
-    downloadProvider.sync();
-    await downloadProvider.whenReady;
-
-    // depending on the nature of the sync, the data can be sync'ed in either direction
-    const options = ['sqlite-hellomem-hello', 'mem-hellosqlite-hello'];
-    const synced = options.filter(
-      o => o === workspace.doc.getText('text').toString()
-    );
-    expect(synced.length).toBe(1);
-  });
-
-  // there is no updates from sqlite for now
-  test.skip('on db update', async () => {
-    provider.connect();
-
-    await setTimeout(200);
-
-    offlineYdoc.getText('text').insert(0, 'sqlite-world');
-
-    triggerDBUpdate?.({
-      workspaceId: id + '-another-id',
-      update: Y.encodeStateAsUpdate(offlineYdoc),
-    });
-
-    // not yet updated (because the workspace id is different)
-    expect(workspace.doc.getText('text').toString()).toBe('');
-
-    triggerDBUpdate?.({
-      workspaceId: id,
-      update: Y.encodeStateAsUpdate(offlineYdoc),
-    });
-
-    expect(workspace.doc.getText('text').toString()).toBe(
-      'sqlite-worldsqlite-hello'
-    );
-  });
-
-  test('disconnect handlers', async () => {
-    const offHandler = vi.fn();
-    let handleUpdate = () => {};
-    let handleSubdocs = () => {};
-    workspace.doc.on = (event: string, fn: () => void) => {
-      if (event === 'update') {
-        handleUpdate = fn;
-      } else if (event === 'subdocs') {
-        handleSubdocs = fn;
-      }
-    };
-    workspace.doc.off = offHandler;
-    provider.connect();
-
-    provider.disconnect();
-
-    expect(triggerDBUpdate).toBe(null);
-    expect(offHandler).toBeCalledWith('update', handleUpdate);
-    expect(offHandler).toBeCalledWith('subdocs', handleSubdocs);
-  });
-});

packages/frontend/workspace/src/providers/awareness/affine/index.ts

@@ -0,0 +1,101 @@
+import { DebugLogger } from '@affine/debug';
+import {
+  applyAwarenessUpdate,
+  type Awareness,
+  encodeAwarenessUpdate,
+  removeAwarenessStates,
+} from 'y-protocols/awareness';
+
+import { getIoManager } from '../../utils/affine-io';
+import { base64ToUint8Array, uint8ArrayToBase64 } from '../../utils/base64';
+import type { AwarenessProvider } from '..';
+
+const logger = new DebugLogger('affine:awareness:socketio');
+
+export type AwarenessChanges = Record<
+  'added' | 'updated' | 'removed',
+  number[]
+>;
+
+export function createAffineAwarenessProvider(
+  workspaceId: string,
+  awareness: Awareness
+): AwarenessProvider {
+  const socket = getIoManager().socket('/');
+
+  const awarenessBroadcast = ({
+    workspaceId,
+    awarenessUpdate,
+  }: {
+    workspaceId: string;
+    awarenessUpdate: string;
+  }) => {
+    if (workspaceId !== workspaceId) {
+      return;
+    }
+    applyAwarenessUpdate(
+      awareness,
+      base64ToUint8Array(awarenessUpdate),
+      'remote'
+    );
+  };
+
+  const awarenessUpdate = (changes: AwarenessChanges, origin: unknown) => {
+    if (origin === 'remote') {
+      return;
+    }
+
+    const changedClients = Object.values(changes).reduce((res, cur) => [
+      ...res,
+      ...cur,
+    ]);
+
+    const update = encodeAwarenessUpdate(awareness, changedClients);
+    uint8ArrayToBase64(update)
+      .then(encodedUpdate => {
+        socket.emit('awareness-update', {
+          workspaceId: workspaceId,
+          awarenessUpdate: encodedUpdate,
+        });
+      })
+      .catch(err => logger.error(err));
+  };
+
+  const newClientAwarenessInitHandler = () => {
+    const awarenessUpdate = encodeAwarenessUpdate(awareness, [
+      awareness.clientID,
+    ]);
+    uint8ArrayToBase64(awarenessUpdate)
+      .then(encodedAwarenessUpdate => {
+        socket.emit('awareness-update', {
+          guid: workspaceId,
+          awarenessUpdate: encodedAwarenessUpdate,
+        });
+      })
+      .catch(err => logger.error(err));
+  };
+
+  const windowBeforeUnloadHandler = () => {
+    removeAwarenessStates(awareness, [awareness.clientID], 'window unload');
+  };
+
+  return {
+    connect: () => {
+      socket.on('server-awareness-broadcast', awarenessBroadcast);
+      socket.on('new-client-awareness-init', newClientAwarenessInitHandler);
+      awareness.on('update', awarenessUpdate);
+
+      window.addEventListener('beforeunload', windowBeforeUnloadHandler);
+
+      socket.emit('awareness-init', workspaceId);
+      socket.connect();
+    },
+    disconnect: () => {
+      awareness.off('update', awarenessUpdate);
+      socket.off('server-awareness-broadcast', awarenessBroadcast);
+      socket.off('new-client-awareness-init', newClientAwarenessInitHandler);
+      window.removeEventListener('unload', windowBeforeUnloadHandler);
+      socket.disconnect();
+    },
+  };
+}

packages/frontend/workspace/src/providers/awareness/broadcast-channel/index.ts

@@ -0,0 +1,63 @@
+import type { Awareness } from 'y-protocols/awareness.js';
+import {
+  applyAwarenessUpdate,
+  encodeAwarenessUpdate,
+} from 'y-protocols/awareness.js';
+
+import type { AwarenessProvider } from '..';
+import type { AwarenessChanges } from '../affine';
+
+type ChannelMessage =
+  | { type: 'connect' }
+  | { type: 'update'; update: Uint8Array };
+
+export function createBroadcastChannelAwarenessProvider(
+  workspaceId: string,
+  awareness: Awareness
+): AwarenessProvider {
+  const channel = new BroadcastChannel('awareness:' + workspaceId);
+
+  function handleAwarenessUpdate(changes: AwarenessChanges, origin: unknown) {
+    if (origin === 'remote') {
+      return;
+    }
+
+    const changedClients = Object.values(changes).reduce((res, cur) => [
+      ...res,
+      ...cur,
+    ]);
+
+    const update = encodeAwarenessUpdate(awareness, changedClients);
+    channel.postMessage({
+      type: 'update',
+      update: update,
+    } satisfies ChannelMessage);
+  }
+
+  function handleChannelMessage(event: MessageEvent<ChannelMessage>) {
+    if (event.data.type === 'update') {
+      const update = event.data.update;
+      applyAwarenessUpdate(awareness, update, 'remote');
+    }
+    if (event.data.type === 'connect') {
+      channel.postMessage({
+        type: 'update',
+        update: encodeAwarenessUpdate(awareness, [awareness.clientID]),
+      } satisfies ChannelMessage);
+    }
+  }
+
+  return {
+    connect() {
+      channel.postMessage({
+        type: 'connect',
+      } satisfies ChannelMessage);
+      awareness.on('update', handleAwarenessUpdate);
+      channel.addEventListener('message', handleChannelMessage);
+    },
+    disconnect() {
+      awareness.off('update', handleAwarenessUpdate);
+      channel.removeEventListener('message', handleChannelMessage);
+    },
+  };
+}

packages/frontend/workspace/src/providers/awareness/index.ts

@@ -0,0 +1,7 @@
+export interface AwarenessProvider {
+  connect(): void;
+  disconnect(): void;
+}
+
+export * from './affine';
+export * from './broadcast-channel';

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

@@ -1,98 +0,0 @@
-import { DebugLogger } from '@affine/debug';
-import { fetchWithTraceReport } from '@affine/graphql';
-import type { ActiveDocProvider, DocProviderCreator } from '@blocksuite/store';
-import { Workspace } from '@blocksuite/store';
-import type { Doc } from 'yjs';
-
-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 fetchWithTraceReport(
-    runtimeConfig.serverUrlPrefix +
-      `/api/workspaces/${rootGuid}/docs/${pageGuid}`,
-    {
-      priority: 'high',
-    }
-  );
-  if (response.ok) {
-    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 (typeof buffer !== 'boolean') {
-    Y.applyUpdate(doc, new Uint8Array(buffer), 'affine-cloud');
-  }
-}
-
-export const createCloudDownloadProvider: DocProviderCreator = (
-  id,
-  doc
-): ActiveDocProvider => {
-  let _resolve: () => void;
-  let _reject: (error: unknown) => void;
-  const promise = new Promise<void>((resolve, reject) => {
-    _resolve = resolve;
-    _reject = reject;
-  });
-
-  return {
-    flavour: 'affine-cloud-download',
-    active: true,
-    sync() {
-      downloadBinary(id, doc)
-        .then(() => {
-          logger.info(`Downloaded ${id}`);
-          _resolve();
-        })
-        .catch(_reject);
-    },
-    get whenReady() {
-      return promise;
-    },
-  };
-};
-
-export const createMergeCloudSnapshotProvider: DocProviderCreator = (
-  id,
-  doc
-): ActiveDocProvider => {
-  let _resolve: () => void;
-  const promise = new Promise<void>(resolve => {
-    _resolve = resolve;
-  });
-
-  return {
-    flavour: 'affine-cloud-merge-snapshot',
-    active: true,
-    sync() {
-      downloadBinary(id, doc)
-        .then(() => {
-          logger.info(`Downloaded ${id}`);
-          _resolve();
-        })
-        // ignore error
-        .catch(e => {
-          console.error(e);
-          _resolve();
-        });
-    },
-    get whenReady() {
-      return promise;
-    },
-  };
-};

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

@@ -1,160 +1,142 @@
-import { DebugLogger } from '@affine/debug';
-import type {
-  AffineSocketIOProvider,
-  LocalIndexedDBBackgroundProvider,
-  LocalIndexedDBDownloadProvider,
-} from '@affine/env/workspace';
-import { assertExists } from '@blocksuite/global/utils';
+/**
+ * The `Provider` is responsible for sync `Y.Doc` with the local database and the Affine Cloud, serving as the source of
+ * Affine's local-first collaborative magic.
+ *
+ * When Affine boot, the `Provider` is tasked with reading content from the local database and loading it into the
+ * workspace, continuously storing any changes made by the user into the local database.
+ *
+ * When using Affine Cloud, the `Provider` also handles sync content with the Cloud.
+ *
+ * Additionally, the `Provider` is responsible for implementing a local-first capability, allowing users to edit offline
+ * with changes stored in the local database and sync with the Cloud when the network is restored.
+ */
+
 import type { DocProviderCreator } from '@blocksuite/store';
-import { Workspace } from '@blocksuite/store';
-import { createBroadcastChannelProvider } from '@blocksuite/store/providers/broadcast-channel';
+
 import {
-  createIndexedDBDatasource,
-  createIndexedDBProvider as create,
-} from '@toeverything/y-indexeddb';
-import { createLazyProvider } from 'y-provider';
-import { encodeStateVector } from 'yjs';
+  createAffineAwarenessProvider,
+  createBroadcastChannelAwarenessProvider,
+} from './awareness';
+import { createAffineStorage } from './storage/affine';
+import { createIndexedDBStorage } from './storage/indexeddb';
+import { createSQLiteStorage } from './storage/sqlite';
+import { SyncEngine } from './sync';
 
-import { createAffineDataSource } from '../affine';
-import {
-  createCloudDownloadProvider,
-  createMergeCloudSnapshotProvider,
-  downloadBinaryFromCloud,
-} from './cloud';
-import {
-  createSQLiteDBDownloadProvider,
-  createSQLiteProvider,
-} from './sqlite-providers';
-
-const Y = Workspace.Y;
-const logger = new DebugLogger('indexeddb-provider');
-
-const createAffineSocketIOProvider: DocProviderCreator = (
-  id,
-  doc,
-  { awareness }
-): AffineSocketIOProvider => {
-  const dataSource = createAffineDataSource(id, doc, awareness);
-  const lazyProvider = createLazyProvider(doc, dataSource, {
-    origin: 'affine-socket-io',
-  });
-
-  Object.assign(lazyProvider, { flavour: 'affine-socket-io' });
-
-  return lazyProvider as unknown as AffineSocketIOProvider;
-};
-
-const createIndexedDBBackgroundProvider: DocProviderCreator = (
-  id,
-  blockSuiteWorkspace
-): LocalIndexedDBBackgroundProvider => {
-  const indexeddbProvider = create(blockSuiteWorkspace);
-
-  let connected = false;
-  return {
-    flavour: 'local-indexeddb-background',
-    datasource: indexeddbProvider.datasource,
-    passive: true,
-    get status() {
-      return indexeddbProvider.status;
-    },
-    subscribeStatusChange: indexeddbProvider.subscribeStatusChange,
-    get connected() {
-      return connected;
-    },
-    cleanup: () => {
-      indexeddbProvider.cleanup().catch(console.error);
-    },
-    connect: () => {
-      logger.info('connect indexeddb provider', id);
-      indexeddbProvider.connect();
-    },
-    disconnect: () => {
-      assertExists(indexeddbProvider);
-      logger.info('disconnect indexeddb provider', id);
-      indexeddbProvider.disconnect();
-      connected = false;
-    },
-  };
-};
-
-const indexedDBDownloadOrigin = 'indexeddb-download-provider';
-
-const createIndexedDBDownloadProvider: DocProviderCreator = (
-  id,
-  doc
-): LocalIndexedDBDownloadProvider => {
-  const datasource = createIndexedDBDatasource({});
-  let _resolve: () => void;
-  let _reject: (error: unknown) => void;
-  const promise = new Promise<void>((resolve, reject) => {
-    _resolve = resolve;
-    _reject = reject;
-  });
-
-  return {
-    flavour: 'local-indexeddb',
-    active: true,
-    get whenReady() {
-      return promise;
-    },
-    cleanup: () => {
-      // todo: cleanup data
-    },
-    sync: () => {
-      logger.info('sync indexeddb provider', id);
-      datasource
-        .queryDocState(doc.guid, {
-          stateVector: encodeStateVector(doc),
-        })
-        .then(docState => {
-          if (docState) {
-            Y.applyUpdate(doc, docState.missing, indexedDBDownloadOrigin);
-          }
-          _resolve();
-        })
-        .catch(_reject);
-    },
-  };
-};
-
-export {
-  createAffineSocketIOProvider,
-  createBroadcastChannelProvider,
-  createIndexedDBBackgroundProvider,
-  createIndexedDBDownloadProvider,
-  createSQLiteDBDownloadProvider,
-  createSQLiteProvider,
-  downloadBinaryFromCloud,
-};
+export * from './sync';
 
 export const createLocalProviders = (): DocProviderCreator[] => {
-  const providers = [
-    createIndexedDBBackgroundProvider,
-    createIndexedDBDownloadProvider,
-  ] as DocProviderCreator[];
+  return [
+    (_, doc, { awareness }) => {
+      const engine = new SyncEngine(
+        doc,
+        environment.isDesktop
+          ? createSQLiteStorage(doc.guid)
+          : createIndexedDBStorage(doc.guid),
+        []
+      );
 
-  if (runtimeConfig.enableBroadcastChannelProvider) {
-    providers.push(createBroadcastChannelProvider);
-  }
+      const awarenessProviders = [
+        createBroadcastChannelAwarenessProvider(doc.guid, awareness),
+      ];
 
-  if (environment.isDesktop && runtimeConfig.enableSQLiteProvider) {
-    providers.push(createSQLiteProvider, createSQLiteDBDownloadProvider);
-  }
+      let connected = false;
 
-  return providers;
+      return {
+        flavour: '_',
+        passive: true,
+        active: true,
+        sync() {
+          if (!connected) {
+            engine.start();
+
+            for (const provider of awarenessProviders) {
+              provider.connect();
+            }
+            connected = true;
+          }
+        },
+        get whenReady() {
+          return engine.waitForLoadedRootDoc();
+        },
+        connect() {
+          if (!connected) {
+            engine.start();
+
+            for (const provider of awarenessProviders) {
+              provider.connect();
+            }
+            connected = true;
+          }
+        },
+        disconnect() {
+          // TODO: actually disconnect
+        },
+        get connected() {
+          return connected;
+        },
+        engine,
+      };
+    },
+  ];
 };
 
 export const createAffineProviders = (): DocProviderCreator[] => {
-  return (
-    [
-      ...createLocalProviders(),
-      runtimeConfig.enableCloud && createAffineSocketIOProvider,
-      runtimeConfig.enableCloud && createMergeCloudSnapshotProvider,
-    ] as DocProviderCreator[]
-  ).filter(v => Boolean(v));
+  return [
+    (_, doc, { awareness }) => {
+      const engine = new SyncEngine(
+        doc,
+        environment.isDesktop
+          ? createSQLiteStorage(doc.guid)
+          : createIndexedDBStorage(doc.guid),
+        [createAffineStorage(doc.guid)]
+      );
+
+      const awarenessProviders = [
+        createBroadcastChannelAwarenessProvider(doc.guid, awareness),
+        createAffineAwarenessProvider(doc.guid, awareness),
+      ];
+
+      let connected = false;
+
+      return {
+        flavour: '_',
+        passive: true,
+        active: true,
+        sync() {
+          if (!connected) {
+            engine.start();
+
+            for (const provider of awarenessProviders) {
+              provider.connect();
+            }
+            connected = true;
+          }
+        },
+        get whenReady() {
+          return engine.waitForLoadedRootDoc();
+        },
+        connect() {
+          if (!connected) {
+            engine.start();
+
+            for (const provider of awarenessProviders) {
+              provider.connect();
+            }
+            connected = true;
+          }
+        },
+        disconnect() {
+          // TODO: actually disconnect
+        },
+        get connected() {
+          return connected;
+        },
+        engine,
+      };
+    },
+  ];
 };
 
 export const createAffinePublicProviders = (): DocProviderCreator[] => {
-  return [createCloudDownloadProvider];
+  return [];
 };

packages/frontend/workspace/src/providers/logger.ts

@@ -1,3 +0,0 @@
-import { DebugLogger } from '@affine/debug';
-
-export const localProviderLogger = new DebugLogger('local-provider');

packages/frontend/workspace/src/providers/sqlite-providers.ts

@@ -1,133 +0,0 @@
-import type {
-  SQLiteDBDownloadProvider,
-  SQLiteProvider,
-} from '@affine/env/workspace';
-import { assertExists } from '@blocksuite/global/utils';
-import type { DocProviderCreator } from '@blocksuite/store';
-import { Workspace as BlockSuiteWorkspace } from '@blocksuite/store';
-import { createLazyProvider, type DocDataSource } from 'y-provider';
-import type { Doc } from 'yjs';
-
-import { localProviderLogger as logger } from './logger';
-
-const Y = BlockSuiteWorkspace.Y;
-
-const sqliteOrigin = 'sqlite-provider-origin';
-
-const createDatasource = (workspaceId: string): DocDataSource => {
-  if (!window.apis?.db) {
-    throw new Error('sqlite datasource is not available');
-  }
-
-  return {
-    queryDocState: async guid => {
-      const update = await window.apis.db.getDocAsUpdates(
-        workspaceId,
-        workspaceId === guid ? undefined : guid
-      );
-
-      if (update) {
-        return {
-          missing: update,
-        };
-      }
-
-      return false;
-    },
-    sendDocUpdate: async (guid, update) => {
-      return window.apis.db.applyDocUpdate(
-        workspaceId,
-        update,
-        workspaceId === guid ? undefined : guid
-      );
-    },
-  };
-};
-
-/**
- * A provider that is responsible for syncing updates the workspace with the local SQLite database.
- */
-export const createSQLiteProvider: DocProviderCreator = (
-  id,
-  rootDoc
-): SQLiteProvider => {
-  const datasource = createDatasource(id);
-  let provider: ReturnType<typeof createLazyProvider> | null = null;
-  let connected = false;
-  return {
-    flavour: 'sqlite',
-    datasource,
-    passive: true,
-    get status() {
-      assertExists(provider);
-      return provider.status;
-    },
-    subscribeStatusChange(onStatusChange) {
-      assertExists(provider);
-      return provider.subscribeStatusChange(onStatusChange);
-    },
-    connect: () => {
-      provider = createLazyProvider(rootDoc, datasource, { origin: 'sqlite' });
-      provider.connect();
-      connected = true;
-    },
-    disconnect: () => {
-      provider?.disconnect();
-      provider = null;
-      connected = false;
-    },
-    get connected() {
-      return connected;
-    },
-  };
-};
-
-/**
- * A provider that is responsible for DOWNLOADING updates from the local SQLite database.
- */
-export const createSQLiteDBDownloadProvider: DocProviderCreator = (
-  id,
-  rootDoc
-): SQLiteDBDownloadProvider => {
-  const { apis } = window;
-
-  let _resolve: () => void;
-  let _reject: (error: unknown) => void;
-  const promise = new Promise<void>((resolve, reject) => {
-    _resolve = resolve;
-    _reject = reject;
-  });
-
-  async function syncUpdates(doc: Doc) {
-    logger.info('syncing updates from sqlite', doc.guid);
-    const subdocId = doc.guid === id ? undefined : doc.guid;
-    const updates = await apis.db.getDocAsUpdates(id, subdocId);
-
-    if (updates) {
-      Y.applyUpdate(doc, updates, sqliteOrigin);
-    }
-
-    return true;
-  }
-
-  return {
-    flavour: 'sqlite-download',
-    active: true,
-    get whenReady() {
-      return promise;
-    },
-    cleanup: () => {
-      // todo
-    },
-    sync: () => {
-      logger.info('connect sqlite download provider', id);
-      syncUpdates(rootDoc)
-        .then(() => {
-          _resolve();
-        })
-        .catch(error => {
-          _reject(error);
-        });
-    },
-  };
-};

packages/frontend/workspace/src/providers/storage/affine/index.ts

@@ -0,0 +1,162 @@
+import { DebugLogger } from '@affine/debug';
+
+import { getIoManager } from '../../utils/affine-io';
+import { base64ToUint8Array, uint8ArrayToBase64 } from '../../utils/base64';
+import type { Storage } from '..';
+import { MultipleBatchSyncSender } from './batch-sync-sender';
+
+const logger = new DebugLogger('affine:storage:socketio');
+
+export function createAffineStorage(
+  workspaceId: string
+): Storage & { disconnect: () => void } {
+  logger.debug('createAffineStorage', workspaceId);
+  const socket = getIoManager().socket('/');
+
+  const syncSender = new MultipleBatchSyncSender(async (guid, updates) => {
+    const payload = await Promise.all(
+      updates.map(update => uint8ArrayToBase64(update))
+    );
+
+    return new Promise(resolve => {
+      socket.emit(
+        'client-update-v2',
+        {
+          workspaceId,
+          guid,
+          updates: payload,
+        },
+        (response: {
+          // TODO: reuse `EventError` with server
+          error?: any;
+          data: any;
+        }) => {
+          // TODO: raise error with different code to users
+          if (response.error) {
+            logger.error('client-update-v2 error', {
+              workspaceId,
+              guid,
+              response,
+            });
+          }
+
+          resolve({
+            accepted: !response.error,
+            // TODO: reuse `EventError` with server
+            retry: response.error?.code === 'INTERNAL',
+          });
+        }
+      );
+    });
+  });
+
+  // TODO: handle error
+  socket.on('connect', () => {
+    socket.emit(
+      'client-handshake',
+      workspaceId,
+      (response: { error?: any }) => {
+        if (!response.error) {
+          syncSender.start();
+        }
+      }
+    );
+  });
+
+  socket.connect();
+
+  return {
+    name: 'socketio',
+    async pull(docId, state) {
+      const stateVector = state ? await uint8ArrayToBase64(state) : undefined;
+
+      return new Promise((resolve, reject) => {
+        logger.debug('doc-load-v2', {
+          workspaceId: workspaceId,
+          guid: docId,
+          stateVector,
+        });
+        socket.emit(
+          'doc-load-v2',
+          {
+            workspaceId: workspaceId,
+            guid: docId,
+            stateVector,
+          },
+          (
+            response: // TODO: reuse `EventError` with server
+            { error: any } | { data: { missing: string; state: string } }
+          ) => {
+            logger.debug('doc-load callback', {
+              workspaceId: workspaceId,
+              guid: docId,
+              stateVector,
+              response,
+            });
+
+            if ('error' in response) {
+              // TODO: result `EventError` with server
+              if (response.error.code === 'DOC_NOT_FOUND') {
+                resolve(null);
+              } else {
+                reject(new Error(response.error.message));
+              }
+            } else {
+              resolve({
+                data: base64ToUint8Array(response.data.missing),
+                state: response.data.state
+                  ? base64ToUint8Array(response.data.state)
+                  : undefined,
+              });
+            }
+          }
+        );
+      });
+    },
+    async push(docId, update) {
+      logger.debug('client-update-v2', {
+        workspaceId,
+        guid: docId,
+        update,
+      });
+
+      await syncSender.send(docId, update);
+    },
+    async subscribe(cb, disconnect) {
+      const response: { error?: any } = await socket
+        .timeout(10000)
+        .emitWithAck('client-handshake', workspaceId);
+
+      if (response.error) {
+        throw new Error('client-handshake error, ' + response.error);
+      }
+
+      const handleUpdate = async (message: {
+        workspaceId: string;
+        guid: string;
+        updates: string[];
+      }) => {
+        if (message.workspaceId === workspaceId) {
+          message.updates.forEach(update => {
+            cb(message.guid, base64ToUint8Array(update));
+          });
+        }
+      };
+      socket.on('server-updates', handleUpdate);
+
+      socket.on('disconnect', reason => {
+        socket.off('server-updates', handleUpdate);
+        disconnect(reason);
+      });
+
+      return () => {
+        socket.off('server-updates', handleUpdate);
+      };
+    },
+    disconnect() {
+      syncSender.stop();
+      socket.emit('client-leave', workspaceId);
+      socket.disconnect();
+    },
+  };
+}

packages/frontend/workspace/src/providers/storage/index.ts

@@ -0,0 +1,29 @@
+export interface Storage {
+  /**
+   * for debug
+   */
+  name: string;
+
+  pull(
+    docId: string,
+    state: Uint8Array
+  ): Promise<{ data: Uint8Array; state?: Uint8Array } | null>;
+  push(docId: string, data: Uint8Array): Promise<void>;
+
+  /**
+   * Subscribe to updates from peer
+   *
+   * @param cb callback to handle updates
+   * @param disconnect callback to handle disconnect, reason can be something like 'network-error'
+   *
+   * @returns unsubscribe function
+   */
+  subscribe(
+    cb: (docId: string, data: Uint8Array) => void,
+    disconnect: (reason: string) => void
+  ): Promise<() => void>;
+}
+
+export * from './affine';
+export * from './indexeddb';
+export * from './sqlite';

packages/frontend/workspace/src/providers/storage/indexeddb/index.ts

@@ -0,0 +1,139 @@
+import { type DBSchema, type IDBPDatabase, openDB } from 'idb';
+import {
+  applyUpdate,
+  diffUpdate,
+  Doc,
+  encodeStateAsUpdate,
+  encodeStateVectorFromUpdate,
+} from 'yjs';
+
+import type { Storage } from '..';
+
+export const dbVersion = 1;
+export const DEFAULT_DB_NAME = 'affine-local';
+
+export function mergeUpdates(updates: Uint8Array[]) {
+  if (updates.length === 0) {
+    return new Uint8Array();
+  }
+  if (updates.length === 1) {
+    return updates[0];
+  }
+  const doc = new Doc();
+  doc.transact(() => {
+    updates.forEach(update => {
+      applyUpdate(doc, update);
+    });
+  });
+  return encodeStateAsUpdate(doc);
+}
+
+type UpdateMessage = {
+  timestamp: number;
+  update: Uint8Array;
+};
+
+type WorkspacePersist = {
+  id: string;
+  updates: UpdateMessage[];
+};
+
+interface BlockSuiteBinaryDB extends DBSchema {
+  workspace: {
+    key: string;
+    value: WorkspacePersist;
+  };
+  milestone: {
+    key: string;
+    value: unknown;
+  };
+}
+
+export function upgradeDB(db: IDBPDatabase<BlockSuiteBinaryDB>) {
+  db.createObjectStore('workspace', { keyPath: 'id' });
+  db.createObjectStore('milestone', { keyPath: 'id' });
+}
+
+type ChannelMessage = {
+  type: 'db-updated';
+  payload: { docId: string; update: Uint8Array };
+};
+
+export function createIndexedDBStorage(
+  workspaceId: string,
+  dbName = DEFAULT_DB_NAME,
+  mergeCount = 1
+): Storage {
+  let dbPromise: Promise<IDBPDatabase<BlockSuiteBinaryDB>> | null = null;
+  const getDb = async () => {
+    if (dbPromise === null) {
+      dbPromise = openDB<BlockSuiteBinaryDB>(dbName, dbVersion, {
+        upgrade: upgradeDB,
+      });
+    }
+    return dbPromise;
+  };
+
+  // indexeddb could be shared between tabs, so we use broadcast channel to notify other tabs
+  const channel = new BroadcastChannel('indexeddb:' + workspaceId);
+
+  return {
+    name: 'indexeddb',
+    async pull(docId, state) {
+      const db = await getDb();
+      const store = db
+        .transaction('workspace', 'readonly')
+        .objectStore('workspace');
+      const data = await store.get(docId);
+
+      if (!data) {
+        return null;
+      }
+
+      const { updates } = data;
+      const update = mergeUpdates(updates.map(({ update }) => update));
+
+      const diff = state ? diffUpdate(update, state) : update;
+
+      return { data: diff, state: encodeStateVectorFromUpdate(update) };
+    },
+    async push(docId, update) {
+      const db = await getDb();
+      const store = db
+        .transaction('workspace', 'readwrite')
+        .objectStore('workspace');
+
+      // TODO: maybe we do not need to get data every time
+      const { updates } = (await store.get(docId)) ?? { updates: [] };
+      let rows: UpdateMessage[] = [
+        ...updates,
+        { timestamp: Date.now(), update },
+      ];
+      if (mergeCount && rows.length >= mergeCount) {
+        const merged = mergeUpdates(rows.map(({ update }) => update));
+        rows = [{ timestamp: Date.now(), update: merged }];
+      }
+      await store.put({
+        id: docId,
+        updates: rows,
+      });
+      channel.postMessage({
+        type: 'db-updated',
+        payload: { docId, update },
+      } satisfies ChannelMessage);
+    },
+    async subscribe(cb, _disconnect) {
+      function onMessage(event: MessageEvent<ChannelMessage>) {
+        const { type, payload } = event.data;
+        if (type === 'db-updated') {
+          const { docId, update } = payload;
+          cb(docId, update);
+        }
+      }
+      channel.addEventListener('message', onMessage);
+      return () => {
+        channel.removeEventListener('message', onMessage);
+      };
+    },
+  };
+}

packages/frontend/workspace/src/providers/storage/sqlite/index.ts

@@ -0,0 +1,38 @@
+import { encodeStateVectorFromUpdate } from 'yjs';
+
+import type { Storage } from '..';
+
+export function createSQLiteStorage(workspaceId: string): Storage {
+  if (!window.apis?.db) {
+    throw new Error('sqlite datasource is not available');
+  }
+
+  return {
+    name: 'sqlite',
+    async pull(docId, _state) {
+      const update = await window.apis.db.getDocAsUpdates(
+        workspaceId,
+        workspaceId === docId ? undefined : docId
+      );
+
+      if (update) {
+        return {
+          data: update,
+          state: encodeStateVectorFromUpdate(update),
+        };
+      }
+
+      return null;
+    },
+    async push(docId, data) {
+      return window.apis.db.applyDocUpdate(
+        workspaceId,
+        data,
+        workspaceId === docId ? undefined : docId
+      );
+    },
+    async subscribe(_cb, _disconnect) {
+      return () => {};
+    },
+  };
+}

packages/frontend/workspace/src/providers/sync/__tests__/engine.spec.ts

@@ -0,0 +1,172 @@
+import 'fake-indexeddb/auto';
+
+import { setTimeout } from 'node:timers/promises';
+
+import { __unstableSchemas, AffineSchemas } from '@blocksuite/blocks/models';
+import { Schema, Workspace } from '@blocksuite/store';
+import { beforeEach, describe, expect, test, vi } from 'vitest';
+import { Doc } from 'yjs';
+
+import { createIndexedDBStorage } from '../../storage';
+import { SyncEngine, SyncEngineStep, SyncPeerStep } from '../';
+import { createTestStorage } from './test-storage';
+
+const schema = new Schema();
+
+schema.register(AffineSchemas).register(__unstableSchemas);
+
+beforeEach(() => {
+  vi.useFakeTimers({ toFake: ['requestIdleCallback'] });
+});
+
+describe('SyncEngine', () => {
+  test('basic - indexeddb', async () => {
+    let prev: any;
+    {
+      const workspace = new Workspace({
+        id: 'test',
+        isSSR: true,
+        schema,
+      });
+
+      const syncEngine = new SyncEngine(
+        workspace.doc,
+        createIndexedDBStorage(workspace.doc.guid),
+        [
+          createIndexedDBStorage(workspace.doc.guid + '1'),
+          createIndexedDBStorage(workspace.doc.guid + '2'),
+        ]
+      );
+      syncEngine.start();
+
+      const page = workspace.createPage({
+        id: 'page0',
+      });
+      await page.load();
+      const pageBlockId = page.addBlock('affine:page', {
+        title: new page.Text(''),
+      });
+      page.addBlock('affine:surface', {}, pageBlockId);
+      const frameId = page.addBlock('affine:note', {}, pageBlockId);
+      page.addBlock('affine:paragraph', {}, frameId);
+      await syncEngine.waitForSynced();
+      syncEngine.stop();
+      prev = workspace.doc.toJSON();
+    }
+
+    {
+      const workspace = new Workspace({
+        id: 'test',
+        isSSR: true,
+        schema,
+      });
+      const syncEngine = new SyncEngine(
+        workspace.doc,
+        createIndexedDBStorage(workspace.doc.guid),
+        []
+      );
+      syncEngine.start();
+      await syncEngine.waitForSynced();
+      expect(workspace.doc.toJSON()).toEqual({
+        ...prev,
+      });
+      syncEngine.stop();
+    }
+
+    {
+      const workspace = new Workspace({
+        id: 'test',
+        isSSR: true,
+        schema,
+      });
+      const syncEngine = new SyncEngine(
+        workspace.doc,
+        createIndexedDBStorage(workspace.doc.guid + '1'),
+        []
+      );
+      syncEngine.start();
+      await syncEngine.waitForSynced();
+      expect(workspace.doc.toJSON()).toEqual({
+        ...prev,
+      });
+      syncEngine.stop();
+    }
+
+    {
+      const workspace = new Workspace({
+        id: 'test',
+        isSSR: true,
+        schema,
+      });
+      const syncEngine = new SyncEngine(
+        workspace.doc,
+        createIndexedDBStorage(workspace.doc.guid + '2'),
+        []
+      );
+      syncEngine.start();
+      await syncEngine.waitForSynced();
+      expect(workspace.doc.toJSON()).toEqual({
+        ...prev,
+      });
+      syncEngine.stop();
+    }
+  });
+
+  test('status', async () => {
+    const ydoc = new Doc({ guid: 'test - status' });
+
+    const localStorage = createTestStorage(createIndexedDBStorage(ydoc.guid));
+    const remoteStorage = createTestStorage(createIndexedDBStorage(ydoc.guid));
+
+    localStorage.pausePull();
+    localStorage.pausePush();
+    remoteStorage.pausePull();
+    remoteStorage.pausePush();
+
+    const syncEngine = new SyncEngine(ydoc, localStorage, [remoteStorage]);
+    expect(syncEngine.status.step).toEqual(SyncEngineStep.Stopped);
+
+    syncEngine.start();
+    await setTimeout(100);
+
+    expect(syncEngine.status.step).toEqual(SyncEngineStep.Syncing);
+    expect(syncEngine.status.local?.step).toEqual(SyncPeerStep.LoadingRootDoc);
+
+    localStorage.resumePull();
+    await setTimeout(100);
+
+    expect(syncEngine.status.step).toEqual(SyncEngineStep.Syncing);
+    expect(syncEngine.status.local?.step).toEqual(SyncPeerStep.Synced);
+    expect(syncEngine.status.remotes[0]?.step).toEqual(
+      SyncPeerStep.LoadingRootDoc
+    );
+
+    remoteStorage.resumePull();
+    await setTimeout(100);
+
+    expect(syncEngine.status.step).toEqual(SyncEngineStep.Synced);
+    expect(syncEngine.status.local?.step).toEqual(SyncPeerStep.Synced);
+    expect(syncEngine.status.remotes[0]?.step).toEqual(SyncPeerStep.Synced);
+
+    ydoc.getArray('test').insert(0, [1, 2, 3]);
+    await setTimeout(100);
+
+    expect(syncEngine.status.step).toEqual(SyncEngineStep.Syncing);
+    expect(syncEngine.status.local?.step).toEqual(SyncPeerStep.Syncing);
+    expect(syncEngine.status.remotes[0]?.step).toEqual(SyncPeerStep.Syncing);
+
+    localStorage.resumePush();
+    await setTimeout(100);
+
+    expect(syncEngine.status.step).toEqual(SyncEngineStep.Syncing);
+    expect(syncEngine.status.local?.step).toEqual(SyncPeerStep.Synced);
+    expect(syncEngine.status.remotes[0]?.step).toEqual(SyncPeerStep.Syncing);
+
+    remoteStorage.resumePush();
+    await setTimeout(100);
+
+    expect(syncEngine.status.step).toEqual(SyncEngineStep.Synced);
+    expect(syncEngine.status.local?.step).toEqual(SyncPeerStep.Synced);
+    expect(syncEngine.status.remotes[0]?.step).toEqual(SyncPeerStep.Synced);
+  });
+});

packages/frontend/workspace/src/providers/sync/__tests__/peer.spec.ts

@@ -0,0 +1,94 @@
+import 'fake-indexeddb/auto';
+
+import { __unstableSchemas, AffineSchemas } from '@blocksuite/blocks/models';
+import { Schema, Workspace } from '@blocksuite/store';
+import { beforeEach, describe, expect, test, vi } from 'vitest';
+
+import { createIndexedDBStorage } from '../../storage';
+import { SyncPeer, SyncPeerStep } from '../';
+
+const schema = new Schema();
+
+schema.register(AffineSchemas).register(__unstableSchemas);
+
+beforeEach(() => {
+  vi.useFakeTimers({ toFake: ['requestIdleCallback'] });
+});
+
+describe('SyncPeer', () => {
+  test('basic - indexeddb', async () => {
+    let prev: any;
+    {
+      const workspace = new Workspace({
+        id: 'test',
+        isSSR: true,
+        schema,
+      });
+
+      const syncPeer = new SyncPeer(
+        workspace.doc,
+        createIndexedDBStorage(workspace.doc.guid)
+      );
+      await syncPeer.waitForLoaded();
+
+      const page = workspace.createPage({
+        id: 'page0',
+      });
+      await page.load();
+      const pageBlockId = page.addBlock('affine:page', {
+        title: new page.Text(''),
+      });
+      page.addBlock('affine:surface', {}, pageBlockId);
+      const frameId = page.addBlock('affine:note', {}, pageBlockId);
+      page.addBlock('affine:paragraph', {}, frameId);
+      await syncPeer.waitForSynced();
+      syncPeer.stop();
+      prev = workspace.doc.toJSON();
+    }
+
+    {
+      const workspace = new Workspace({
+        id: 'test',
+        isSSR: true,
+        schema,
+      });
+      const syncPeer = new SyncPeer(
+        workspace.doc,
+        createIndexedDBStorage(workspace.doc.guid)
+      );
+      await syncPeer.waitForSynced();
+      expect(workspace.doc.toJSON()).toEqual({
+        ...prev,
+      });
+      syncPeer.stop();
+    }
+  });
+
+  test('status', async () => {
+    const workspace = new Workspace({
+      id: 'test - status',
+      isSSR: true,
+      schema,
+    });
+
+    const syncPeer = new SyncPeer(
+      workspace.doc,
+      createIndexedDBStorage(workspace.doc.guid)
+    );
+    expect(syncPeer.status.step).toBe(SyncPeerStep.LoadingRootDoc);
+    await syncPeer.waitForSynced();
+    expect(syncPeer.status.step).toBe(SyncPeerStep.Synced);
+
+    const page = workspace.createPage({
+      id: 'page0',
+    });
+    expect(syncPeer.status.step).toBe(SyncPeerStep.LoadingSubDoc);
+    await page.load();
+    await syncPeer.waitForSynced();
+    page.addBlock('affine:page', {
+      title: new page.Text(''),
+    });
+    expect(syncPeer.status.step).toBe(SyncPeerStep.Syncing);
+    syncPeer.stop();
+  });
+});

packages/frontend/workspace/src/providers/sync/__tests__/test-storage.ts

@@ -0,0 +1,42 @@
+import type { Storage } from '../../storage';
+
+export function createTestStorage(origin: Storage) {
+  const controler = {
+    pausedPull: Promise.resolve(),
+    resumePull: () => {},
+    pausedPush: Promise.resolve(),
+    resumePush: () => {},
+  };
+
+  return {
+    name: `${origin.name}(testing)`,
+    pull(docId: string, state: Uint8Array) {
+      return controler.pausedPull.then(() => origin.pull(docId, state));
+    },
+    push(docId: string, data: Uint8Array) {
+      return controler.pausedPush.then(() => origin.push(docId, data));
+    },
+    subscribe(
+      cb: (docId: string, data: Uint8Array) => void,
+      disconnect: (reason: string) => void
+    ) {
+      return origin.subscribe(cb, disconnect);
+    },
+    pausePull() {
+      controler.pausedPull = new Promise(resolve => {
+        controler.resumePull = resolve;
+      });
+    },
+    resumePull() {
+      controler.resumePull?.();
+    },
+    pausePush() {
+      controler.pausedPush = new Promise(resolve => {
+        controler.resumePush = resolve;
+      });
+    },
+    resumePush() {
+      controler.resumePush?.();
+    },
+  };
+}

packages/frontend/workspace/src/providers/sync/engine.ts

@@ -0,0 +1,247 @@
+import { DebugLogger } from '@affine/debug';
+import { Slot } from '@blocksuite/global/utils';
+import type { Doc } from 'yjs';
+
+import type { Storage } from '../storage';
+import { SyncPeer, type SyncPeerStatus, SyncPeerStep } from './peer';
+
+export const MANUALLY_STOP = 'manually-stop';
+
+export enum SyncEngineStep {
+  Stopped = 0,
+  Syncing = 1,
+  Synced = 2,
+}
+
+export interface SyncEngineStatus {
+  step: SyncEngineStep;
+  local: SyncPeerStatus | null;
+  remotes: (SyncPeerStatus | null)[];
+  retrying: boolean;
+}
+
+/**
+ * # SyncEngine
+ *
+ * ```
+ *                    ┌────────────┐
+ *                    │ SyncEngine │
+ *                    └─────┬──────┘
+ *                          │
+ *                          ▼
+ *                    ┌────────────┐
+ *                    │  SyncPeer  │
+ *          ┌─────────┤   local    ├─────────┐
+ *          │         └─────┬──────┘         │
+ *          │               │                │
+ *          ▼               ▼                ▼
+ *   ┌────────────┐   ┌────────────┐   ┌────────────┐
+ *   │  SyncPeer  │   │  SyncPeer  │   │  SyncPeer  │
+ *   │   Remote   │   │   Remote   │   │   Remote   │
+ *   └────────────┘   └────────────┘   └────────────┘
+ * ```
+ *
+ * Sync engine manage sync peers
+ *
+ * Sync steps:
+ * 1. start local sync
+ * 2. wait for local sync complete
+ * 3. start remote sync
+ * 4. continuously sync local and remote
+ */
+export class SyncEngine {
+  get rootDocId() {
+    return this.rootDoc.guid;
+  }
+
+  logger = new DebugLogger('affine:sync-engine:' + this.rootDocId);
+  private _status: SyncEngineStatus;
+  onStatusChange = new Slot<SyncEngineStatus>();
+  private set status(s: SyncEngineStatus) {
+    this.logger.info('status change', SyncEngineStep[s.step]);
+    this._status = s;
+    this.onStatusChange.emit(s);
+  }
+
+  get status() {
+    return this._status;
+  }
+
+  private abort = new AbortController();
+
+  constructor(
+    private rootDoc: Doc,
+    private local: Storage,
+    private remotes: Storage[]
+  ) {
+    this._status = {
+      step: SyncEngineStep.Stopped,
+      local: null,
+      remotes: remotes.map(() => null),
+      retrying: false,
+    };
+  }
+
+  start() {
+    if (this.status.step !== SyncEngineStep.Stopped) {
+      this.stop();
+    }
+    this.abort = new AbortController();
+
+    this.sync(this.abort.signal).catch(err => {
+      // should never reach here
+      this.logger.error(err);
+    });
+  }
+
+  stop() {
+    this.abort.abort(MANUALLY_STOP);
+    this._status = {
+      step: SyncEngineStep.Stopped,
+      local: null,
+      remotes: this.remotes.map(() => null),
+      retrying: false,
+    };
+  }
+
+  // main sync process, should never return until abort
+  async sync(signal: AbortSignal) {
+    const state: {
+      localPeer: SyncPeer | null;
+      remotePeers: (SyncPeer | null)[];
+    } = {
+      localPeer: null,
+      remotePeers: this.remotes.map(() => null),
+    };
+
+    const cleanUp: (() => void)[] = [];
+    try {
+      // Step 1: start local sync peer
+      state.localPeer = new SyncPeer(this.rootDoc, this.local);
+
+      cleanUp.push(
+        state.localPeer.onStatusChange.on(() => {
+          if (!signal.aborted)
+            this.updateSyncingState(state.localPeer, state.remotePeers);
+        }).dispose
+      );
+
+      this.updateSyncingState(state.localPeer, state.remotePeers);
+
+      // Step 2: wait for local sync complete
+      await state.localPeer.waitForLoaded(signal);
+
+      // Step 3: start remote sync peer
+      state.remotePeers = this.remotes.map(remote => {
+        const peer = new SyncPeer(this.rootDoc, remote);
+        cleanUp.push(
+          peer.onStatusChange.on(() => {
+            if (!signal.aborted)
+              this.updateSyncingState(state.localPeer, state.remotePeers);
+          }).dispose
+        );
+        return peer;
+      });
+
+      this.updateSyncingState(state.localPeer, state.remotePeers);
+
+      // Step 4: continuously sync local and remote
+
+      // wait for abort
+      await new Promise((_, reject) => {
+        if (signal.aborted) {
+          reject(signal.reason);
+        }
+        signal.addEventListener('abort', () => {
+          reject(signal.reason);
+        });
+      });
+    } catch (error) {
+      if (error === MANUALLY_STOP) {
+        return;
+      }
+      throw error;
+    } finally {
+      // stop peers
+      state.localPeer?.stop();
+      for (const remotePeer of state.remotePeers) {
+        remotePeer?.stop();
+      }
+      for (const clean of cleanUp) {
+        clean();
+      }
+    }
+  }
+
+  updateSyncingState(local: SyncPeer | null, remotes: (SyncPeer | null)[]) {
+    let step = SyncEngineStep.Synced;
+    const allPeer = [local, ...remotes];
+    for (const peer of allPeer) {
+      if (!peer || peer.status.step !== SyncPeerStep.Synced) {
+        step = SyncEngineStep.Syncing;
+        break;
+      }
+    }
+    this.status = {
+      step,
+      local: local?.status ?? null,
+      remotes: remotes.map(peer => peer?.status ?? null),
+      retrying: allPeer.some(
+        peer => peer?.status.step === SyncPeerStep.Retrying
+      ),
+    };
+  }
+
+  async waitForSynced(abort?: AbortSignal) {
+    if (this.status.step == SyncEngineStep.Synced) {
+      return;
+    } else {
+      return Promise.race([
+        new Promise<void>(resolve => {
+          this.onStatusChange.on(status => {
+            if (status.step == SyncEngineStep.Synced) {
+              resolve();
+            }
+          });
+        }),
+        new Promise((_, reject) => {
+          if (abort?.aborted) {
+            reject(abort?.reason);
+          }
+          abort?.addEventListener('abort', () => {
+            reject(abort.reason);
+          });
+        }),
+      ]);
+    }
+  }
+
+  async waitForLoadedRootDoc(abort?: AbortSignal) {
+    function isLoadedRootDoc(status: SyncEngineStatus) {
+      return ![status.local, ...status.remotes].some(
+        peer => !peer || peer.step <= SyncPeerStep.LoadingRootDoc
+      );
+    }
+    if (isLoadedRootDoc(this.status)) {
+      return;
+    } else {
+      return Promise.race([
+        new Promise<void>(resolve => {
+          this.onStatusChange.on(status => {
+            if (isLoadedRootDoc(status)) {
+              resolve();
+            }
+          });
+        }),
+        new Promise((_, reject) => {
+          if (abort?.aborted) {
+            reject(abort?.reason);
+          }
+          abort?.addEventListener('abort', () => {
+            reject(abort.reason);
+          });
+        }),
+      ]);
+    }
+  }
+}

packages/frontend/workspace/src/providers/sync/index.ts

@@ -0,0 +1,18 @@
+/**
+ *
+ * **SyncEngine**
+ *
+ * Manages one local storage and multiple remote storages.
+ *
+ * Responsible for creating SyncPeers for synchronization, following the local-first strategy.
+ *
+ * **SyncPeer**
+ *
+ * Responsible for synchronizing a single storage with Y.Doc.
+ *
+ * Carries the main synchronization logic.
+ *
+ */
+
+export * from './engine';
+export * from './peer';

packages/frontend/workspace/src/providers/sync/peer.ts

@@ -0,0 +1,448 @@
+import { DebugLogger } from '@affine/debug';
+import { Slot } from '@blocksuite/global/utils';
+import { isEqual } from '@blocksuite/global/utils';
+import type { Doc } from 'yjs';
+import { applyUpdate, encodeStateAsUpdate, encodeStateVector } from 'yjs';
+
+import { mergeUpdates, type Storage } from '../storage';
+import { AsyncQueue } from '../utils/async-queue';
+import { throwIfAborted } from '../utils/throw-if-aborted';
+import { MANUALLY_STOP } from './engine';
+
+export enum SyncPeerStep {
+  Stopped = 0,
+  Retrying = 1,
+  LoadingRootDoc = 2,
+  LoadingSubDoc = 3,
+  Loaded = 4.5,
+  Syncing = 5,
+  Synced = 6,
+}
+
+export interface SyncPeerStatus {
+  step: SyncPeerStep;
+  totalDocs: number;
+  loadedDocs: number;
+  pendingPullUpdates: number;
+  pendingPushUpdates: number;
+}
+
+/**
+ * # SyncPeer
+ * A SyncPeer is responsible for syncing one Storage with one Y.Doc and its subdocs.
+ *
+ * ```
+ *                    ┌─────┐
+ *                    │Start│
+ *                    └──┬──┘
+ *                       │
+ *    ┌──────┐     ┌─────▼──────┐        ┌────┐
+ *    │listen◄─────┤pull rootdoc│        │peer│
+ *    └──┬───┘     └─────┬──────┘        └──┬─┘
+ *       │               │ onLoad()         │
+ *    ┌──▼───┐     ┌─────▼──────┐      ┌────▼────┐
+ *    │listen◄─────┤pull subdocs│      │subscribe│
+ *    └──┬───┘     └─────┬──────┘      └────┬────┘
+ *       │               │ onReady()        │
+ *    ┌──▼──┐      ┌─────▼───────┐       ┌──▼──┐
+ *    │queue├──────►apply updates◄───────┤queue│
+ *    └─────┘      └─────────────┘       └─────┘
+ * ```
+ *
+ * listen: listen for updates from ydoc, typically from user modifications.
+ * subscribe: listen for updates from storage, typically from other users.
+ *
+ */
+export class SyncPeer {
+  private _status: SyncPeerStatus = {
+    step: SyncPeerStep.LoadingRootDoc,
+    totalDocs: 1,
+    loadedDocs: 0,
+    pendingPullUpdates: 0,
+    pendingPushUpdates: 0,
+  };
+  onStatusChange = new Slot<SyncPeerStatus>();
+  abort = new AbortController();
+  get name() {
+    return this.storage.name;
+  }
+  logger = new DebugLogger('affine:sync-peer:' + this.name);
+
+  constructor(
+    private rootDoc: Doc,
+    private storage: Storage
+  ) {
+    this.logger.debug('peer start');
+
+    this.syncRetryLoop(this.abort.signal).catch(err => {
+      // should not reach here
+      console.error(err);
+    });
+  }
+
+  private set status(s: SyncPeerStatus) {
+    if (!isEqual(s, this._status)) {
+      this.logger.debug('status change', s);
+      this._status = s;
+      this.onStatusChange.emit(s);
+    }
+  }
+
+  get status() {
+    return this._status;
+  }
+
+  /**
+   * stop sync
+   *
+   * SyncPeer is one-time use, this peer should be discarded after call stop().
+   */
+  stop() {
+    this.logger.debug('peer stop');
+    this.abort.abort(MANUALLY_STOP);
+  }
+
+  /**
+   * auto retry after 5 seconds if sync failed
+   */
+  async syncRetryLoop(abort: AbortSignal) {
+    while (abort.aborted === false) {
+      try {
+        await this.sync(abort);
+      } catch (err) {
+        if (err === MANUALLY_STOP) {
+          return;
+        }
+
+        this.logger.error('sync error', err);
+      }
+      try {
+        this.logger.error('retry after 5 seconds');
+        this.status = {
+          step: SyncPeerStep.Retrying,
+          totalDocs: 1,
+          loadedDocs: 0,
+          pendingPullUpdates: 0,
+          pendingPushUpdates: 0,
+        };
+        await Promise.race([
+          new Promise<void>(resolve => {
+            setTimeout(resolve, 5 * 1000);
+          }),
+          new Promise((_, reject) => {
+            // exit if manually stopped
+            if (abort.aborted) {
+              reject(abort.reason);
+            }
+            abort.addEventListener('abort', () => {
+              reject(abort.reason);
+            });
+          }),
+        ]);
+      } catch (err) {
+        if (err === MANUALLY_STOP) {
+          return;
+        }
+
+        // should never reach here
+        throw err;
+      }
+    }
+  }
+
+  private state: {
+    connectedDocs: Map<string, Doc>;
+    pushUpdatesQueue: AsyncQueue<{
+      docId: string;
+      data: Uint8Array[];
+    }>;
+    pushingUpdate: boolean;
+    pullUpdatesQueue: AsyncQueue<{
+      docId: string;
+      data: Uint8Array;
+    }>;
+    subdocLoading: boolean;
+    subdocsLoadQueue: AsyncQueue<Doc>;
+  } = {
+    connectedDocs: new Map(),
+    pushUpdatesQueue: new AsyncQueue(),
+    pushingUpdate: false,
+    pullUpdatesQueue: new AsyncQueue(),
+    subdocLoading: false,
+    subdocsLoadQueue: new AsyncQueue(),
+  };
+
+  initState() {
+    this.state.connectedDocs.clear();
+    this.state.pushUpdatesQueue.clear();
+    this.state.pullUpdatesQueue.clear();
+    this.state.subdocsLoadQueue.clear();
+    this.state.pushingUpdate = false;
+    this.state.subdocLoading = false;
+  }
+
+  /**
+   * main synchronization logic
+   */
+  async sync(abortOuter: AbortSignal) {
+    this.initState();
+    const abortInner = new AbortController();
+
+    abortOuter.addEventListener('abort', reason => {
+      abortInner.abort(reason);
+    });
+
+    let dispose: (() => void) | null = null;
+    try {
+      this.reportSyncStatus();
+
+      // start listen storage updates
+      dispose = await this.storage.subscribe(
+        this.handleStorageUpdates,
+        reason => {
+          // abort if storage disconnect, should trigger retry loop
+          abortInner.abort('subscribe disconnect:' + reason);
+        }
+      );
+      throwIfAborted(abortInner.signal);
+
+      // Step 1: load root doc
+      await this.connectDoc(this.rootDoc, abortInner.signal);
+
+      // Step 2: load subdocs
+      this.state.subdocsLoadQueue.push(
+        ...Array.from(this.rootDoc.getSubdocs())
+      );
+      this.reportSyncStatus();
+
+      this.rootDoc.on('subdocs', this.handleSubdocsUpdate);
+
+      // Finally: start sync
+      await Promise.all([
+        // load subdocs
+        (async () => {
+          while (throwIfAborted(abortInner.signal)) {
+            const subdoc = await this.state.subdocsLoadQueue.next(
+              abortInner.signal
+            );
+            this.state.subdocLoading = true;
+            this.reportSyncStatus();
+            await this.connectDoc(subdoc, abortInner.signal);
+            this.state.subdocLoading = false;
+            this.reportSyncStatus();
+          }
+        })(),
+        // pull updates
+        (async () => {
+          while (throwIfAborted(abortInner.signal)) {
+            const { docId, data } = await this.state.pullUpdatesQueue.next(
+              abortInner.signal
+            );
+            // don't apply empty data or Uint8Array([0, 0])
+            if (
+              !(
+                data.byteLength === 0 ||
+                (data.byteLength === 2 && data[0] === 0 && data[1] === 0)
+              )
+            ) {
+              const subdoc = this.state.connectedDocs.get(docId);
+              if (subdoc) {
+                applyUpdate(subdoc, data, this.name);
+              }
+            }
+            this.reportSyncStatus();
+          }
+        })(),
+        // push updates
+        (async () => {
+          while (throwIfAborted(abortInner.signal)) {
+            const { docId, data } = await this.state.pushUpdatesQueue.next(
+              abortInner.signal
+            );
+            this.state.pushingUpdate = true;
+            this.reportSyncStatus();
+
+            const merged = mergeUpdates(data);
+
+            // don't push empty data or Uint8Array([0, 0])
+            if (
+              !(
+                merged.byteLength === 0 ||
+                (merged.byteLength === 2 && merged[0] === 0 && merged[1] === 0)
+              )
+            ) {
+              await this.storage.push(docId, merged);
+            }
+
+            this.state.pushingUpdate = false;
+            this.reportSyncStatus();
+          }
+        })(),
+      ]);
+    } finally {
+      dispose?.();
+      for (const docs of this.state.connectedDocs.values()) {
+        this.disconnectDoc(docs);
+      }
+      this.rootDoc.off('subdocs', this.handleSubdocsUpdate);
+    }
+  }
+
+  async connectDoc(doc: Doc, abort: AbortSignal) {
+    const { data: docData, state: inStorageState } =
+      (await this.storage.pull(doc.guid, encodeStateVector(doc))) ?? {};
+    throwIfAborted(abort);
+
+    if (docData) {
+      applyUpdate(doc, docData, 'load');
+    }
+
+    // diff root doc and in-storage, save updates to pendingUpdates
+    this.state.pushUpdatesQueue.push({
+      docId: doc.guid,
+      data: [encodeStateAsUpdate(doc, inStorageState)],
+    });
+
+    this.state.connectedDocs.set(doc.guid, doc);
+
+    // start listen root doc changes
+    doc.on('update', this.handleYDocUpdates);
+
+    // mark rootDoc as loaded
+    doc.emit('sync', [true]);
+
+    this.reportSyncStatus();
+  }
+
+  disconnectDoc(doc: Doc) {
+    doc.off('update', this.handleYDocUpdates);
+    this.state.connectedDocs.delete(doc.guid);
+    this.reportSyncStatus();
+  }
+
+  // handle updates from ydoc
+  handleYDocUpdates = (update: Uint8Array, origin: string, doc: Doc) => {
+    // don't push updates from storage
+    if (origin === this.name) {
+      return;
+    }
+
+    const exist = this.state.pushUpdatesQueue.find(
+      ({ docId }) => docId === doc.guid
+    );
+    if (exist) {
+      exist.data.push(update);
+    } else {
+      this.state.pushUpdatesQueue.push({
+        docId: doc.guid,
+        data: [update],
+      });
+    }
+
+    this.reportSyncStatus();
+  };
+
+  // handle subdocs changes, append new subdocs to queue, remove subdocs from queue
+  handleSubdocsUpdate = ({
+    added,
+    removed,
+  }: {
+    added: Set<Doc>;
+    removed: Set<Doc>;
+  }) => {
+    for (const subdoc of added) {
+      this.state.subdocsLoadQueue.push(subdoc);
+    }
+
+    for (const subdoc of removed) {
+      this.disconnectDoc(subdoc);
+      this.state.subdocsLoadQueue.remove(doc => doc === subdoc);
+    }
+    this.reportSyncStatus();
+  };
+
+  // handle updates from storage
+  handleStorageUpdates = (docId: string, data: Uint8Array) => {
+    this.state.pullUpdatesQueue.push({
+      docId,
+      data,
+    });
+    this.reportSyncStatus();
+  };
+
+  reportSyncStatus() {
+    let step;
+    if (this.state.connectedDocs.size === 0) {
+      step = SyncPeerStep.LoadingRootDoc;
+    } else if (this.state.subdocsLoadQueue.length || this.state.subdocLoading) {
+      step = SyncPeerStep.LoadingSubDoc;
+    } else if (
+      this.state.pullUpdatesQueue.length ||
+      this.state.pushUpdatesQueue.length ||
+      this.state.pushingUpdate
+    ) {
+      step = SyncPeerStep.Syncing;
+    } else {
+      step = SyncPeerStep.Synced;
+    }
+
+    this.status = {
+      step: step,
+      totalDocs:
+        this.state.connectedDocs.size + this.state.subdocsLoadQueue.length,
+      loadedDocs: this.state.connectedDocs.size,
+      pendingPullUpdates:
+        this.state.pullUpdatesQueue.length + (this.state.subdocLoading ? 1 : 0),
+      pendingPushUpdates:
+        this.state.pushUpdatesQueue.length + (this.state.pushingUpdate ? 1 : 0),
+    };
+  }
+
+  async waitForSynced(abort?: AbortSignal) {
+    if (this.status.step >= SyncPeerStep.Synced) {
+      return;
+    } else {
+      return Promise.race([
+        new Promise<void>(resolve => {
+          this.onStatusChange.on(status => {
+            if (status.step >= SyncPeerStep.Synced) {
+              resolve();
+            }
+          });
+        }),
+        new Promise((_, reject) => {
+          if (abort?.aborted) {
+            reject(abort?.reason);
+          }
+          abort?.addEventListener('abort', () => {
+            reject(abort.reason);
+          });
+        }),
+      ]);
+    }
+  }
+
+  async waitForLoaded(abort?: AbortSignal) {
+    if (this.status.step > SyncPeerStep.Loaded) {
+      return;
+    } else {
+      return Promise.race([
+        new Promise<void>(resolve => {
+          this.onStatusChange.on(status => {
+            if (status.step > SyncPeerStep.Loaded) {
+              resolve();
+            }
+          });
+        }),
+        new Promise((_, reject) => {
+          if (abort?.aborted) {
+            reject(abort?.reason);
+          }
+          abort?.addEventListener('abort', () => {
+            reject(abort.reason);
+          });
+        }),
+      ]);
+    }
+  }
+}

packages/frontend/workspace/src/providers/utils/__tests__/async-queue.spec.ts

@@ -0,0 +1,45 @@
+import { describe, expect, test, vi } from 'vitest';
+
+import { AsyncQueue } from '../async-queue';
+
+describe('async-queue', () => {
+  test('push & pop', async () => {
+    const queue = new AsyncQueue();
+    queue.push(1, 2, 3);
+    expect(queue.length).toBe(3);
+    expect(await queue.next()).toBe(1);
+    expect(await queue.next()).toBe(2);
+    expect(await queue.next()).toBe(3);
+    expect(queue.length).toBe(0);
+  });
+
+  test('await', async () => {
+    const queue = new AsyncQueue<number>();
+    queue.push(1, 2);
+    expect(await queue.next()).toBe(1);
+    expect(await queue.next()).toBe(2);
+
+    let v = -1;
+
+    // setup 2 pop tasks
+    queue.next().then(next => {
+      v = next;
+    });
+    queue.next().then(next => {
+      v = next;
+    });
+
+    // Wait for 100ms
+    await new Promise(resolve => setTimeout(resolve, 100));
+    // v should not be changed
+    expect(v).toBe(-1);
+
+    // push 3, should trigger the first pop task
+    queue.push(3);
+    await vi.waitFor(() => v === 3);
+
+    // push 4, should trigger the second pop task
+    queue.push(4);
+    await vi.waitFor(() => v === 4);
+  });
+});

packages/frontend/workspace/src/providers/utils/__tests__/throw-if-aborted.spec.ts

@@ -0,0 +1,13 @@
+import { describe, expect, test } from 'vitest';
+
+import { throwIfAborted } from '../throw-if-aborted';
+
+describe('throw-if-aborted', () => {
+  test('basic', async () => {
+    const abortController = new AbortController();
+    const abortSignal = abortController.signal;
+    expect(throwIfAborted(abortSignal)).toBe(true);
+    abortController.abort('TEST_ABORT');
+    expect(() => throwIfAborted(abortSignal)).toThrowError('TEST_ABORT');
+  });
+});

packages/frontend/workspace/src/providers/utils/affine-io.ts

@@ -0,0 +1,15 @@
+import { Manager } from 'socket.io-client';
+
+let ioManager: Manager | null = null;
+
+// use lazy initialization socket.io io manager
+export function getIoManager(): Manager {
+  if (ioManager) {
+    return ioManager;
+  }
+  ioManager = new Manager(runtimeConfig.serverUrlPrefix + '/', {
+    autoConnect: false,
+    transports: ['websocket'],
+  });
+  return ioManager;
+}

packages/frontend/workspace/src/providers/utils/async-queue.ts

@@ -0,0 +1,66 @@
+export class AsyncQueue<T> {
+  private _queue: T[];
+
+  private _resolveUpdate: (() => void) | null = null;
+  private _waitForUpdate: Promise<void> | null = null;
+
+  constructor(init: T[] = []) {
+    this._queue = init;
+  }
+
+  get length() {
+    return this._queue.length;
+  }
+
+  async next(abort?: AbortSignal): Promise<T> {
+    const update = this._queue.shift();
+    if (update) {
+      return update;
+    } else {
+      if (!this._waitForUpdate) {
+        this._waitForUpdate = new Promise(resolve => {
+          this._resolveUpdate = resolve;
+        });
+      }
+
+      await Promise.race([
+        this._waitForUpdate,
+        new Promise((_, reject) => {
+          if (abort?.aborted) {
+            reject(abort?.reason);
+          }
+          abort?.addEventListener('abort', () => {
+            reject(abort.reason);
+          });
+        }),
+      ]);
+
+      return this.next(abort);
+    }
+  }
+
+  push(...updates: T[]) {
+    this._queue.push(...updates);
+    if (this._resolveUpdate) {
+      const resolve = this._resolveUpdate;
+      this._resolveUpdate = null;
+      this._waitForUpdate = null;
+      resolve();
+    }
+  }
+
+  remove(predicate: (update: T) => boolean) {
+    const index = this._queue.findIndex(predicate);
+    if (index !== -1) {
+      this._queue.splice(index, 1);
+    }
+  }
+
+  find(predicate: (update: T) => boolean) {
+    return this._queue.find(predicate);
+  }
+
+  clear() {
+    this._queue = [];
+  }
+}

packages/frontend/workspace/src/providers/utils/base64.ts

@@ -1,20 +1,3 @@
-import type { Doc as YDoc } from 'yjs';
-
-export type SubdocEvent = {
-  loaded: Set<YDoc>;
-  removed: Set<YDoc>;
-  added: Set<YDoc>;
-};
-
-export type UpdateHandler = (update: Uint8Array, origin: unknown) => void;
-export type SubdocsHandler = (event: SubdocEvent) => void;
-export type DestroyHandler = () => void;
-
-export type AwarenessChanges = Record<
-  'added' | 'updated' | 'removed',
-  number[]
->;
-
 export function uint8ArrayToBase64(array: Uint8Array): Promise<string> {
   return new Promise<string>(resolve => {
     // Create a blob from the Uint8Array

packages/frontend/workspace/src/providers/utils/throw-if-aborted.ts

@@ -0,0 +1,7 @@
+// because AbortSignal.throwIfAborted is not available in abortcontroller-polyfill
+export function throwIfAborted(abort?: AbortSignal) {
+  if (abort?.aborted) {
+    throw new Error(abort.reason);
+  }
+  return true;
+}