feat: support filtered subscriptions

packages/runtime/src/services/documents/documents.ts

@@ -1,4 +1,4 @@
-import { QueryParser } from '@morten-olsen/stash-query-dsl';
+import { isMatch, QueryParser, type QueryFilter } from '@morten-olsen/stash-query-dsl';
 
 import { DatabaseService, tableNames, type TableRows } from '../database/database.js';
 import { SplittingService } from '../splitter/splitter.js';
@@ -24,10 +24,16 @@ type DocumentsServiceEvents = {
   deleted: (document: Document) => void;
 };
 
+type DocumentServiceFilterSubscriber = {
+  filter?: QueryFilter | string;
+  fn: (document: Document) => void;
+  abortSignal?: AbortSignal;
+};
+
 class DocumentsService extends EventEmitter<DocumentsServiceEvents> {
   #services: Services;
   #subscribeListenAbortController: AbortController;
-  #databaseListenAbortController?: AbortController;
+  #databaseListenAbortController?: Promise<AbortController>;
 
   constructor(services: Services) {
     super();
@@ -43,29 +49,8 @@ class DocumentsService extends EventEmitter<DocumentsServiceEvents> {
     this.#services = services;
   }
 
-  public find = async (filter: DocumentFilter): Promise<DocumentFindResult> => {
+  #setupListen = async () => {
-    const databaseService = this.#services.get(DatabaseService);
+    const abortController = new AbortController();
-    const db = await databaseService.getInstance();
-    let query = db<TableRows['documents']>(tableNames.documents);
-    if (filter) {
-      const parser = this.#services.get(QueryParser);
-      query = applyQueryFilter(
-        query,
-        typeof filter.condition === 'string' ? parser.parse(filter.condition) : filter.condition,
-      );
-    }
-    query = query.limit(filter.limit).offset(filter.offset);
-    const items = await query;
-    return {
-      items: items.map(mapFromDocumentRow),
-    };
-  };
-
-  #listen = async () => {
-    if (this.#databaseListenAbortController) {
-      return;
-    }
-    this.#databaseListenAbortController = new AbortController();
     const databaseService = this.#services.get(DatabaseService);
     await databaseService.listen();
     databaseService.on(
@@ -90,8 +75,56 @@ class DocumentsService extends EventEmitter<DocumentsServiceEvents> {
           this.emit('deleted', oldDocument);
         }
       },
-      { abortSignal: this.#databaseListenAbortController.signal },
+      { abortSignal: abortController.signal },
     );
+    return abortController;
+  };
+
+  #listen = async () => {
+    if (!this.#databaseListenAbortController) {
+      this.#databaseListenAbortController = this.#setupListen();
+    }
+    return this.#databaseListenAbortController;
+  };
+
+  public subscribe = async (options: DocumentServiceFilterSubscriber) => {
+    const abortController = new AbortController();
+    const queryParser = this.#services.get(QueryParser);
+    const filter = typeof options.filter === 'string' ? queryParser.parse(options.filter) : options.filter;
+
+    this.on(
+      'upserted',
+      (document) => {
+        if (filter && !isMatch(document, filter)) {
+          return;
+        }
+        options.fn(document);
+      },
+      { abortSignal: abortController.signal },
+    );
+
+    options.abortSignal?.addEventListener('abort', () => abortController.abort());
+
+    await this.#listen();
+    return () => abortController.abort();
+  };
+
+  public find = async (filter: DocumentFilter): Promise<DocumentFindResult> => {
+    const databaseService = this.#services.get(DatabaseService);
+    const db = await databaseService.getInstance();
+    let query = db<TableRows['documents']>(tableNames.documents);
+    if (filter) {
+      const parser = this.#services.get(QueryParser);
+      query = applyQueryFilter(
+        query,
+        typeof filter.condition === 'string' ? parser.parse(filter.condition) : filter.condition,
+      );
+    }
+    query = query.limit(filter.limit).offset(filter.offset);
+    const items = await query;
+    return {
+      items: items.map(mapFromDocumentRow),
+    };
   };
 
   public get = async (id: string): Promise<Document> => {
@@ -218,7 +251,7 @@ class DocumentsService extends EventEmitter<DocumentsServiceEvents> {
   [destroy] = async () => {
     this.#subscribeListenAbortController.abort();
     if (this.#databaseListenAbortController) {
-      this.#databaseListenAbortController.abort();
+      (await this.#databaseListenAbortController).abort();
     }
   };
 }

packages/server/src/api.ts

@@ -28,8 +28,11 @@ class BaseError extends Error {
 }
 
 const createApi = async (runtime: StashRuntime = new StashRuntime()) => {
-  runtime.documents.on('upserted', (document) => {
+  runtime.documents.subscribe({
-    console.log(document);
+    filter: "metadata.foo = 'bar'",
+    fn: (document) => {
+      console.log(document);
+    },
   });
   const app = fastify().withTypeProvider<ZodTypeProvider>();
   app.setValidatorCompiler(validatorCompiler);