diff --git a/src/client/dlq.test.ts b/src/client/dlq.test.ts index 402e165..cac49c4 100644 --- a/src/client/dlq.test.ts +++ b/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"; @@ -312,6 +337,18 @@ describe("DLQ", () => { }); }); + test("should throw when a filter field is an empty array", async () => { + await mockQStashServer({ + execute: async () => { + await expect(client.dlq.resume({ filter: { workflowUrl: [] } })).rejects.toThrow( + "Empty array provided for filter field 'workflowUrl'" + ); + }, + responseFields: { status: 200, body: {} }, + receivesRequest: false, + }); + }); + test("should resume DLQ messages with filters", async () => { const responses = [ { workflowRunId: `wfr-${nanoid()}`, workflowCreatedAt: "2023-01-01T00:00:00Z" }, diff --git a/src/client/filter-types.ts b/src/client/filter-types.ts index be933f1..a8e3640 100644 --- a/src/client/filter-types.ts +++ b/src/client/filter-types.ts @@ -2,15 +2,24 @@ import { WorkflowRunLog } from "./types"; -type RequireAtLeastOne = { [K in keyof T]-?: Required> }[keyof T]; +// Intersecting with T keeps every field visible to autocomplete/inference; +// the single-property union alone only enforces the "at least one" constraint. +type RequireAtLeastOne = T & { [K in keyof T]-?: Required> }[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 +28,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 +46,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 & { +type CancelFilterWithoutUrl = RequireAtLeastOne & { workflowUrl?: never; workflowUrlStartingWith?: never; }; @@ -130,4 +158,5 @@ export type WorkflowRunCancelFilters = export type WorkflowLogsListFilters = UniversalFilterFields & Pick & + HostPathFilterFields & WorkflowLogsFilterFields; diff --git a/src/client/index.test.ts b/src/client/index.test.ts index 765606f..ece690d 100644 --- a/src/client/index.test.ts +++ b/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 () => { @@ -172,6 +270,24 @@ describe("workflow client", () => { }); }); + test("should throw when a filter field is an empty array", async () => { + await mockQStashServer({ + execute: async () => { + await expect(client.cancel({ filter: { workflowUrl: [] } })).rejects.toThrow( + "Empty array provided for filter field 'workflowUrl'" + ); + await expect(client.cancel({ filter: { workflowUrlStartingWith: [] } })).rejects.toThrow( + "Empty array provided for filter field 'workflowUrlStartingWith'" + ); + await expect(client.cancel({ filter: { callerIp: [] } })).rejects.toThrow( + "Empty array provided for filter field 'callerIp'" + ); + }, + responseFields: { status: 200, body: {} }, + receivesRequest: false, + }); + }); + test("should not send request when an array contains an empty string", async () => { await mockQStashServer({ execute: async () => { @@ -560,6 +676,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 +731,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 +1396,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 () => { diff --git a/src/client/index.ts b/src/client/index.ts index 0e1f717..34b4087 100644 --- a/src/client/index.ts +++ b/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 diff --git a/src/client/utils.ts b/src/client/utils.ts index c4f0bb9..9559b4a 100644 --- a/src/client/utils.ts +++ b/src/client/utils.ts @@ -217,7 +217,8 @@ const DEFAULT_BULK_COUNT = 100; * // => { all: true, count: 100 } * ``` * - * @throws {QstashError} If an empty `dlqIds` or `workflowRunIds` array is provided + * @throws {QstashError} If an empty `dlqIds` or `workflowRunIds` array is provided, + * or if any filter field is an empty array */ export function buildBulkActionQueryParameters( request: WorkflowDLQActionFilters | WorkflowRunCancelFilters, @@ -257,6 +258,17 @@ export function buildBulkActionQueryParameters( ); } + // An empty array serializes to no query parameter at all, silently widening + // the scope of a destructive bulk action. Fail fast instead. + for (const [field, value] of Object.entries(filter)) { + if (Array.isArray(value) && value.length === 0) { + throw new QstashError( + `Empty array provided for filter field '${field}'. ` + + "If you intend to target all records, use { all: true } explicitly." + ); + } + } + // When translateWorkflowUrl is set (cancel filters), translate // workflowUrl/workflowUrlStartingWith into the server's query params: // - workflowUrl → workflowUrl + workflowUrlExactMatch=true (exact match) @@ -271,11 +283,11 @@ export function buildBulkActionQueryParameters( ); } - const urlParams: Record = {}; + const urlParams: Record = {}; if (workflowUrlStartingWith) { - urlParams.workflowUrl = workflowUrlStartingWith as string; + urlParams.workflowUrl = workflowUrlStartingWith as string | string[]; } else if (workflowUrl) { - urlParams.workflowUrl = workflowUrl as string; + urlParams.workflowUrl = workflowUrl as string | string[]; urlParams.workflowUrlExactMatch = true; }