Allow server-side filtering for readAll method

packages/db-client/src/streams/readAll.ts

@@ -10,9 +10,16 @@ import type {
   Direction,
   AllStreamResolvedEvent,
   StreamingRead,
+  Filter,
 } from "../types";
 import { debug, convertGrpcEvent } from "../utils";
-import { BACKWARDS, FORWARDS, START } from "../constants";
+import {
+  BACKWARDS,
+  EVENT_TYPE,
+  FORWARDS,
+  START,
+  STREAM_NAME,
+} from "../constants";
 import { Client } from "../Client";
 
 import { ReadStream } from "./utils/ReadStream";
@@ -40,6 +47,10 @@ export interface ReadAllOptions extends BaseOptions {
    * @defaultValue FORWARDS
    */
   direction?: Direction;
+  /**
+   * Filters events or streams based upon a predicate.
+   */
+  filter?: Filter;
 }
 
 declare module "../Client" {
@@ -63,6 +74,7 @@ Client.prototype.readAll = function (
     fromPosition = START,
     resolveLinkTos = false,
     direction = FORWARDS,
+    filter,
     ...baseOptions
   }: ReadAllOptions = {},
   readableOptions: ReadableOptions = {}
@@ -112,6 +124,50 @@ Client.prototype.readAll = function (
     }
   }
 
+  if (filter) {
+    const expr = new ReadReq.Options.FilterOptions.Expression();
+
+    if ("prefixes" in filter) {
+      expr.setPrefixList(filter.prefixes);
+    }
+
+    if ("regex" in filter) {
+      expr.setRegex(filter.regex);
+    }
+
+    const filterOptions = new ReadReq.Options.FilterOptions();
+
+    switch (filter.filterOn) {
+      case STREAM_NAME: {
+        filterOptions.setStreamIdentifier(expr);
+        break;
+      }
+      case EVENT_TYPE: {
+        filterOptions.setEventType(expr);
+        break;
+      }
+    }
+
+    if (typeof filter.maxSearchWindow === "number") {
+      if (filter.maxSearchWindow <= 0) {
+        throw new Error("MaxSearchWindow must be greater than 0.");
+      }
+      filterOptions.setMax(filter.maxSearchWindow);
+    } else {
+      filterOptions.setCount(new Empty());
+    }
+
+    if (filter.checkpointInterval <= 0) {
+      throw new Error("CheckpointInterval must be greater than 0.");
+    }
+
+    filterOptions.setCheckpointintervalmultiplier(filter.checkpointInterval);
+
+    options.setFilter(filterOptions);
+  } else {
+    options.setNoFilter(new Empty());
+  }
+
   req.setOptions(options);
 
   debug.command("readAll: %O", {

packages/test/src/streams/readAll.test.ts

@@ -49,6 +49,29 @@ describe("readAll", () => {
       expect(notSystemStreams[0]).toBe(STREAM_NAME_A);
     });
 
+    test("with filter", async () => {
+      let count = 0;
+      const notSystemStreams = [];
+
+      for await (const { event } of client.readAll({
+        filter: {
+          prefixes: [STREAM_NAME_A],
+          filterOn: "streamName",
+          checkpointInterval: 1000,
+        },
+      })) {
+        count++;
+
+        if (event && !event.streamId.startsWith("$")) {
+          notSystemStreams.push(event.streamId);
+        }
+      }
+
+      expect(count).toEqual(4);
+      expect(notSystemStreams.length).toEqual(4);
+      expect(Array.from(new Set(notSystemStreams))).toEqual([STREAM_NAME_A]);
+    });
+
     test("from position", async () => {
       let eventToExtract!: AllStreamResolvedEvent;