feat(infra): subscribe for indexer

packages/common/infra/src/framework/core/index.ts

@@ -6,5 +6,6 @@ export * from './error';
 export { createEvent, OnEvent } from './event';
 export { Framework } from './framework';
 export { createIdentifier } from './identifier';
-export type { FrameworkProvider, ResolveOptions } from './provider';
+export type { ResolveOptions } from './provider';
+export { FrameworkProvider } from './provider';
 export type { GeneralIdentifier } from './types';

packages/common/infra/src/framework/react/index.tsx

@@ -9,6 +9,12 @@ export const FrameworkStackContext = React.createContext<FrameworkProvider[]>([
   Framework.EMPTY.provider(),
 ]);
 
+export function useFramework(): FrameworkProvider {
+  const stack = useContext(FrameworkStackContext);
+
+  return stack[stack.length - 1]; // never null, because the default value
+}
+
 export function useService<T extends Service>(
   identifier: GeneralIdentifier<T>
 ): T {

packages/common/infra/src/modules/doc/entities/record.ts

@@ -13,7 +13,6 @@ export type DocMode = 'edgeless' | 'page';
  */
 export class DocRecord extends Entity<{ id: string }> {
   id: string = this.props.id;
-  meta: Partial<DocMeta> | null = null;
   constructor(private readonly docsStore: DocsStore) {
     super();
   }

packages/common/infra/src/modules/doc/services/docs.ts

@@ -1,6 +1,10 @@
+import { Unreachable } from '@affine/env/constant';
+
 import { Service } from '../../../framework';
+import { initEmptyPage } from '../../../initialization';
 import { ObjectPool } from '../../../utils';
 import type { Doc } from '../entities/doc';
+import type { DocMode } from '../entities/record';
 import { DocRecordList } from '../entities/record-list';
 import { DocScope } from '../scopes/doc';
 import type { DocsStore } from '../stores/docs';
@@ -46,4 +50,22 @@ export class DocsService extends Service {
 
     return { doc: obj, release };
   }
+
+  createDoc(
+    options: {
+      mode?: DocMode;
+      title?: string;
+    } = {}
+  ) {
+    const doc = this.store.createBlockSuiteDoc();
+    initEmptyPage(doc, options.title);
+    const docRecord = this.list.doc$(doc.id).value;
+    if (!docRecord) {
+      throw new Unreachable();
+    }
+    if (options.mode) {
+      docRecord.setMode(options.mode);
+    }
+    return docRecord;
+  }
 }

packages/common/infra/src/modules/doc/stores/docs.ts

@@ -18,6 +18,10 @@ export class DocsStore extends Store {
     return this.workspaceService.workspace.docCollection.getDoc(id);
   }
 
+  createBlockSuiteDoc() {
+    return this.workspaceService.workspace.docCollection.createDoc();
+  }
+
   watchDocIds() {
     return new Observable<string[]>(subscriber => {
       const emit = () => {

packages/common/infra/src/modules/workspace/entities/engine.ts

@@ -3,6 +3,7 @@ import type { Doc as YDoc } from 'yjs';
 import { Entity } from '../../../framework';
 import { AwarenessEngine, BlobEngine, DocEngine } from '../../../sync';
 import { throwIfAborted } from '../../../utils';
+import { WorkspaceEngineBeforeStart } from '../events';
 import type { WorkspaceEngineProvider } from '../providers/flavour';
 import type { WorkspaceService } from '../services/workspace';
 
@@ -33,6 +34,7 @@ export class WorkspaceEngine extends Entity<{
   }
 
   start() {
+    this.eventBus.emit(WorkspaceEngineBeforeStart, this);
     this.doc.start();
     this.awareness.connect(this.workspaceService.workspace.awareness);
     this.blob.start();

packages/common/infra/src/modules/workspace/events/index.ts

@@ -0,0 +1,6 @@
+import { createEvent } from '../../../framework';
+import type { WorkspaceEngine } from '../entities/engine';
+
+export const WorkspaceEngineBeforeStart = createEvent<WorkspaceEngine>(
+  'WorkspaceEngineBeforeStart'
+);

packages/common/infra/src/modules/workspace/index.ts

@@ -1,5 +1,6 @@
 export type { WorkspaceProfileInfo } from './entities/profile';
 export { Workspace } from './entities/workspace';
+export { WorkspaceEngineBeforeStart } from './events';
 export { globalBlockSuiteSchema } from './global-schema';
 export type { WorkspaceMetadata } from './metadata';
 export type { WorkspaceOpenOptions } from './open-options';

packages/common/infra/src/sync/doc/index.ts

@@ -23,6 +23,7 @@ export {
 } from './storage';
 
 export class DocEngine {
+  clientId: string;
   localPart: DocEngineLocalPart;
   remotePart: DocEngineRemotePart | null;
 
@@ -80,11 +81,11 @@ export class DocEngine {
     storage: DocStorage,
     private readonly server?: DocServer | null
   ) {
-    const clientId = nanoid();
+    this.clientId = nanoid();
     this.storage = new DocStorageInner(storage);
-    this.localPart = new DocEngineLocalPart(clientId, this.storage);
+    this.localPart = new DocEngineLocalPart(this.clientId, this.storage);
     this.remotePart = this.server
-      ? new DocEngineRemotePart(clientId, this.storage, this.server)
+      ? new DocEngineRemotePart(this.clientId, this.storage, this.server)
       : null;
   }
 

packages/common/infra/src/sync/index.ts

@@ -4,3 +4,11 @@ export type { BlobStatus, BlobStorage } from './blob/blob';
 export { BlobEngine, EmptyBlobStorage } from './blob/blob';
 export { BlobStorageOverCapacity } from './blob/error';
 export * from './doc';
+export * from './indexer';
+export {
+  IndexedDBIndex,
+  IndexedDBIndexStorage,
+} from './indexer/impl/indexeddb';
+export { MemoryIndex, MemoryIndexStorage } from './indexer/impl/memory';
+export * from './job';
+export { IndexedDBJobQueue } from './job/impl/indexeddb';

packages/common/infra/src/sync/indexer/__tests__/black-box.spec.ts

@@ -3,7 +3,8 @@
  */
 import 'fake-indexeddb/auto';
 
-import { beforeEach, describe, expect, test } from 'vitest';
+import { map } from 'rxjs';
+import { beforeEach, describe, expect, test, vitest } from 'vitest';
 
 import { defineSchema, Document, type Index } from '..';
 import { IndexedDBIndex } from '../impl/indexeddb';
@@ -21,7 +22,7 @@ describe.each([
   { name: 'memory', backend: MemoryIndex },
   { name: 'idb', backend: IndexedDBIndex },
 ])('index tests($name)', ({ backend }) => {
-  async function initData(
+  async function writeData(
     data: Record<
       string,
       Partial<Record<keyof typeof schema, string | string[]>>
@@ -50,7 +51,7 @@ describe.each([
   });
 
   test('basic', async () => {
-    await initData({
+    await writeData({
       '1': {
         title: 'hello world',
       },
@@ -79,7 +80,7 @@ describe.each([
   });
 
   test('basic integer', async () => {
-    await initData({
+    await writeData({
       '1': {
         title: 'hello world',
         size: '100',
@@ -109,7 +110,7 @@ describe.each([
   });
 
   test('fuzz', async () => {
-    await initData({
+    await writeData({
       '1': {
         title: 'hello world',
       },
@@ -137,7 +138,7 @@ describe.each([
   });
 
   test('highlight', async () => {
-    await initData({
+    await writeData({
       '1': {
         title: 'hello world',
         size: '100',
@@ -181,7 +182,7 @@ describe.each([
   });
 
   test('fields', async () => {
-    await initData({
+    await writeData({
       '1': {
         title: 'hello world',
         tag: ['car', 'bike'],
@@ -206,7 +207,7 @@ describe.each([
   });
 
   test('pagination', async () => {
-    await initData(
+    await writeData(
       Array.from({ length: 100 }).reduce((acc: any, _, i) => {
         acc['apple' + i] = {
           tag: ['apple'],
@@ -275,7 +276,7 @@ describe.each([
   });
 
   test('aggr', async () => {
-    await initData({
+    await writeData({
       '1': {
         title: 'hello world',
         tag: ['car', 'bike'],
@@ -315,7 +316,7 @@ describe.each([
   });
 
   test('hits', async () => {
-    await initData(
+    await writeData(
       Array.from({ length: 100 }).reduce((acc: any, _, i) => {
         acc['apple' + i] = {
           title: 'apple',
@@ -412,4 +413,102 @@ describe.each([
       },
     });
   });
+
+  test('subscribe', async () => {
+    await writeData({
+      '1': {
+        title: 'hello world',
+      },
+    });
+
+    let value = null as any;
+    index
+      .search$({
+        type: 'match',
+        field: 'title',
+        match: 'hello world',
+      })
+      .pipe(map(v => (value = v)))
+      .subscribe();
+
+    await vitest.waitFor(
+      () => {
+        expect(value).toEqual({
+          nodes: [
+            {
+              id: '1',
+              score: expect.anything(),
+            },
+          ],
+          pagination: {
+            count: 1,
+            hasMore: false,
+            limit: expect.anything(),
+            skip: 0,
+          },
+        });
+      },
+      {
+        timeout: 5000,
+      }
+    );
+
+    await writeData({
+      '2': {
+        title: 'hello world',
+      },
+    });
+
+    await vitest.waitFor(
+      () => {
+        expect(value).toEqual({
+          nodes: [
+            {
+              id: '1',
+              score: expect.anything(),
+            },
+            {
+              id: '2',
+              score: expect.anything(),
+            },
+          ],
+          pagination: {
+            count: 2,
+            hasMore: false,
+            limit: expect.anything(),
+            skip: 0,
+          },
+        });
+      },
+      {
+        timeout: 5000,
+      }
+    );
+
+    const writer = await index.write();
+    writer.delete('1');
+    await writer.commit();
+
+    await vitest.waitFor(
+      () => {
+        expect(value).toEqual({
+          nodes: [
+            {
+              id: '2',
+              score: expect.anything(),
+            },
+          ],
+          pagination: {
+            count: 1,
+            hasMore: false,
+            limit: expect.anything(),
+            skip: 0,
+          },
+        });
+      },
+      {
+        timeout: 5000,
+      }
+    );
+  });
 });

packages/common/infra/src/sync/indexer/document.ts

@@ -28,7 +28,9 @@ export class Document<S extends Schema = any> {
 
   static from<S extends Schema>(
     id: string,
-    map: Record<keyof S, string | string[]> | Map<keyof S, string | string[]>
+    map:
+      | Partial<Record<keyof S, string | string[]>>
+      | Map<keyof S, string | string[]>
   ): Document<S> {
     const doc = new Document(id);
 

packages/common/infra/src/sync/indexer/impl/indexeddb/data-struct.ts

@@ -74,7 +74,7 @@ export class DataStruct {
   invertedIndex = new Map<string, InvertedIndex>();
 
   constructor(
-    private readonly databaseName: string,
+    readonly databaseName: string,
     schema: Schema
   ) {
     for (const [key, type] of Object.entries(schema)) {