feat: multi-value filters for bulk message cancel + multi-label schedules

src/client/client.test.ts

@@ -7,7 +7,7 @@ import { Client } from "./client";
 import type { PublishToUrlResponse } from "../../dist";
 import { MOCK_QSTASH_SERVER_URL, mockQStashServer } from "./workflow/test-utils";
 import type { HttpClient } from "./http";
-import { eventually } from "./logs.test";
+import { eventually } from "./test-utils";
 
 export const clearQueues = async (client: Client) => {
   const queueDetails = await client.queue().list();

src/client/dlq.test.ts

@@ -5,7 +5,7 @@
 import { sleep } from "bun";
 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
 import { Client } from "./client";
-import { eventually } from "./logs.test";
+import { eventually } from "./test-utils";
 import { MOCK_QSTASH_SERVER_URL, mockQStashServer, expectToReject } from "./workflow/test-utils";
 import type { HttpClient } from "./http";
 
@@ -164,7 +164,7 @@ describe("DLQ", () => {
       const retryDelay = "2000 * retried";
       const randomKey = `flow-control-key-${Date.now()}`;
       const { messageId } = await client.publish({
-        url: "https://httpbin.org/status/400",
+        url: `https://example.com/${randomKey}`,
         body: "hello",
         retries: 0,
         flowControl: {
@@ -176,25 +176,28 @@ describe("DLQ", () => {
         retryDelay,
       });
 
-      await eventually(async () => {
-        const result = await client.dlq.listMessages({
-          filter: {
-            messageId,
-          },
-        });
-        expect(result.messages.length).toBe(1);
-        const message = result.messages[0];
-
-        expect(message.flowControlKey).toBe(randomKey);
-        expect(message.parallelism).toBe(parallelism);
-        expect(message.ratePerSecond).toBe(ratePerSecond);
-        expect(message.rate).toBe(ratePerSecond);
-        expect(message.period).toBe(SECONDS_IN_A_DAY);
-        expect(message.retryDelayExpression).toBe(retryDelay);
-      });
+      await eventually(
+        async () => {
+          const result = await client.dlq.listMessages({
+            filter: {
+              messageId,
+            },
+          });
+          expect(result.messages.length).toBe(1);
+          const message = result.messages[0];
+
+          expect(message.flowControlKey).toBe(randomKey);
+          expect(message.parallelism).toBe(parallelism);
+          expect(message.ratePerSecond).toBe(ratePerSecond);
+          expect(message.rate).toBe(ratePerSecond);
+          expect(message.period).toBe(SECONDS_IN_A_DAY);
+          expect(message.retryDelayExpression).toBe(retryDelay);
+        },
+        { timeout: 25_000, interval: 1000 }
+      );
     },
     {
-      timeout: 10_000,
+      timeout: 30_000,
     }
   );
 
@@ -203,25 +206,28 @@ describe("DLQ", () => {
     async () => {
       const testLabel = `dlq-test-label-${Date.now()}`;
       await client.publish({
-        url: `https://httpbin.org/status/400`, // Any broken link will work
+        url: `https://example.com/${testLabel}`, // Any broken link will work
         retries: 0,
         label: testLabel,
       });
 
-      await sleep(4000);
+      await eventually(
+        async () => {
+          const dlqLogs = await client.dlq.listMessages({
+            filter: {
+              label: testLabel,
+            },
+          });
 
-      const dlqLogs = await client.dlq.listMessages({
-        filter: {
-          label: testLabel,
+          expect(dlqLogs.messages.length).toBeGreaterThanOrEqual(1);
+          for (const message of dlqLogs.messages) {
+            if (message.label !== undefined) {
+              expect(message.label).toBe(testLabel);
+            }
+          }
         },
-      });
-
-      expect(dlqLogs.messages.length).toBeGreaterThanOrEqual(1);
-      for (const message of dlqLogs.messages) {
-        if (message.label !== undefined) {
-          expect(message.label).toBe(testLabel);
-        }
-      }
+        { timeout: 15_000, interval: 1000 }
+      );
     },
     { timeout: 20_000 }
   );
@@ -233,23 +239,26 @@ describe("DLQ", () => {
       const labelTwo = `dlq-multi-b-${Date.now()}`;
 
       const { messageId } = await client.publish({
-        url: `https://httpbin.org/status/400`, // Any broken link will work
+        url: `https://example.com/${labelOne}`, // Any broken link will work
         retries: 0,
         label: [labelOne, labelTwo],
       });
 
-      await sleep(4000);
-
-      const dlqLogs = await client.dlq.listMessages({
-        filter: { label: [labelOne, labelTwo] },
-      });
+      await eventually(
+        async () => {
+          const dlqLogs = await client.dlq.listMessages({
+            filter: { label: [labelOne, labelTwo] },
+          });
 
-      const message = dlqLogs.messages.find((m) => m.messageId === messageId);
-      expect(message).toBeDefined();
-      // legacy `label` carries only the first label
-      expect(message!.label).toBe(labelOne);
-      // new `labels` carries all of them
-      expect(message!.labels).toEqual([labelOne, labelTwo]);
+          const message = dlqLogs.messages.find((m) => m.messageId === messageId);
+          expect(message).toBeDefined();
+          // legacy `label` carries only the first label
+          expect(message!.label).toBe(labelOne);
+          // new `labels` carries all of them
+          expect(message!.labels).toEqual([labelOne, labelTwo]);
+        },
+        { timeout: 15_000, interval: 1000 }
+      );
 
       await client.dlq.delete({ filter: { label: [labelOne, labelTwo] } });
     },
@@ -265,22 +274,32 @@ describe("DLQ", () => {
 
       // msg1: [A, B], msg2: [B, C], msg3: [C]
       const { messageId: messageAB } = await client.publish({
-        url: `https://httpbin.org/status/400`,
+        url: `https://example.com/${labelA}`,
         retries: 0,
         label: [labelA, labelB],
       });
       const { messageId: messageBC } = await client.publish({
-        url: `https://httpbin.org/status/400`,
+        url: `https://example.com/${labelB}`,
         retries: 0,
         label: [labelB, labelC],
       });
       const { messageId: messageC } = await client.publish({
-        url: `https://httpbin.org/status/400`,
+        url: `https://example.com/${labelC}`,
         retries: 0,
         label: labelC,
       });
 
-      await sleep(4000);
+      // wait for all three to land in the DLQ
+      await eventually(
+        async () => {
+          const all = await client.dlq.listMessages({
+            filter: { label: [labelA, labelB, labelC] },
+          });
+          const allIds = new Set(all.messages.map((m) => m.messageId));
+          expect(allIds.has(messageAB) && allIds.has(messageBC) && allIds.has(messageC)).toBe(true);
+        },
+        { timeout: 15_000, interval: 1000 }
+      );
 
       // filtering by [A, B] should match msgAB and msgBC (both share a label)
       // but NOT msgC.
@@ -297,6 +316,40 @@ describe("DLQ", () => {
     { timeout: 20_000 }
   );
 
+  test(
+    "should filter DLQ messages by multiple urls (OR semantics)",
+    async () => {
+      const stamp = Date.now();
+      const urlA = `https://example.com/dlq-url-a-${stamp}`;
+      const urlB = `https://example.com/dlq-url-b-${stamp}`;
+      const urlC = `https://example.com/dlq-url-c-${stamp}`;
+
+      const { messageId: idA } = await client.publish({ url: urlA, retries: 0 });
+      const { messageId: idB } = await client.publish({ url: urlB, retries: 0 });
+      const { messageId: idC } = await client.publish({ url: urlC, retries: 0 });
+
+      // wait for all three to land in the DLQ
+      await eventually(
+        async () => {
+          const all = await client.dlq.listMessages({ filter: { url: [urlA, urlB, urlC] } });
+          const ids = new Set(all.messages.map((m) => m.messageId));
+          expect(ids.has(idA) && ids.has(idB) && ids.has(idC)).toBe(true);
+        },
+        { timeout: 15_000, interval: 1000 }
+      );
+
+      // filtering by [A, B] should match A and B but NOT C.
+      const result = await client.dlq.listMessages({ filter: { url: [urlA, urlB] } });
+      const ids = new Set(result.messages.map((m) => m.messageId));
+      expect(ids.has(idA)).toBe(true);
+      expect(ids.has(idB)).toBe(true);
+      expect(ids.has(idC)).toBe(false);
+
+      await client.dlq.delete({ filter: { url: [urlA, urlB, urlC] } });
+    },
+    { timeout: 25_000 }
+  );
+
   test(
     "should retry multiple messages from DLQ",
     async () => {
@@ -761,7 +814,7 @@ describe("DLQ", () => {
     async () => {
       const flowKey = `dlq-flow-key-${Date.now()}`;
       await client.publish({
-        url: "https://httpbin.org/status/400",
+        url: `https://example.com/${flowKey}`,
         body: "hello",
         retries: 0,
         flowControl: {
@@ -1039,4 +1092,29 @@ describe("DLQ - mocked filter url shape", () => {
       },
     });
   });
+
+  test("should send multi-value url and flowControlKey as repeated query params", async () => {
+    const mockClient = new Client({ token, baseUrl: MOCK_QSTASH_SERVER_URL });
+    await mockQStashServer({
+      execute: async () => {
+        await mockClient.dlq.listMessages({
+          filter: {
+            url: ["https://a.com", "https://b.com"],
+            flowControlKey: ["key-1", "key-2"],
+          },
+        });
+      },
+      responseFields: { body: { messages: [] }, status: 200 },
+      receivesRequest: {
+        method: "GET",
+        token,
+        url: `${MOCK_QSTASH_SERVER_URL}/v2/dlq?url=${encodeURIComponent("https://a.com")}&url=${encodeURIComponent("https://b.com")}&flowControlKey=key-1&flowControlKey=key-2`,
+      },
+      validateRequest: (request) => {
+        const parameters = new URL(request.url).searchParams;
+        expect(parameters.getAll("url")).toEqual(["https://a.com", "https://b.com"]);
+        expect(parameters.getAll("flowControlKey")).toEqual(["key-1", "key-2"]);
+      },
+    });
+  });
 });

src/client/filter-types.ts

@@ -11,11 +11,18 @@ type Exclusive<A, B> = (A & NeverKeys<B>) | (B & NeverKeys<A>);
 
 // ── 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.
    *
@@ -24,16 +31,26 @@ 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[];
 };
 
-/** QStash-specific identity filters (DLQ + message endpoints). */
+/**
+ * QStash-specific identity filters (DLQ + message endpoints).
+ *
+ * Every field except `messageId` supports multi-value filtering: pass an array
+ * to match a record whose value equals any of the given values (OR semantics).
+ */
 type QStashIdentityFields = {
   messageId?: string;
-  url?: string;
-  urlGroup?: string;
-  scheduleId?: string;
-  queueName?: string;
+  /** Filter by destination URL. Supports multiple values. */
+  url?: string | string[];
+  /** Filter by URL Group name. Supports multiple values. */
+  urlGroup?: string | string[];
+  /** Filter by Schedule ID. Supports multiple values. */
+  scheduleId?: string | string[];
+  /** Filter by Queue name. Supports multiple values. */
+  queueName?: string | string[];
 };
 
 /** DLQ-specific response filter. */
@@ -46,6 +63,19 @@ type LogsFilterFields = {
   state?: State;
 };
 
+/**
+ * Filter by the host/path of the destination URL.
+ *
+ * Supported by the message 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 destination URL. Supports multiple values. */
+  host?: string | string[];
+  /** Filter by the path of the destination URL. Supports multiple values. */
+  path?: string | string[];
+};
+
 // ── Composed Filter Field Types ───────────────────────────────
 
 type DLQFilterFields = UniversalFilterFields &
@@ -57,7 +87,15 @@ type DLQFilterFields = UniversalFilterFields &
     api?: string;
   };
 
-type MessageCancelFilterFields = UniversalFilterFields & Omit<QStashIdentityFields, "messageId">;
+/**
+ * Filter fields for the bulk message cancel endpoint.
+ *
+ * Inherits the shared multi-value fields plus the `host`/`path` destination
+ * filters (also supported by logs).
+ */
+type MessageCancelFilterFields = UniversalFilterFields &
+  Omit<QStashIdentityFields, "messageId"> &
+  HostPathFilterFields;
 
 // ── QStash Composed Endpoint Filter Types ─────────────────────
 
@@ -131,6 +169,7 @@ export type LogsListRequest = Exclusive<
 
 export type LogsListFilters = UniversalFilterFields &
   Omit<QStashIdentityFields, "messageId"> &
+  HostPathFilterFields &
   LogsFilterFields & {
     /**
      * @deprecated `api` filter has been removed from the API and will be ignored