feat: multi-value filters for bulk workflow run cancel

src/client/dlq.test.ts

@@ -120,6 +120,31 @@ describe("DLQ", () => {
       });
     });
 
+    test("should list DLQ messages with multi-value workflowUrl/callerIp/flowControlKey", async () => {
+      const urlA = `${MOCK_DESTINATION_HOST}/a`;
+      const urlB = `${MOCK_DESTINATION_HOST}/b`;
+      await mockQStashServer({
+        execute: async () => {
+          await client.dlq.list({
+            filter: {
+              workflowUrl: [urlA, urlB],
+              callerIp: ["1.2.3.4", "5.6.7.8"],
+              flowControlKey: ["key-1", "key-2"],
+            },
+          });
+        },
+        responseFields: {
+          status: 200,
+          body: { messages: [], cursor: undefined },
+        },
+        receivesRequest: {
+          method: "GET",
+          url: `${MOCK_QSTASH_SERVER_URL}/v2/dlq?workflowUrl=${encodeURIComponent(urlA)}&workflowUrl=${encodeURIComponent(urlB)}&callerIp=1.2.3.4&callerIp=5.6.7.8&flowControlKey=key-1&flowControlKey=key-2&source=workflow`,
+          token,
+        },
+      });
+    });
+
     test("should surface both label and labels from the response", async () => {
       const labelOne = "label-one";
       const labelTwo = "label-two";

src/client/filter-types.ts

@@ -6,11 +6,18 @@ type RequireAtLeastOne<T> = { [K in keyof T]-?: Required<Pick<T, K>> }[keyof T];
 
 // ── Filter Field Groups ───────────────────────────────────────
 
-/** Shared filter fields accepted by every qstash & workflow endpoint. */
+/**
+ * Shared filter fields accepted by every qstash & workflow endpoint.
+ *
+ * Most fields support multi-value filtering: pass an array to match a record
+ * whose value equals any of the given values (OR semantics). Separate filters
+ * are combined with AND semantics.
+ */
 type UniversalFilterFields = {
   fromDate?: Date | number;
   toDate?: Date | number;
-  callerIp?: string;
+  /** Filter by the IP address of the publisher. Supports multiple values. */
+  callerIp?: string | string[];
   /**
    * Filter by label.
    *
@@ -19,12 +26,14 @@ type UniversalFilterFields = {
    * filtering by `[label_1, label_2]` returns both.
    */
   label?: string | string[];
-  flowControlKey?: string;
+  /** Filter by Flow Control Key. Supports multiple values. */
+  flowControlKey?: string | string[];
 };
 
 /** Workflow-specific filter fields for DLQ and bulk endpoints. */
 type WorkflowFilterFields = {
-  workflowUrl?: string;
+  /** Filter by workflow URL. Supports multiple values. */
+  workflowUrl?: string | string[];
   workflowRunId?: string;
   workflowCreatedAt?: number;
   failureFunctionState?: string;
@@ -35,24 +44,41 @@ type WorkflowLogsFilterFields = {
   messageId?: string;
 };
 
+/**
+ * Filter by the host/path of the workflow URL.
+ *
+ * Supported by the cancel and logs endpoints (the DLQ endpoint rejects these).
+ * Each supports multiple values: pass an array to match any value.
+ */
+type HostPathFilterFields = {
+  /** Filter by the host of the workflow URL. Supports multiple values. */
+  host?: string | string[];
+  /** Filter by the path of the workflow URL. Supports multiple values. */
+  path?: string | string[];
+};
+
 // ── Composed Filter Field Types ───────────────────────────────
 
 type DLQActionFilterFields = UniversalFilterFields & WorkflowFilterFields;
 
 /** Cancel filter: exact URL match. Cannot combine with `workflowUrlStartingWith`. */
-type CancelFilterWithExactUrl = UniversalFilterFields & {
-  workflowUrl: string;
-  workflowUrlStartingWith?: never;
-};
+type CancelFilterWithExactUrl = UniversalFilterFields &
+  HostPathFilterFields & {
+    /** Filter by exact workflow URL. Supports multiple values. */
+    workflowUrl: string | string[];
+    workflowUrlStartingWith?: never;
+  };
 
 /** Cancel filter: URL prefix match. Cannot combine with `workflowUrl`. */
-type CancelFilterWithPrefixUrl = UniversalFilterFields & {
-  workflowUrlStartingWith: string;
-  workflowUrl?: never;
-};
+type CancelFilterWithPrefixUrl = UniversalFilterFields &
+  HostPathFilterFields & {
+    /** Filter by workflow URL prefix. Supports multiple values. */
+    workflowUrlStartingWith: string | string[];
+    workflowUrl?: never;
+  };
 
 /** Cancel filter: no URL. Requires at least one other filter field. */
-type CancelFilterWithoutUrl = RequireAtLeastOne<UniversalFilterFields> & {
+type CancelFilterWithoutUrl = RequireAtLeastOne<UniversalFilterFields & HostPathFilterFields> & {
   workflowUrl?: never;
   workflowUrlStartingWith?: never;
 };
@@ -130,4 +156,5 @@ export type WorkflowRunCancelFilters =
 
 export type WorkflowLogsListFilters = UniversalFilterFields &
   Pick<WorkflowFilterFields, "workflowUrl" | "workflowRunId" | "workflowCreatedAt"> &
+  HostPathFilterFields &
   WorkflowLogsFilterFields;

src/client/index.test.ts

@@ -133,6 +133,104 @@ describe("workflow client", () => {
       });
     });
 
+    test("should cancel with multiple workflowUrls (exact match)", async () => {
+      const url1 = "https://a.workflow-endpoint.com";
+      const url2 = "https://b.workflow-endpoint.com";
+      await mockQStashServer({
+        execute: async () => {
+          const result = await client.cancel({ filter: { workflowUrl: [url1, url2] } });
+          expect(result).toEqual({ cancelled: 4 });
+        },
+        responseFields: {
+          status: 200,
+          body: { cancelled: 4 },
+        },
+        receivesRequest: {
+          method: "DELETE",
+          url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs?workflowUrl=${encodeURIComponent(url1)}&workflowUrl=${encodeURIComponent(url2)}&workflowUrlExactMatch=true&count=100`,
+          token,
+        },
+      });
+    });
+
+    test("should cancel with multiple workflowUrlStartingWith (prefix match)", async () => {
+      const url1 = "https://a.workflow-endpoint.com/path";
+      const url2 = "https://b.workflow-endpoint.com/path";
+      await mockQStashServer({
+        execute: async () => {
+          const result = await client.cancel({
+            filter: { workflowUrlStartingWith: [url1, url2] },
+          });
+          expect(result).toEqual({ cancelled: 2 });
+        },
+        responseFields: {
+          status: 200,
+          body: { cancelled: 2 },
+        },
+        receivesRequest: {
+          method: "DELETE",
+          url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs?workflowUrl=${encodeURIComponent(url1)}&workflowUrl=${encodeURIComponent(url2)}&count=100`,
+          token,
+        },
+      });
+    });
+
+    test("should cancel with multi-value callerIp and flowControlKey filters", async () => {
+      await mockQStashServer({
+        execute: async () => {
+          await client.cancel({
+            filter: {
+              callerIp: ["1.2.3.4", "5.6.7.8"],
+              flowControlKey: ["key-1", "key-2"],
+            },
+          });
+        },
+        responseFields: {
+          status: 200,
+          body: { cancelled: 3 },
+        },
+        receivesRequest: {
+          method: "DELETE",
+          url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs?callerIp=1.2.3.4&callerIp=5.6.7.8&flowControlKey=key-1&flowControlKey=key-2&count=100`,
+          token,
+        },
+      });
+    });
+
+    test("should cancel with multiple labels (OR semantics)", async () => {
+      await mockQStashServer({
+        execute: async () => {
+          await client.cancel({ filter: { label: ["label-1", "label-2"] } });
+        },
+        responseFields: {
+          status: 200,
+          body: { cancelled: 2 },
+        },
+        receivesRequest: {
+          method: "DELETE",
+          url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs?label=label-1&label=label-2&count=100`,
+          token,
+        },
+      });
+    });
+
+    test("should cancel with host and path filters", async () => {
+      await mockQStashServer({
+        execute: async () => {
+          await client.cancel({ filter: { host: ["a.com", "b.com"], path: "/webhook" } });
+        },
+        responseFields: {
+          status: 200,
+          body: { cancelled: 2 },
+        },
+        receivesRequest: {
+          method: "DELETE",
+          url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs?host=a.com&host=b.com&path=%2Fwebhook&count=100`,
+          token,
+        },
+      });
+    });
+
     test("should cancel all", async () => {
       await mockQStashServer({
         execute: async () => {
@@ -560,6 +658,30 @@ describe("workflow client", () => {
       }
     );
 
+    test(
+      "should cancel with multiple workflowUrls (exact match, OR semantics)",
+      async () => {
+        const urlA = `${MOCK_DESTINATION_HOST}/multi-exact-a-${nanoid()}`;
+        const urlB = `${MOCK_DESTINATION_HOST}/multi-exact-b-${nanoid()}`;
+        const urlC = `${MOCK_DESTINATION_HOST}/multi-exact-c-${nanoid()}`;
+
+        await liveClient.trigger({ url: urlA, delay: "1m" });
+        await liveClient.trigger({ url: urlB, delay: "1m" });
+        const { workflowRunId: idC } = await liveClient.trigger({ url: urlC, delay: "1m" });
+
+        // Cancelling [urlA, urlB] with exact match cancels exactly those two, not urlC.
+        const cancel = await liveClient.cancel({ filter: { workflowUrl: [urlA, urlB] } });
+        expect(cancel).toEqual({ cancelled: 2 });
+
+        // clean up the untouched run
+        const cleanup = await liveClient.cancel(idC);
+        expect(cleanup).toEqual({ cancelled: 1 });
+      },
+      {
+        timeout: 15000,
+      }
+    );
+
     test(
       "should cancel with combined filters (label + workflowUrl)",
       async () => {
@@ -591,6 +713,28 @@ describe("workflow client", () => {
       }
     );
 
+    test(
+      "should cancel by destination host and path",
+      async () => {
+        const comPath = `/cancel-host-com-${nanoid()}`;
+        const orgPath = `/cancel-host-org-${nanoid()}`;
+
+        await liveClient.trigger({ url: `https://example.com${comPath}`, delay: "1m" });
+        await liveClient.trigger({ url: `https://example.org${orgPath}`, delay: "1m" });
+
+        // host example.org must not touch the example.com run
+        const byHost = await liveClient.cancel({ filter: { host: "example.org" } });
+        expect(byHost).toEqual({ cancelled: 1 });
+
+        // the example.com run survived — cancel it via its unique path
+        const byPath = await liveClient.cancel({ filter: { path: comPath } });
+        expect(byPath).toEqual({ cancelled: 1 });
+      },
+      {
+        timeout: 15000,
+      }
+    );
+
     test.skip(
       "should cancel all",
       async () => {
@@ -1234,6 +1378,91 @@ describe("workflow client", () => {
       { timeout: 20000 }
     );
 
+    test(
+      "should filter logs by multiple workflowUrls (OR) - live",
+      async () => {
+        const liveClient = new Client({
+          baseUrl: process.env.QSTASH_URL,
+          token: process.env.QSTASH_TOKEN!,
+        });
+
+        // unique urls so the filter only matches runs from this test
+        const urlA = `${MOCK_DESTINATION_HOST}/log-url-a-${nanoid()}`;
+        const urlB = `${MOCK_DESTINATION_HOST}/log-url-b-${nanoid()}`;
+        const urlC = `${MOCK_DESTINATION_HOST}/log-url-c-${nanoid()}`;
+
+        const { workflowRunId: runA } = await liveClient.trigger({ url: urlA, delay: "1m" });
+        const { workflowRunId: runB } = await liveClient.trigger({ url: urlB, delay: "1m" });
+        const { workflowRunId: runC } = await liveClient.trigger({ url: urlC, delay: "1m" });
+
+        try {
+          // filtering by [urlA, urlB] should return runA and runB but NOT runC.
+          await eventually(
+            async () => {
+              const logs = await liveClient.logs({ filter: { workflowUrl: [urlA, urlB] } });
+              const ids = logs.runs.map((r) => r.workflowRunId);
+              expect(ids).toContain(runA);
+              expect(ids).toContain(runB);
+              expect(ids).not.toContain(runC);
+            },
+            { timeout: 5000, interval: 250 }
+          );
+        } finally {
+          await liveClient.cancel([runA, runB, runC]).catch(() => {});
+        }
+      },
+      { timeout: 20000 }
+    );
+
+    test(
+      "should filter logs by destination host and path - live",
+      async () => {
+        const liveClient = new Client({
+          baseUrl: process.env.QSTASH_URL,
+          token: process.env.QSTASH_TOKEN!,
+        });
+
+        const comPath = `/log-host-com-${nanoid()}`;
+        const orgPath = `/log-host-org-${nanoid()}`;
+
+        const { workflowRunId: comRun } = await liveClient.trigger({
+          url: `https://example.com${comPath}`,
+          delay: "1m",
+        });
+        const { workflowRunId: orgRun } = await liveClient.trigger({
+          url: `https://example.org${orgPath}`,
+          delay: "1m",
+        });
+
+        try {
+          // host discriminates: example.org excludes the example.com run
+          await eventually(
+            async () => {
+              const logs = await liveClient.logs({ filter: { host: "example.org" } });
+              const ids = logs.runs.map((r) => r.workflowRunId);
+              expect(ids).toContain(orgRun);
+              expect(ids).not.toContain(comRun);
+            },
+            { timeout: 10000, interval: 500 }
+          );
+
+          // path discriminates: the unique example.com path excludes the example.org run
+          await eventually(
+            async () => {
+              const logs = await liveClient.logs({ filter: { path: comPath } });
+              const ids = logs.runs.map((r) => r.workflowRunId);
+              expect(ids).toContain(comRun);
+              expect(ids).not.toContain(orgRun);
+            },
+            { timeout: 10000, interval: 500 }
+          );
+        } finally {
+          await liveClient.cancel([comRun, orgRun]).catch(() => {});
+        }
+      },
+      { timeout: 25000 }
+    );
+
     test.skip(
       "should get logs - live",
       async () => {

src/client/index.ts

@@ -43,6 +43,11 @@ export class Client {
    * - With other filters: `cancel({ filter: { label: "my-label" } })`
    * - To target all: `cancel({ all: true })`
    *
+   * Filters support multiple values: pass an array to match a run whose value
+   * equals any of the given values (OR logic). Separate filters are combined with
+   * AND logic. For example:
+   * `cancel({ filter: { workflowUrl: ["https://a.com", "https://b.com"] } })`
+   *
    * Cancels up to `count` workflow runs per call (defaults to 100).
    *
    * ```ts