feat(nbstore): add cloud indexer storage (#12245)

close AF-2613

packages/common/nbstore/src/impls/cloud/index.ts

@@ -3,11 +3,13 @@ import { CloudAwarenessStorage } from './awareness';
 import { CloudBlobStorage } from './blob';
 import { CloudDocStorage } from './doc';
 import { StaticCloudDocStorage } from './doc-static';
+import { CloudIndexerStorage } from './indexer';
 
 export * from './awareness';
 export * from './blob';
 export * from './doc';
 export * from './doc-static';
+export * from './indexer';
 export * from './socket';
 
 export const cloudStorages = [
@@ -15,4 +17,5 @@ export const cloudStorages = [
   StaticCloudDocStorage,
   CloudBlobStorage,
   CloudAwarenessStorage,
+  CloudIndexerStorage,
 ] satisfies StorageConstructor[];

packages/common/nbstore/src/impls/cloud/indexer.ts

@@ -0,0 +1,142 @@
+import {
+  type AggregateInput,
+  indexerAggregateQuery,
+  indexerSearchQuery,
+  type SearchInput,
+} from '@affine/graphql';
+import { Observable } from 'rxjs';
+
+import {
+  type AggregateOptions,
+  type AggregateResult,
+  type IndexerDocument,
+  type IndexerSchema,
+  IndexerStorageBase,
+  type Query,
+  type SearchOptions,
+  type SearchResult,
+} from '../../storage/indexer';
+import { HttpConnection } from './http';
+
+interface CloudIndexerStorageOptions {
+  serverBaseUrl: string;
+  id: string;
+}
+
+export class CloudIndexerStorage extends IndexerStorageBase {
+  static readonly identifier = 'CloudIndexerStorage';
+  readonly isReadonly = true;
+  readonly connection = new HttpConnection(this.options.serverBaseUrl);
+
+  constructor(private readonly options: CloudIndexerStorageOptions) {
+    super();
+  }
+
+  override async search<
+    T extends keyof IndexerSchema,
+    const O extends SearchOptions<T>,
+  >(table: T, query: Query<T>, options?: O): Promise<SearchResult<T, O>> {
+    const res = await this.connection.gql({
+      query: indexerSearchQuery,
+      variables: {
+        id: this.options.id,
+        input: {
+          table,
+          query,
+          options,
+        } as SearchInput,
+      },
+    });
+    const result = res.workspace.search as unknown as SearchResult<T, O>;
+    return result;
+  }
+
+  override search$<
+    T extends keyof IndexerSchema,
+    const O extends SearchOptions<T>,
+  >(table: T, query: Query<T>, options?: O): Observable<SearchResult<T, O>> {
+    return new Observable(observer => {
+      this.search(table, query, options)
+        .then(data => {
+          observer.next(data);
+          observer.complete();
+        })
+        .catch(error => {
+          observer.error(error);
+        });
+    });
+  }
+
+  override async 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>> {
+    const res = await this.connection.gql({
+      query: indexerAggregateQuery,
+      variables: {
+        id: this.options.id,
+        input: { table, query, field, options } as AggregateInput,
+      },
+    });
+    const result = res.workspace.aggregate as unknown as AggregateResult<T, O>;
+    return result;
+  }
+
+  override 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 new Observable(observer => {
+      this.aggregate(table, query, field, options)
+        .then(data => {
+          observer.next(data);
+          observer.complete();
+        })
+        .catch(error => {
+          observer.error(error);
+        });
+    });
+  }
+
+  override deleteByQuery<T extends keyof IndexerSchema>(
+    _table: T,
+    _query: Query<T>
+  ): Promise<void> {
+    return Promise.resolve();
+  }
+
+  override insert<T extends keyof IndexerSchema>(
+    _table: T,
+    _document: IndexerDocument<T>
+  ): Promise<void> {
+    return Promise.resolve();
+  }
+
+  override delete<T extends keyof IndexerSchema>(
+    _table: T,
+    _id: string
+  ): Promise<void> {
+    return Promise.resolve();
+  }
+
+  override update<T extends keyof IndexerSchema>(
+    _table: T,
+    _document: IndexerDocument<T>
+  ): Promise<void> {
+    return Promise.resolve();
+  }
+
+  override refresh<T extends keyof IndexerSchema>(_table: T): Promise<void> {
+    return Promise.resolve();
+  }
+}