diff --git a/src/client/client.test.ts b/src/client/client.test.ts index eb97d38..20724ed 100644 --- a/src/client/client.test.ts +++ b/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(); diff --git a/src/client/dlq.test.ts b/src/client/dlq.test.ts index 5a3718f..cf96309 100644 --- a/src/client/dlq.test.ts +++ b/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"]); + }, + }); + }); }); diff --git a/src/client/filter-types.ts b/src/client/filter-types.ts index df6caf5..5a94111 100644 --- a/src/client/filter-types.ts +++ b/src/client/filter-types.ts @@ -11,11 +11,18 @@ type Exclusive = (A & NeverKeys) | (B & NeverKeys); // ── 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; +/** + * 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 & + HostPathFilterFields; // ── QStash Composed Endpoint Filter Types ───────────────────── @@ -131,6 +169,7 @@ export type LogsListRequest = Exclusive< export type LogsListFilters = UniversalFilterFields & Omit & + HostPathFilterFields & LogsFilterFields & { /** * @deprecated `api` filter has been removed from the API and will be ignored diff --git a/src/client/flow-control.test.ts b/src/client/flow-control.test.ts index d240983..a91c8f9 100644 --- a/src/client/flow-control.test.ts +++ b/src/client/flow-control.test.ts @@ -4,6 +4,7 @@ import { describe, expect, test } from "bun:test"; import { Client } from "./client"; import { MOCK_QSTASH_SERVER_URL, mockQStashServer, expectToReject } from "./workflow/test-utils"; +import { eventually } from "./test-utils"; describe("FlowControl empty id guard", () => { test("should not send request when get is called with an empty string", async () => { @@ -209,12 +210,20 @@ describe("FlowControl", () => { }, }); - // Reset the rate - await client.flowControl.resetRate(flowControlKey); + // Pause delivery so in-flight dispatches don't keep consuming the rate + // after we reset it (otherwise rateCount creeps back up). + await client.flowControl.pause(flowControlKey); - // Verify rate was reset by checking the flow control info - const info = await client.flowControl.get(flowControlKey); - expect(info.rateCount).toBe(0); + // Reset the rate and verify it was cleared. We reset inside the retry loop + // so the reset takes effect even if the pause/reset are eventually consistent. + await eventually( + async () => { + await client.flowControl.resetRate(flowControlKey); + const info = await client.flowControl.get(flowControlKey); + expect(info.rateCount).toBe(0); + }, + { timeout: 15_000, interval: 1000 } + ); // Clean up // eslint-disable-next-line @typescript-eslint/no-deprecated diff --git a/src/client/logs.test.ts b/src/client/logs.test.ts index cb87527..04064d7 100644 --- a/src/client/logs.test.ts +++ b/src/client/logs.test.ts @@ -3,6 +3,7 @@ import { describe, expect, test } from "bun:test"; import { Client } from "./client"; import { MOCK_QSTASH_SERVER_URL, mockQStashServer } from "./workflow/test-utils"; +import { eventually } from "./test-utils"; describe("logs", () => { test( @@ -101,6 +102,74 @@ describe("logs", () => { { timeout: 15_000 } ); + test( + "should filter logs by multiple urls (OR semantics)", + async () => { + const stamp = Date.now(); + const urlA = `https://example.com/log-url-a-${stamp}`; + const urlB = `https://example.com/log-url-b-${stamp}`; + const urlC = `https://example.com/log-url-c-${stamp}`; + + const { messageId: idA } = await client.publish({ url: urlA, body: "url-or-a" }); + const { messageId: idB } = await client.publish({ url: urlB, body: "url-or-b" }); + const { messageId: idC } = await client.publish({ url: urlC, body: "url-or-c" }); + + // filtering by [A, B] should match A and B but NOT C. + await eventually( + async () => { + const result = await client.logs({ filter: { url: [urlA, urlB] } }); + const ids = new Set(result.logs.map((l) => l.messageId)); + expect(ids.has(idA)).toBe(true); + expect(ids.has(idB)).toBe(true); + expect(ids.has(idC)).toBe(false); + }, + { interval: 1000 } + ); + }, + { timeout: 15_000 } + ); + + test( + "should filter logs by destination host and path", + async () => { + const stamp = Date.now(); + const comPath = `/log-host-com-${stamp}`; + const orgPath = `/log-host-org-${stamp}`; + + const { messageId: comId } = await client.publish({ + url: `https://example.com${comPath}`, + body: "host-com", + }); + const { messageId: orgId } = await client.publish({ + url: `https://example.org${orgPath}`, + body: "host-org", + }); + + // host discriminates: filtering host example.org excludes the example.com message. + await eventually( + async () => { + const byHost = await client.logs({ filter: { host: "example.org" } }); + const ids = new Set(byHost.logs.map((l) => l.messageId)); + expect(ids.has(orgId)).toBe(true); + expect(ids.has(comId)).toBe(false); + }, + { interval: 1000 } + ); + + // path discriminates: filtering the unique example.com path excludes the example.org message. + await eventually( + async () => { + const byPath = await client.logs({ filter: { path: comPath } }); + const ids = new Set(byPath.logs.map((l) => l.messageId)); + expect(ids.has(comId)).toBe(true); + expect(ids.has(orgId)).toBe(false); + }, + { interval: 1000 } + ); + }, + { timeout: 15_000 } + ); + test("should use cursor", async () => { await client.logs({ filter: { @@ -173,6 +242,33 @@ describe("logs - mocked filter url shape", () => { }, }); }); + + test("should send multi-value url and host/path filters as repeated query params", async () => { + const client = new Client({ token, baseUrl: MOCK_QSTASH_SERVER_URL }); + await mockQStashServer({ + execute: async () => { + await client.logs({ + filter: { + url: ["https://a.com", "https://b.com"], + host: ["a.com", "b.com"], + path: "/webhook", + }, + }); + }, + responseFields: { body: { events: [] }, status: 200 }, + receivesRequest: { + method: "GET", + token, + url: `${MOCK_QSTASH_SERVER_URL}/v2/events?url=${encodeURIComponent("https://a.com")}&url=${encodeURIComponent("https://b.com")}&host=a.com&host=b.com&path=%2Fwebhook`, + }, + validateRequest: (request) => { + const parameters = new URL(request.url).searchParams; + expect(parameters.getAll("url")).toEqual(["https://a.com", "https://b.com"]); + expect(parameters.getAll("host")).toEqual(["a.com", "b.com"]); + expect(parameters.getAll("path")).toEqual(["/webhook"]); + }, + }); + }); }); describe("events (deprecated)", () => { @@ -197,32 +293,3 @@ describe("events (deprecated)", () => { expect(Array.isArray(result.events)).toBe(true); }); }); - -const EVENTUALLY_TIMEOUT = 5000; - -export const eventually = async function ( - function_: () => Promise | void, - options: { - timeout?: number; - interval?: number; - } = {} -): Promise { - const { timeout = EVENTUALLY_TIMEOUT, interval = 100 } = options; - - const startTime = Date.now(); - - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - while (true) { - try { - await function_(); - // Success case - all assertions passed - return; - } catch (error) { - const lastError = error as Error; - if (Date.now() - startTime >= timeout) { - throw new Error(`Assertions not satisfied within timeout: ${lastError.message}`); - } - await new Promise((resolve) => setTimeout(resolve, interval)); - } - } -}; diff --git a/src/client/messages.test.ts b/src/client/messages.test.ts index 552582e..708e4e4 100644 --- a/src/client/messages.test.ts +++ b/src/client/messages.test.ts @@ -255,6 +255,79 @@ describe("Messages", () => { { timeout: 20_000 } ); + test( + "should cancel messages by multiple flowControlKeys (OR semantics)", + async () => { + const keyA = `cancel-multi-fc-a-${Date.now()}`; + const keyB = `cancel-multi-fc-b-${Date.now()}`; + const keyC = `cancel-multi-fc-c-${Date.now()}`; + + for (const key of [keyA, keyB, keyC]) { + await client.publish({ + url: "https://httpbin.org/status/200", + body: "hello", + delay: "10d", + flowControl: { key, parallelism: 1 }, + }); + } + + // Cancelling [A, B] should match the A and B messages but NOT C. + const result = await client.messages.cancel({ + filter: { flowControlKey: [keyA, keyB] }, + }); + expect(result.cancelled).toBe(2); + + // C survived and can still be cancelled on its own. + const remaining = await client.messages.cancel({ filter: { flowControlKey: keyC } }); + expect(remaining.cancelled).toBe(1); + }, + { timeout: 20_000 } + ); + + test( + "should cancel messages by destination path (single and multi-value)", + async () => { + const stamp = Date.now(); + const pathA = `/cancel-path-a-${stamp}`; + const pathB = `/cancel-path-b-${stamp}`; + const pathC = `/cancel-path-c-${stamp}`; + + for (const path of [pathA, pathB, pathC]) { + await client.publish({ url: `https://example.com${path}`, body: "hello", delay: "10d" }); + } + + // Unique paths make this deterministic: [A, B] cancels exactly two. + const result = await client.messages.cancel({ filter: { path: [pathA, pathB] } }); + expect(result.cancelled).toBe(2); + + // C is untouched and matched by its own path. + const remaining = await client.messages.cancel({ filter: { path: pathC } }); + expect(remaining.cancelled).toBe(1); + }, + { timeout: 20_000 } + ); + + test( + "should cancel messages by destination host (host discriminates)", + async () => { + const stamp = Date.now(); + const comPath = `/cancel-host-com-${stamp}`; + const orgPath = `/cancel-host-org-${stamp}`; + + await client.publish({ url: `https://example.com${comPath}`, body: "hello", delay: "10d" }); + await client.publish({ url: `https://example.org${orgPath}`, body: "hello", delay: "10d" }); + + // Cancelling host example.org must not touch the example.com message. + const orgResult = await client.messages.cancel({ filter: { host: "example.org" } }); + expect(orgResult.cancelled).toBeGreaterThanOrEqual(1); + + // The example.com message survived — proven by cancelling it via its path. + const comResult = await client.messages.cancel({ filter: { path: comPath } }); + expect(comResult.cancelled).toBe(1); + }, + { timeout: 20_000 } + ); + test( "should respect count: 1 with all: true", async () => { diff --git a/src/client/messages.ts b/src/client/messages.ts index 314a208..4b8f228 100644 --- a/src/client/messages.ts +++ b/src/client/messages.ts @@ -181,6 +181,11 @@ export class Messages { * - A filter object: `cancel({ filter: { flowControlKey: "key", label: "label" } })` * - All messages: `cancel({ all: true })` * + * Filters support multiple values: pass an array to match a message whose value + * equals any of the given values (OR logic). Separate filters are combined with + * AND logic. For example: + * `cancel({ filter: { url: ["https://a.com", "https://b.com"], host: "a.com" } })` + * * Pass `count` to limit the number of messages processed per call (defaults to 100). * Call in a loop until `cancelled` is 0: * diff --git a/src/client/schedules.test.ts b/src/client/schedules.test.ts index f5c8e0b..b736626 100644 --- a/src/client/schedules.test.ts +++ b/src/client/schedules.test.ts @@ -310,6 +310,37 @@ describe("Schedules", () => { }, }); }); + + test("should create schedule with multiple labels", async () => { + const qstashToken = nanoid(); + + const client = new Client({ + baseUrl: MOCK_QSTASH_SERVER_URL, + token: qstashToken, + }); + await mockQStashServer({ + execute: async () => { + await client.schedules.create({ + scheduleId: "test-labels", + destination: MOCK_SERVER_URL, + cron: "* * * * 1", + label: ["label-1", "label-2"], + }); + }, + responseFields: { + body: { scheduleId: "test-labels" }, + status: 200, + }, + receivesRequest: { + method: "POST", + token: qstashToken, + url: "http://localhost:8080/v2/schedules/https://requestcatcher.com/", + headers: { + "upstash-label": "label-1,label-2", + }, + }, + }); + }); }); describe("Schedules empty id guard", () => { diff --git a/src/client/schedules.ts b/src/client/schedules.ts index 319de53..d564b72 100644 --- a/src/client/schedules.ts +++ b/src/client/schedules.ts @@ -1,5 +1,5 @@ /* eslint-disable @typescript-eslint/no-deprecated */ -import { assertNonEmptyId, prefixHeaders, wrapWithGlobalHeaders } from "./utils"; +import { assertNonEmptyId, prefixHeaders, serializeLabel, wrapWithGlobalHeaders } from "./utils"; import type { Requester } from "./http"; import type { BodyInit, FlowControl, HeadersInit, HTTPMethods } from "./types"; import type { Duration } from "./duration"; @@ -45,9 +45,19 @@ export type Schedule = { /** * The label assigned to the schedule for filtering purposes. + * + * @deprecated Use `labels` instead. When a schedule has multiple labels, this + * field only contains the first one. */ label?: string; + /** + * The labels assigned to the schedule for filtering purposes. + * + * A schedule can have multiple labels when created with `label: string[]`. + */ + labels?: string[]; + /** * The timestamp of the last scheduled execution. */ @@ -177,11 +187,14 @@ export type CreateScheduleRequest = { flowControl?: FlowControl; /** - * Assign a label to the schedule to filter logs later. + * Assign one or more labels to the schedule to filter logs later. + * + * Pass an array to attach multiple labels to a single schedule; they are sent + * as a comma-separated value in the `Upstash-Label` header. * * @default undefined */ - label?: string; + label?: string | string[]; /** * Configure which fields should be redacted in logs. @@ -293,7 +306,7 @@ export class Schedules { } if (request.label !== undefined) { - headers.set("Upstash-Label", request.label); + headers.set("Upstash-Label", serializeLabel(request.label)); } if (request.redact !== undefined) { diff --git a/src/client/test-utils.ts b/src/client/test-utils.ts new file mode 100644 index 0000000..e0ea4ff --- /dev/null +++ b/src/client/test-utils.ts @@ -0,0 +1,28 @@ +const EVENTUALLY_TIMEOUT = 5000; + +export const eventually = async function ( + function_: () => Promise | void, + options: { + timeout?: number; + interval?: number; + } = {} +): Promise { + const { timeout = EVENTUALLY_TIMEOUT, interval = 100 } = options; + + const startTime = Date.now(); + + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + while (true) { + try { + await function_(); + // Success case - all assertions passed + return; + } catch (error) { + const lastError = error as Error; + if (Date.now() - startTime >= timeout) { + throw new Error(`Assertions not satisfied within timeout: ${lastError.message}`); + } + await new Promise((resolve) => setTimeout(resolve, interval)); + } + } +}; diff --git a/src/client/utils.test.ts b/src/client/utils.test.ts index 7f808bd..d61d668 100644 --- a/src/client/utils.test.ts +++ b/src/client/utils.test.ts @@ -6,7 +6,7 @@ import { serializeLabel, } from "./utils"; import { describe, expect, test } from "bun:test"; -import type { DLQBulkActionFilters } from "./filter-types"; +import type { DLQBulkActionFilters, MessageCancelFilters } from "./filter-types"; describe("prefixHeaders", () => { test("should keep headers starting with upstash as they are - Upstash-", () => { @@ -80,6 +80,42 @@ describe("buildFilterPayload", () => { const result = buildBulkActionFilterPayload(request); expect(result).toHaveProperty("callerIp", "1.2.3.4"); }); + + test("should pass multi-value cancel filters through as arrays", () => { + const request: MessageCancelFilters = { + filter: { + url: ["https://a.com", "https://b.com"], + callerIp: ["1.2.3.4", "5.6.7.8"], + flowControlKey: ["key-1", "key-2"], + scheduleId: ["scd_1", "scd_2"], + queueName: ["queue-1", "queue-2"], + }, + }; + const result = buildBulkActionFilterPayload(request); + expect(result).toHaveProperty("url", ["https://a.com", "https://b.com"]); + expect(result).toHaveProperty("callerIp", ["1.2.3.4", "5.6.7.8"]); + expect(result).toHaveProperty("flowControlKey", ["key-1", "key-2"]); + expect(result).toHaveProperty("scheduleId", ["scd_1", "scd_2"]); + expect(result).toHaveProperty("queueName", ["queue-1", "queue-2"]); + }); + + test("should rename multi-value urlGroup to topicName", () => { + const request: MessageCancelFilters = { + filter: { urlGroup: ["group-1", "group-2"] }, + }; + const result = buildBulkActionFilterPayload(request); + expect(result).toHaveProperty("topicName", ["group-1", "group-2"]); + expect(result).not.toHaveProperty("urlGroup"); + }); + + test("should pass host and path cancel filters through", () => { + const request: MessageCancelFilters = { + filter: { host: ["a.com", "b.com"], path: "/webhook" }, + }; + const result = buildBulkActionFilterPayload(request); + expect(result).toHaveProperty("host", ["a.com", "b.com"]); + expect(result).toHaveProperty("path", "/webhook"); + }); }); describe("serializeLabel", () => { diff --git a/src/client/utils.ts b/src/client/utils.ts index 7fc715c..09be226 100644 --- a/src/client/utils.ts +++ b/src/client/utils.ts @@ -286,7 +286,7 @@ export function buildBulkActionFilterPayload(request: DLQBulkActionFilters | Mes // Filter branch const count = "count" in request ? request.count ?? DEFAULT_BULK_COUNT : DEFAULT_BULK_COUNT; return { - ...renameUrlGroup(request.filter as Record & { urlGroup?: string }), + ...renameUrlGroup(request.filter as Record & { urlGroup?: string | string[] }), count, cursor, }; @@ -295,9 +295,9 @@ export function buildBulkActionFilterPayload(request: DLQBulkActionFilters | Mes /** * Renames `urlGroup` to `topicName` in a filter object. */ -export function renameUrlGroup( +export function renameUrlGroup( filter: T -): Omit & { topicName?: string } { +): Omit & { topicName?: string | string[] } { // eslint-disable-next-line @typescript-eslint/no-unused-vars const { urlGroup, api, ...rest } = filter; return { ...rest, ...(urlGroup === undefined ? {} : { topicName: urlGroup }) };