feat: init affine blob storage (#2045)

packages/workspace/package.json

@@ -3,6 +3,7 @@
   "private": true,
   "exports": {
     "./atom": "./src/atom.ts",
+    "./blob": "./src/blob/index.ts",
     "./utils": "./src/utils.ts",
     "./type": "./src/type.ts",
     "./local/crud": "./src/local/crud.ts",

packages/workspace/src/affine/__tests__/api.spec.ts

@@ -8,6 +8,7 @@ import { readFile } from 'node:fs/promises';
 import { MessageCode } from '@affine/env/constant';
 import { createStatusApis } from '@affine/workspace/affine/api/status';
 import { KeckProvider } from '@affine/workspace/affine/keck';
+import { WorkspaceFlavour } from '@affine/workspace/type';
 import { createEmptyBlockSuiteWorkspace } from '@affine/workspace/utils';
 import user1 from '@affine-test/fixtures/built-in-user1.json';
 import user2 from '@affine-test/fixtures/built-in-user2.json';
@@ -119,7 +120,7 @@ async function createWorkspace(
 ): Promise<string> {
   const workspace = createEmptyBlockSuiteWorkspace(
     faker.datatype.uuid(),
-    _ => undefined
+    WorkspaceFlavour.LOCAL
   );
   if (callback) {
     callback(workspace);
@@ -408,9 +409,15 @@ describe('api', () => {
         );
       });
       const binary = await workspaceApis.downloadWorkspace(id, false);
-      const workspace = createEmptyBlockSuiteWorkspace(id, () => undefined);
+      const workspace = createEmptyBlockSuiteWorkspace(
+        id,
+        WorkspaceFlavour.LOCAL
+      );
       Workspace.Y.applyUpdate(workspace.doc, new Uint8Array(binary));
-      const workspace2 = createEmptyBlockSuiteWorkspace(id, () => undefined);
+      const workspace2 = createEmptyBlockSuiteWorkspace(
+        id,
+        WorkspaceFlavour.LOCAL
+      );
       {
         const wsUrl = `ws://127.0.0.1:3000/api/sync/`;
         const provider = new KeckProvider(wsUrl, workspace.id, workspace.doc, {
@@ -459,7 +466,7 @@ describe('api', () => {
         );
         const publicWorkspace = createEmptyBlockSuiteWorkspace(
           id,
-          () => undefined
+          WorkspaceFlavour.LOCAL
         );
         Workspace.Y.applyUpdate(publicWorkspace.doc, new Uint8Array(binary));
         const publicPage = publicWorkspace.getPage(pageId) as Page;

packages/workspace/src/affine/__tests__/sync.spec.ts

@@ -1,3 +1,8 @@
+/**
+ * @vitest-environment happy-dom
+ */
+import 'fake-indexeddb/auto';
+
 import type { Workspace } from '@affine/workspace/affine/api';
 import {
   createWorkspaceApis,
@@ -6,6 +11,7 @@ import {
 import { KeckProvider } from '@affine/workspace/affine/keck';
 import type { LoginResponse } from '@affine/workspace/affine/login';
 import { loginResponseSchema } from '@affine/workspace/affine/login';
+import { WorkspaceFlavour } from '@affine/workspace/type';
 import { createEmptyBlockSuiteWorkspace } from '@affine/workspace/utils';
 import user1 from '@affine-test/fixtures/built-in-user1.json';
 import user2 from '@affine-test/fixtures/built-in-user2.json';
@@ -80,11 +86,17 @@ describe('ydoc sync', () => {
       const binary = await workspaceApis.downloadWorkspace(root.id);
       const workspace1 = createEmptyBlockSuiteWorkspace(
         root.id,
-        (k: string) => ({ api: '/api/workspace', token: user1Token.token }[k])
+        WorkspaceFlavour.AFFINE,
+        {
+          workspaceApis,
+        }
       );
       const workspace2 = createEmptyBlockSuiteWorkspace(
         root.id,
-        (k: string) => ({ api: '/api/workspace', token: user2Token.token }[k])
+        WorkspaceFlavour.AFFINE,
+        {
+          workspaceApis,
+        }
       );
       BlockSuiteWorkspace.Y.applyUpdate(workspace1.doc, new Uint8Array(binary));
       BlockSuiteWorkspace.Y.applyUpdate(workspace2.doc, new Uint8Array(binary));

packages/workspace/src/blob/index.ts

@@ -0,0 +1,106 @@
+import type { BlobStorage } from '@blocksuite/store';
+import { createIndexeddbStorage } from '@blocksuite/store';
+import { openDB } from 'idb';
+import type { DBSchema } from 'idb/build/entry';
+
+import type { createWorkspaceApis } from '../affine/api';
+
+type UploadingBlob = {
+  key: string;
+  arrayBuffer: ArrayBuffer;
+  type: string;
+};
+
+interface AffineBlob extends DBSchema {
+  uploading: {
+    key: string;
+    value: UploadingBlob;
+  };
+  // todo: migrate blob storage from `createIndexeddbStorage`
+}
+
+export const createAffineBlobStorage = (
+  workspaceId: string,
+  workspaceApis: ReturnType<typeof createWorkspaceApis>
+): BlobStorage => {
+  const storage = createIndexeddbStorage(workspaceId);
+  const dbPromise = openDB<AffineBlob>('affine-blob', 1, {
+    upgrade(db) {
+      db.createObjectStore('uploading', { keyPath: 'key' });
+    },
+  });
+  dbPromise.then(async db => {
+    const t = db.transaction('uploading', 'readwrite').objectStore('uploading');
+    await t.getAll().then(blobs =>
+      blobs.map(({ arrayBuffer, type }) =>
+        workspaceApis.uploadBlob(workspaceId, arrayBuffer, type).then(key => {
+          const t = db
+            .transaction('uploading', 'readwrite')
+            .objectStore('uploading');
+          return t.delete(key);
+        })
+      )
+    );
+  });
+  return {
+    crud: {
+      get: async key => {
+        const blob = await storage.crud.get(key);
+        if (!blob) {
+          const buffer = await workspaceApis.getBlob(workspaceId, key);
+          return new Blob([buffer]);
+        } else {
+          return blob;
+        }
+      },
+      set: async (key, value) => {
+        const db = await dbPromise;
+        const t = db
+          .transaction('uploading', 'readwrite')
+          .objectStore('uploading');
+        let uploaded = false;
+        t.put({
+          key,
+          arrayBuffer: await value.arrayBuffer(),
+          type: value.type,
+        }).then(() => {
+          // delete the uploading blob after uploaded
+          if (uploaded) {
+            const t = db
+              .transaction('uploading', 'readwrite')
+              .objectStore('uploading');
+            t.delete(key);
+          }
+        });
+        await Promise.all([
+          storage.crud.set(key, value),
+          workspaceApis
+            .uploadBlob(workspaceId, await value.arrayBuffer(), value.type)
+            .then(async () => {
+              uploaded = true;
+              const t = db
+                .transaction('uploading', 'readwrite')
+                .objectStore('uploading');
+              // delete the uploading blob after uploaded
+              if (await t.get(key)) {
+                await t.delete(key);
+              }
+            }),
+        ]);
+        return key;
+      },
+      delete: async (key: string) => {
+        await Promise.all([
+          storage.crud.delete(key),
+          // we don't support deleting a blob in API?
+          // workspaceApis.deleteBlob(workspaceId, key)
+        ]);
+      },
+      list: async () => {
+        const blobs = await storage.crud.list();
+        // we don't support listing blobs in API?
+        return [...blobs];
+      },
+    },
+  };
+};

packages/workspace/src/local/crud.ts

@@ -43,7 +43,7 @@ export const CRUD: WorkspaceCRUD<WorkspaceFlavour.LOCAL> = {
     }
     const blockSuiteWorkspace = createEmptyBlockSuiteWorkspace(
       id,
-      (_: string) => undefined
+      WorkspaceFlavour.LOCAL
     );
     const workspace: LocalWorkspace = {
       id,
@@ -62,7 +62,7 @@ export const CRUD: WorkspaceCRUD<WorkspaceFlavour.LOCAL> = {
     const id = nanoid();
     const blockSuiteWorkspace = createEmptyBlockSuiteWorkspace(
       id,
-      (_: string) => undefined
+      WorkspaceFlavour.LOCAL
     );
     BlockSuiteWorkspace.Y.applyUpdateV2(blockSuiteWorkspace.doc, binary);
     const persistence = createIndexedDBProvider(id, blockSuiteWorkspace.doc);

packages/workspace/src/utils.ts

@@ -1,16 +1,47 @@
+import type { createWorkspaceApis } from '@affine/workspace/affine/api';
+import { createAffineBlobStorage } from '@affine/workspace/blob';
 import { __unstableSchemas, AffineSchemas } from '@blocksuite/blocks/models';
-import type { BlobOptionsGetter, Generator } from '@blocksuite/store';
-import { Workspace } from '@blocksuite/store';
+import type { Generator } from '@blocksuite/store';
+import { createIndexeddbStorage, Workspace } from '@blocksuite/store';
+
+import { WorkspaceFlavour } from './type';
 
 const hashMap = new Map<string, Workspace>();
-export const createEmptyBlockSuiteWorkspace = (
+
+export function createEmptyBlockSuiteWorkspace(
   id: string,
-  blobOptionsGetter?: BlobOptionsGetter,
-  config?: {
+  flavour: WorkspaceFlavour.AFFINE,
+  config: {
+    workspaceApis: ReturnType<typeof createWorkspaceApis>;
     cachePrefix?: string;
     idGenerator?: Generator;
   }
-): Workspace => {
+): Workspace;
+export function createEmptyBlockSuiteWorkspace(
+  id: string,
+  flavour: WorkspaceFlavour.LOCAL,
+  config?: {
+    workspaceApis?: ReturnType<typeof createWorkspaceApis>;
+    cachePrefix?: string;
+    idGenerator?: Generator;
+  }
+): Workspace;
+export function createEmptyBlockSuiteWorkspace(
+  id: string,
+  flavour: WorkspaceFlavour,
+  config?: {
+    workspaceApis?: ReturnType<typeof createWorkspaceApis>;
+    cachePrefix?: string;
+    idGenerator?: Generator;
+  }
+): Workspace {
+  if (
+    flavour === WorkspaceFlavour.AFFINE &&
+    !config?.workspaceApis?.getBlob &&
+    !config?.workspaceApis?.uploadBlob
+  ) {
+    throw new Error('workspaceApis is required for affine flavour');
+  }
   const prefix: string = config?.cachePrefix ?? '';
   const cacheKey = `${prefix}${id}`;
   if (hashMap.has(cacheKey)) {
@@ -20,11 +51,16 @@ export const createEmptyBlockSuiteWorkspace = (
   const workspace = new Workspace({
     id,
     isSSR: typeof window === 'undefined',
-    blobOptionsGetter,
+    blobStorages:
+      flavour === WorkspaceFlavour.AFFINE
+        ? [id => createAffineBlobStorage(id, config!.workspaceApis!)]
+        : typeof window === 'undefined'
+        ? []
+        : [createIndexeddbStorage],
     idGenerator,
   })
     .register(AffineSchemas)
     .register(__unstableSchemas);
   hashMap.set(cacheKey, workspace);
   return workspace;
-};
+}