feat(nbstore): add indexer storage (#10953)

packages/common/nbstore/src/worker/client.ts

@@ -1,20 +1,35 @@
 import { OpClient, transfer } from '@toeverything/infra/op';
+import type { Observable } from 'rxjs';
 import { v4 as uuid } from 'uuid';
 
 import { DummyConnection } from '../connection';
-import { AwarenessFrontend, BlobFrontend, DocFrontend } from '../frontend';
 import {
+  AwarenessFrontend,
+  BlobFrontend,
+  DocFrontend,
+  IndexerFrontend,
+} from '../frontend';
+import {
+  type AggregateOptions,
+  type AggregateResult,
   type AwarenessRecord,
   type BlobRecord,
   type BlobStorage,
   type DocRecord,
   type DocStorage,
   type DocUpdate,
+  type IndexerDocument,
+  type IndexerSchema,
+  type IndexerStorage,
   type ListedBlobRecord,
+  type Query,
+  type SearchOptions,
+  type SearchResult,
 } from '../storage';
 import type { AwarenessSync } from '../sync/awareness';
 import type { BlobSync } from '../sync/blob';
 import type { DocSync } from '../sync/doc';
+import type { IndexerSync } from '../sync/indexer';
 import type { StoreInitOptions, WorkerManagerOps, WorkerOps } from './ops';
 
 export type { StoreInitOptions as WorkerInitOptions } from './ops';
@@ -85,6 +100,12 @@ export class StoreClient {
     this.docFrontend = new DocFrontend(this.docStorage, this.docSync);
     this.blobFrontend = new BlobFrontend(this.blobStorage, this.blobSync);
     this.awarenessFrontend = new AwarenessFrontend(this.awarenessSync);
+    this.indexerStorage = new WorkerIndexerStorage(this.client);
+    this.indexerSync = new WorkerIndexerSync(this.client);
+    this.indexerFrontend = new IndexerFrontend(
+      this.indexerStorage,
+      this.indexerSync
+    );
   }
 
   private readonly docStorage: WorkerDocStorage;
@@ -92,14 +113,18 @@ export class StoreClient {
   private readonly docSync: WorkerDocSync;
   private readonly blobSync: WorkerBlobSync;
   private readonly awarenessSync: WorkerAwarenessSync;
+  private readonly indexerStorage: WorkerIndexerStorage;
+  private readonly indexerSync: WorkerIndexerSync;
 
   readonly docFrontend: DocFrontend;
   readonly blobFrontend: BlobFrontend;
   readonly awarenessFrontend: AwarenessFrontend;
+  readonly indexerFrontend: IndexerFrontend;
 }
 
 class WorkerDocStorage implements DocStorage {
   constructor(private readonly client: OpClient<WorkerOps>) {}
+  spaceId = '';
 
   readonly storageType = 'doc';
   readonly isReadonly = false;
@@ -316,3 +341,146 @@ class WorkerAwarenessSync implements AwarenessSync {
     };
   }
 }
+
+class WorkerIndexerStorage implements IndexerStorage {
+  constructor(private readonly client: OpClient<WorkerOps>) {}
+  readonly storageType = 'indexer';
+  readonly isReadonly = true;
+  connection = new DummyConnection();
+
+  search<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
+    table: T,
+    query: Query<T>,
+    options?: O
+  ): Promise<SearchResult<T, O>> {
+    return this.client.call('indexerStorage.search', { table, query, options });
+  }
+  aggregate<T extends keyof IndexerSchema, const O extends AggregateOptions<T>>(
+    table: T,
+    query: Query<T>,
+    field: keyof IndexerSchema[T],
+    options?: O
+  ): Promise<AggregateResult<T, O>> {
+    return this.client.call('indexerStorage.aggregate', {
+      table,
+      query,
+      field: field as string,
+      options,
+    });
+  }
+  search$<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
+    table: T,
+    query: Query<T>,
+    options?: O
+  ): Observable<SearchResult<T, O>> {
+    return this.client.ob$('indexerStorage.subscribeSearch', {
+      table,
+      query,
+      options,
+    });
+  }
+  aggregate$<
+    T extends keyof IndexerSchema,
+    const O extends AggregateOptions<T>,
+  >(
+    table: T,
+    query: Query<T>,
+    field: keyof IndexerSchema[T],
+    options?: O
+  ): Observable<AggregateResult<T, O>> {
+    return this.client.ob$('indexerStorage.subscribeAggregate', {
+      table,
+      query,
+      field: field as string,
+      options,
+    });
+  }
+  deleteByQuery<T extends keyof IndexerSchema>(
+    _table: T,
+    _query: Query<T>
+  ): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
+  insert<T extends keyof IndexerSchema>(
+    _table: T,
+    _document: IndexerDocument<T>
+  ): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
+  delete<T extends keyof IndexerSchema>(_table: T, _id: string): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
+  update<T extends keyof IndexerSchema>(
+    _table: T,
+    _document: IndexerDocument<T>
+  ): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
+  refresh<T extends keyof IndexerSchema>(_table: T): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
+}
+
+class WorkerIndexerSync implements IndexerSync {
+  constructor(private readonly client: OpClient<WorkerOps>) {}
+  waitForCompleted(signal?: AbortSignal): Promise<void> {
+    return new Promise((resolve, reject) => {
+      const abortListener = () => {
+        reject(signal?.reason);
+        subscription.unsubscribe();
+      };
+
+      signal?.addEventListener('abort', abortListener);
+
+      const subscription = this.client
+        .ob$('indexerSync.waitForCompleted')
+        .subscribe({
+          complete() {
+            signal?.removeEventListener('abort', abortListener);
+            resolve();
+          },
+          error(err) {
+            signal?.removeEventListener('abort', abortListener);
+            reject(err);
+          },
+        });
+    });
+  }
+  waitForDocCompleted(docId: string, signal?: AbortSignal): Promise<void> {
+    return new Promise((resolve, reject) => {
+      const abortListener = () => {
+        reject(signal?.reason);
+        subscription.unsubscribe();
+      };
+
+      signal?.addEventListener('abort', abortListener);
+
+      const subscription = this.client
+        .ob$('indexerSync.waitForDocCompleted', docId)
+        .subscribe({
+          complete() {
+            signal?.removeEventListener('abort', abortListener);
+            resolve();
+          },
+          error(err) {
+            signal?.removeEventListener('abort', abortListener);
+            reject(err);
+          },
+        });
+    });
+  }
+  get state$() {
+    return this.client.ob$('indexerSync.state');
+  }
+  docState$(docId: string) {
+    return this.client.ob$('indexerSync.docState', docId);
+  }
+  addPriority(docId: string, priority: number) {
+    const subscription = this.client
+      .ob$('indexerSync.addPriority', { docId, priority })
+      .subscribe();
+    return () => {
+      subscription.unsubscribe();
+    };
+  }
+}

packages/common/nbstore/src/worker/consumer.ts

@@ -1,4 +1,3 @@
-import { MANUALLY_STOP } from '@toeverything/infra';
 import { OpConsumer } from '@toeverything/infra/op';
 import { Observable } from 'rxjs';
 
@@ -7,6 +6,7 @@ import { SpaceStorage } from '../storage';
 import type { AwarenessRecord } from '../storage/awareness';
 import { Sync } from '../sync';
 import type { PeerStorageOptions } from '../sync/types';
+import { MANUALLY_STOP } from '../utils/throw-if-aborted';
 import type { StoreInitOptions, WorkerManagerOps, WorkerOps } from './ops';
 
 export type { WorkerManagerOps };
@@ -57,6 +57,14 @@ class StoreConsumer {
     return this.ensureSync.awareness;
   }
 
+  get indexerStorage() {
+    return this.ensureLocal.get('indexer');
+  }
+
+  get indexerSync() {
+    return this.ensureSync.indexer;
+  }
+
   constructor(
     private readonly availableStorageImplementations: StorageConstructor[],
     init: StoreInitOptions
@@ -262,6 +270,48 @@ class StoreConsumer {
         }),
       'awarenessSync.collect': ({ collectId, awareness }) =>
         collectJobs.get(collectId)?.(awareness),
+      'indexerStorage.aggregate': ({ table, query, field, options }) =>
+        this.indexerStorage.aggregate(table, query, field, options),
+      'indexerStorage.search': ({ table, query, options }) =>
+        this.indexerStorage.search(table, query, options),
+      'indexerStorage.subscribeSearch': ({ table, query, options }) =>
+        this.indexerStorage.search$(table, query, options),
+      'indexerStorage.subscribeAggregate': ({ table, query, field, options }) =>
+        this.indexerStorage.aggregate$(table, query, field, options),
+      'indexerSync.state': () => this.indexerSync.state$,
+      'indexerSync.docState': (docId: string) =>
+        this.indexerSync.docState$(docId),
+      'indexerSync.addPriority': ({ docId, priority }) =>
+        new Observable(() => {
+          const undo = this.indexerSync.addPriority(docId, priority);
+          return () => undo();
+        }),
+      'indexerSync.waitForCompleted': () =>
+        new Observable(subscriber => {
+          this.indexerSync
+            .waitForCompleted()
+            .then(() => {
+              subscriber.next();
+              subscriber.complete();
+            })
+            .catch(error => {
+              subscriber.error(error);
+            });
+        }),
+      'indexerSync.waitForDocCompleted': (docId: string) =>
+        new Observable(subscriber => {
+          const abortController = new AbortController();
+          this.indexerSync
+            .waitForDocCompleted(docId, abortController.signal)
+            .then(() => {
+              subscriber.next();
+              subscriber.complete();
+            })
+            .catch(error => {
+              subscriber.error(error);
+            });
+          return () => abortController.abort(MANUALLY_STOP);
+        }),
     });
   }
 }

packages/common/nbstore/src/worker/ops.ts

@@ -1,5 +1,7 @@
 import type { AvailableStorageImplementations } from '../impls';
 import type {
+  AggregateOptions,
+  AggregateResult,
   BlobRecord,
   DocClock,
   DocClocks,
@@ -7,11 +9,15 @@ import type {
   DocRecord,
   DocUpdate,
   ListedBlobRecord,
+  Query,
+  SearchOptions,
+  SearchResult,
   StorageType,
 } from '../storage';
 import type { AwarenessRecord } from '../storage/awareness';
 import type { BlobSyncBlobState, BlobSyncState } from '../sync/blob';
 import type { DocSyncDocState, DocSyncState } from '../sync/doc';
+import type { IndexerDocSyncState, IndexerSyncState } from '../sync/indexer';
 
 type StorageInitOptions = Values<{
   [key in keyof AvailableStorageImplementations]: {
@@ -61,6 +67,35 @@ interface GroupedWorkerOps {
     collect: [{ collectId: string; awareness: AwarenessRecord }, void];
   };
 
+  indexerStorage: {
+    search: [
+      { table: string; query: Query<any>; options?: SearchOptions<any> },
+      SearchResult<any, any>,
+    ];
+    aggregate: [
+      {
+        table: string;
+        query: Query<any>;
+        field: string;
+        options?: AggregateOptions<any>;
+      },
+      AggregateResult<any, any>,
+    ];
+    subscribeSearch: [
+      { table: string; query: Query<any>; options?: SearchOptions<any> },
+      SearchResult<any, any>,
+    ];
+    subscribeAggregate: [
+      {
+        table: string;
+        query: Query<any>;
+        field: string;
+        options?: AggregateOptions<any>;
+      },
+      AggregateResult<any, any>,
+    ];
+  };
+
   docSync: {
     state: [void, DocSyncState];
     docState: [string, DocSyncDocState];
@@ -91,6 +126,14 @@ interface GroupedWorkerOps {
     ];
     collect: [{ collectId: string; awareness: AwarenessRecord }, void];
   };
+
+  indexerSync: {
+    state: [void, IndexerSyncState];
+    docState: [string, IndexerDocSyncState];
+    addPriority: [{ docId: string; priority: number }, boolean];
+    waitForCompleted: [void, void];
+    waitForDocCompleted: [string, void];
+  };
 }
 
 type Values<T> = T extends { [k in keyof T]: any } ? T[keyof T] : never;