feat(core): use cloud indexer for search (#12899)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Added enhanced error handling and user-friendly error messages in
quick search and document search menus.
  - Introduced loading state indicators for search operations.
  - Quick Search now provides explicit error feedback in the UI.

- **Improvements**
- Search and aggregation operations can now prefer remote or local
indexers based on user or system preference.
- Streamlined indexer logic for more consistent and reliable search
experiences.
- Refined error handling in messaging and synchronization layers for
improved stability.
- Enhanced error object handling in messaging for clearer error
propagation.
- Updated cloud workspace storage to always use IndexedDB locally and
CloudIndexer remotely.
- Shifted indexer operations to use synchronized indexer layer for
better consistency.
  - Simplified indexer client by consolidating storage and sync layers.
- Improved error propagation in messaging handlers by wrapping error
objects.
- Updated document search to prioritize remote indexer results by
default.

- **Bug Fixes**
- Improved robustness of search features by handling errors gracefully
and preventing potential runtime issues.

- **Style**
  - Added new styles for displaying error messages in search interfaces.

- **Chores**
- Removed the obsolete "Enable Cloud Indexer" feature flag; cloud
indexer behavior is now always enabled where applicable.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/common/infra/src/op/__tests__/consumer.spec.ts

@@ -43,7 +43,10 @@ describe('op consumer', () => {
     expect(ctx.postMessage.mock.lastCall).toMatchInlineSnapshot(`
       [
         {
-          "error": [Error: Handler for operation [add] is not registered.],
+          "error": {
+            "message": "Handler for operation [add] is not registered.",
+            "name": "Error",
+          },
           "id": "add:1",
           "type": "return",
         },

packages/common/infra/src/op/client.ts

@@ -61,7 +61,7 @@ export class OpClient<Ops extends OpSchema> extends AutoMessageHandler {
     }
 
     if ('error' in msg) {
-      pending.reject(msg.error);
+      pending.reject(Object.assign(new Error(), msg.error));
     } else {
       pending.resolve(msg.data);
     }
@@ -86,7 +86,7 @@ export class OpClient<Ops extends OpSchema> extends AutoMessageHandler {
         return;
       }
 
-      ob.error(msg.error);
+      ob.error(Object.assign(new Error(), msg.error));
     };
 
   private readonly handleSubscriptionCompleteMessage: MessageHandlers['complete'] =

packages/common/infra/src/op/consumer.ts

@@ -1,4 +1,5 @@
 import EventEmitter2 from 'eventemitter2';
+import { pick } from 'lodash-es';
 import { defer, from, fromEvent, Observable, of, take, takeUntil } from 'rxjs';
 
 import { MANUALLY_STOP } from '../utils';
@@ -70,7 +71,15 @@ export class OpConsumer<Ops extends OpSchema> extends AutoMessageHandler {
           this.port.postMessage({
             type: 'return',
             id: msg.id,
-            error: error as Error,
+            error: pick(error, [
+              'name',
+              'message',
+              'code',
+              'type',
+              'status',
+              'data',
+              'stacktrace',
+            ]),
           } satisfies ReturnMessage);
         },
         complete: () => {
@@ -100,7 +109,15 @@ export class OpConsumer<Ops extends OpSchema> extends AutoMessageHandler {
           this.port.postMessage({
             type: 'error',
             id: msg.id,
-            error: error as Error,
+            error: pick(error, [
+              'name',
+              'message',
+              'code',
+              'type',
+              'status',
+              'data',
+              'stacktrace',
+            ]),
           } satisfies SubscriptionErrorMessage);
         },
         complete: () => {

packages/common/nbstore/src/frontend/indexer.ts

@@ -1,20 +1,13 @@
-import { switchMap } from 'rxjs';
-
 import type {
   AggregateOptions,
   IndexerSchema,
-  IndexerStorage,
   Query,
   SearchOptions,
 } from '../storage';
-import type { IndexerSync } from '../sync/indexer';
-import { fromPromise } from '../utils/from-promise';
+import type { IndexerPreferOptions, IndexerSync } from '../sync/indexer';
 
 export class IndexerFrontend {
-  constructor(
-    public readonly storage: IndexerStorage,
-    public readonly sync: IndexerSync
-  ) {}
+  constructor(public readonly sync: IndexerSync) {}
 
   get state$() {
     return this.sync.state$;
@@ -27,47 +20,47 @@ export class IndexerFrontend {
   async search<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
     table: T,
     query: Query<T>,
-    options?: O
+    options?: O & { prefer?: IndexerPreferOptions }
   ) {
-    await this.waitForConnected();
-    return this.storage.search(table, query, options);
+    return this.sync.search(table, query, options);
   }
 
   async aggregate<
     T extends keyof IndexerSchema,
     const O extends AggregateOptions<T>,
-  >(table: T, query: Query<T>, field: keyof IndexerSchema[T], options?: O) {
-    await this.waitForConnected();
-    return this.storage.aggregate(table, query, field, options);
+  >(
+    table: T,
+    query: Query<T>,
+    field: keyof IndexerSchema[T],
+    options?: O & { prefer?: IndexerPreferOptions }
+  ) {
+    return this.sync.aggregate(table, query, field, options);
   }
 
   search$<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
     table: T,
     query: Query<T>,
-    options?: O
+    options?: O & { prefer?: IndexerPreferOptions }
   ) {
-    return fromPromise(signal => this.waitForConnected(signal)).pipe(
-      switchMap(() => this.storage.search$(table, query, options))
-    );
+    return this.sync.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) {
-    return fromPromise(signal => this.waitForConnected(signal)).pipe(
-      switchMap(() => this.storage.aggregate$(table, query, field, options))
-    );
+  >(
+    table: T,
+    query: Query<T>,
+    field: keyof IndexerSchema[T],
+    options?: O & { prefer?: IndexerPreferOptions }
+  ) {
+    return this.sync.aggregate$(table, query, field, options);
   }
 
   addPriority(docId: string, priority: number) {
     return this.sync.addPriority(docId, priority);
   }
 
-  private waitForConnected(signal?: AbortSignal) {
-    return this.storage.connection.waitForConnected(signal);
-  }
-
   waitForCompleted(signal?: AbortSignal) {
     return this.sync.waitForCompleted(signal);
   }

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

@@ -9,7 +9,11 @@ import type { PeerStorageOptions } from './types';
 
 export type { BlobSyncState } from './blob';
 export type { DocSyncDocState, DocSyncState } from './doc';
-export type { IndexerDocSyncState, IndexerSyncState } from './indexer';
+export type {
+  IndexerDocSyncState,
+  IndexerPreferOptions,
+  IndexerSyncState,
+} from './indexer';
 
 export interface SyncState {
   doc?: DocSyncState;
@@ -65,7 +69,19 @@ export class Sync {
         ])
       ),
     });
-    this.indexer = new IndexerSyncImpl(doc, indexer, indexerSync);
+    this.indexer = new IndexerSyncImpl(
+      doc,
+      {
+        local: indexer,
+        remotes: Object.fromEntries(
+          Object.entries(storages.remotes).map(([peerId, remote]) => [
+            peerId,
+            remote.get('indexer'),
+          ])
+        ),
+      },
+      indexerSync
+    );
 
     this.state$ = this.doc.state$.pipe(map(doc => ({ doc })));
   }

packages/common/nbstore/src/sync/indexer/index.ts

@@ -1,4 +1,5 @@
 import { readAllDocsFromRootDoc } from '@affine/reader';
+import { omit } from 'lodash-es';
 import {
   filter,
   first,
@@ -7,21 +8,33 @@ import {
   ReplaySubject,
   share,
   Subject,
+  switchMap,
   throttleTime,
 } from 'rxjs';
 import { applyUpdate, Doc as YDoc } from 'yjs';
 
 import {
+  type AggregateOptions,
+  type AggregateResult,
   type DocStorage,
   IndexerDocument,
+  type IndexerSchema,
   type IndexerStorage,
+  type Query,
+  type SearchOptions,
+  type SearchResult,
 } from '../../storage';
+import { DummyIndexerStorage } from '../../storage/dummy/indexer';
 import type { IndexerSyncStorage } from '../../storage/indexer-sync';
 import { AsyncPriorityQueue } from '../../utils/async-priority-queue';
+import { fromPromise } from '../../utils/from-promise';
 import { takeUntilAbort } from '../../utils/take-until-abort';
 import { MANUALLY_STOP, throwIfAborted } from '../../utils/throw-if-aborted';
+import type { PeerStorageOptions } from '../types';
 import { crawlingDocData } from './crawler';
 
+export type IndexerPreferOptions = 'local' | 'remote';
+
 export interface IndexerSyncState {
   /**
    * Number of documents currently in the indexing queue
@@ -59,6 +72,35 @@ export interface IndexerSync {
   addPriority(docId: string, priority: number): () => void;
   waitForCompleted(signal?: AbortSignal): Promise<void>;
   waitForDocCompleted(docId: string, signal?: AbortSignal): Promise<void>;
+
+  search<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
+    table: T,
+    query: Query<T>,
+    options?: O & { prefer?: IndexerPreferOptions }
+  ): Promise<SearchResult<T, O>>;
+
+  aggregate<T extends keyof IndexerSchema, const O extends AggregateOptions<T>>(
+    table: T,
+    query: Query<T>,
+    field: keyof IndexerSchema[T],
+    options?: O & { prefer?: IndexerPreferOptions }
+  ): Promise<AggregateResult<T, O>>;
+
+  search$<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
+    table: T,
+    query: Query<T>,
+    options?: O & { prefer?: IndexerPreferOptions }
+  ): Observable<SearchResult<T, O>>;
+
+  aggregate$<
+    T extends keyof IndexerSchema,
+    const O extends AggregateOptions<T>,
+  >(
+    table: T,
+    query: Query<T>,
+    field: keyof IndexerSchema[T],
+    options?: O & { prefer?: IndexerPreferOptions }
+  ): Observable<AggregateResult<T, O>>;
 }
 
 export class IndexerSyncImpl implements IndexerSync {
@@ -70,6 +112,9 @@ export class IndexerSyncImpl implements IndexerSync {
   private readonly rootDocId = this.doc.spaceId;
   private readonly status = new IndexerSyncStatus(this.rootDocId);
 
+  private readonly indexer: IndexerStorage;
+  private readonly remote?: IndexerStorage;
+
   state$ = this.status.state$.pipe(
     // throttle the state to 1 second to avoid spamming the UI
     throttleTime(1000, undefined, {
@@ -106,9 +151,13 @@ export class IndexerSyncImpl implements IndexerSync {
 
   constructor(
     readonly doc: DocStorage,
-    readonly indexer: IndexerStorage,
+    readonly peers: PeerStorageOptions<IndexerStorage>,
     readonly indexerSync: IndexerSyncStorage
-  ) {}
+  ) {
+    // sync feature only works on local indexer
+    this.indexer = this.peers.local;
+    this.remote = Object.values(this.peers.remotes).find(remote => !!remote);
+  }
 
   start() {
     if (this.abort) {
@@ -439,6 +488,116 @@ export class IndexerSyncImpl implements IndexerSync {
       })
     );
   }
+
+  async search<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
+    table: T,
+    query: Query<T>,
+    options?: O & { prefer?: IndexerPreferOptions }
+  ): Promise<SearchResult<T, O>> {
+    if (
+      options?.prefer === 'remote' &&
+      this.remote &&
+      !(this.remote instanceof DummyIndexerStorage)
+    ) {
+      await this.remote.connection.waitForConnected();
+      return await this.remote.search(table, query, omit(options, 'prefer'));
+    } else {
+      await this.indexer.connection.waitForConnected();
+      return await this.indexer.search(table, query, omit(options, 'prefer'));
+    }
+  }
+
+  async aggregate<
+    T extends keyof IndexerSchema,
+    const O extends AggregateOptions<T>,
+  >(
+    table: T,
+    query: Query<T>,
+    field: keyof IndexerSchema[T],
+    options?: O & { prefer?: IndexerPreferOptions }
+  ): Promise<AggregateResult<T, O>> {
+    if (
+      options?.prefer === 'remote' &&
+      this.remote &&
+      !(this.remote instanceof DummyIndexerStorage)
+    ) {
+      await this.remote.connection.waitForConnected();
+      return await this.remote.aggregate(
+        table,
+        query,
+        field,
+        omit(options, 'prefer')
+      );
+    } else {
+      await this.indexer.connection.waitForConnected();
+      return await this.indexer.aggregate(
+        table,
+        query,
+        field,
+        omit(options, 'prefer')
+      );
+    }
+  }
+
+  search$<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
+    table: T,
+    query: Query<T>,
+    options?: O & { prefer?: IndexerPreferOptions }
+  ): Observable<SearchResult<T, O>> {
+    if (
+      options?.prefer === 'remote' &&
+      this.remote &&
+      !(this.remote instanceof DummyIndexerStorage)
+    ) {
+      const remote = this.remote;
+      return fromPromise(signal =>
+        remote.connection.waitForConnected(signal)
+      ).pipe(
+        switchMap(() => remote.search$(table, query, omit(options, 'prefer')))
+      );
+    } else {
+      return fromPromise(signal =>
+        this.indexer.connection.waitForConnected(signal)
+      ).pipe(
+        switchMap(() =>
+          this.indexer.search$(table, query, omit(options, 'prefer'))
+        )
+      );
+    }
+  }
+
+  aggregate$<
+    T extends keyof IndexerSchema,
+    const O extends AggregateOptions<T>,
+  >(
+    table: T,
+    query: Query<T>,
+    field: keyof IndexerSchema[T],
+    options?: O & { prefer?: IndexerPreferOptions }
+  ): Observable<AggregateResult<T, O>> {
+    if (
+      options?.prefer === 'remote' &&
+      this.remote &&
+      !(this.remote instanceof DummyIndexerStorage)
+    ) {
+      const remote = this.remote;
+      return fromPromise(signal =>
+        remote.connection.waitForConnected(signal)
+      ).pipe(
+        switchMap(() =>
+          remote.aggregate$(table, query, field, omit(options, 'prefer'))
+        )
+      );
+    } else {
+      return fromPromise(signal =>
+        this.indexer.connection.waitForConnected(signal)
+      ).pipe(
+        switchMap(() =>
+          this.indexer.aggregate$(table, query, field, omit(options, 'prefer'))
+        )
+      );
+    }
+  }
 }
 
 class IndexerSyncStatus {

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

@@ -18,9 +18,7 @@ import {
   type DocRecord,
   type DocStorage,
   type DocUpdate,
-  type IndexerDocument,
   type IndexerSchema,
-  type IndexerStorage,
   type ListedBlobRecord,
   type Query,
   type SearchOptions,
@@ -29,7 +27,7 @@ import {
 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 { IndexerPreferOptions, IndexerSync } from '../sync/indexer';
 import type { StoreInitOptions, WorkerManagerOps, WorkerOps } from './ops';
 
 export type { StoreInitOptions as WorkerInitOptions } from './ops';
@@ -100,12 +98,8 @@ 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
-    );
+    this.indexerFrontend = new IndexerFrontend(this.indexerSync);
   }
 
   private readonly docStorage: WorkerDocStorage;
@@ -113,7 +107,6 @@ 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;
@@ -348,26 +341,23 @@ class WorkerAwarenessSync implements AwarenessSync {
   }
 }
 
-class WorkerIndexerStorage implements IndexerStorage {
+class WorkerIndexerSync implements IndexerSync {
   constructor(private readonly client: OpClient<WorkerOps>) {}
-  readonly storageType = 'indexer';
-  readonly isReadonly = true;
-  connection = new WorkerIndexerConnection(this.client);
 
   search<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
     table: T,
     query: Query<T>,
-    options?: O
+    options?: O & { prefer?: IndexerPreferOptions }
   ): Promise<SearchResult<T, O>> {
-    return this.client.call('indexerStorage.search', { table, query, options });
+    return this.client.call('indexerSync.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
+    options?: O & { prefer?: IndexerPreferOptions }
   ): Promise<AggregateResult<T, O>> {
-    return this.client.call('indexerStorage.aggregate', {
+    return this.client.call('indexerSync.aggregate', {
       table,
       query,
       field: field as string,
@@ -377,9 +367,9 @@ class WorkerIndexerStorage implements IndexerStorage {
   search$<T extends keyof IndexerSchema, const O extends SearchOptions<T>>(
     table: T,
     query: Query<T>,
-    options?: O
+    options?: O & { prefer?: IndexerPreferOptions }
   ): Observable<SearchResult<T, O>> {
-    return this.client.ob$('indexerStorage.subscribeSearch', {
+    return this.client.ob$('indexerSync.subscribeSearch', {
       table,
       query,
       options,
@@ -392,59 +382,16 @@ class WorkerIndexerStorage implements IndexerStorage {
     table: T,
     query: Query<T>,
     field: keyof IndexerSchema[T],
-    options?: O
+    options?: O & { prefer?: IndexerPreferOptions }
   ): Observable<AggregateResult<T, O>> {
-    return this.client.ob$('indexerStorage.subscribeAggregate', {
+    return this.client.ob$('indexerSync.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 WorkerIndexerConnection extends DummyConnection {
-  constructor(private readonly client: OpClient<WorkerOps>) {
-    super();
-  }
-
-  promise: Promise<void> | undefined;
-
-  override waitForConnected(): Promise<void> {
-    if (this.promise) {
-      return this.promise;
-    }
-    this.promise = this.client.call('indexerStorage.waitForConnected');
-    return this.promise;
-  }
-}
-
-class WorkerIndexerSync implements IndexerSync {
-  constructor(private readonly client: OpClient<WorkerOps>) {}
   waitForCompleted(signal?: AbortSignal): Promise<void> {
     return new Promise((resolve, reject) => {
       const abortListener = () => {

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

@@ -265,16 +265,6 @@ 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),
-      'indexerStorage.waitForConnected': (_, ctx) =>
-        this.indexerStorage.connection.waitForConnected(ctx.signal),
       'indexerSync.state': () => this.indexerSync.state$,
       'indexerSync.docState': (docId: string) =>
         this.indexerSync.docState$(docId),
@@ -287,6 +277,14 @@ class StoreConsumer {
         this.indexerSync.waitForCompleted(ctx.signal),
       'indexerSync.waitForDocCompleted': (docId: string, ctx) =>
         this.indexerSync.waitForDocCompleted(docId, ctx.signal),
+      'indexerSync.aggregate': ({ table, query, field, options }) =>
+        this.indexerSync.aggregate(table, query, field, options),
+      'indexerSync.search': ({ table, query, options }) =>
+        this.indexerSync.search(table, query, options),
+      'indexerSync.subscribeSearch': ({ table, query, options }) =>
+        this.indexerSync.search$(table, query, options),
+      'indexerSync.subscribeAggregate': ({ table, query, field, options }) =>
+        this.indexerSync.aggregate$(table, query, field, options),
     });
   }
 }

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

@@ -1,6 +1,5 @@
 import type { AvailableStorageImplementations } from '../impls';
 import type {
-  AggregateOptions,
   AggregateResult,
   BlobRecord,
   DocClock,
@@ -10,7 +9,6 @@ import type {
   DocUpdate,
   ListedBlobRecord,
   Query,
-  SearchOptions,
   SearchResult,
   StorageType,
 } from '../storage';
@@ -69,36 +67,6 @@ interface GroupedWorkerOps {
     waitForConnected: [void, 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>,
-    ];
-    waitForConnected: [void, void];
-  };
-
   docSync: {
     state: [void, DocSyncState];
     docState: [string, DocSyncDocState];
@@ -137,6 +105,40 @@ interface GroupedWorkerOps {
     addPriority: [{ docId: string; priority: number }, boolean];
     waitForCompleted: [void, void];
     waitForDocCompleted: [string, void];
+    search: [
+      {
+        table: string;
+        query: Query<any>;
+        options?: any;
+      },
+      SearchResult<any, any>,
+    ];
+    aggregate: [
+      {
+        table: string;
+        query: Query<any>;
+        field: string;
+        options?: any;
+      },
+      AggregateResult<any, any>,
+    ];
+    subscribeSearch: [
+      {
+        table: string;
+        query: Query<any>;
+        options?: any;
+      },
+      SearchResult<any, any>,
+    ];
+    subscribeAggregate: [
+      {
+        table: string;
+        query: Query<any>;
+        field: string;
+        options?: any;
+      },
+      AggregateResult<any, any>,
+    ];
   };
 }