From 0a04295883f142ba4ff40ca954bc846a058ffa85 Mon Sep 17 00:00:00 2001 From: CahidArda Date: Fri, 15 Nov 2024 15:21:32 +0300 Subject: [PATCH 1/5] fix: update cancel method --- src/client/index.test.ts | 157 +++++++++++++++++++++++++++++++++++---- src/client/index.ts | 73 +++++++++++++++--- 2 files changed, 205 insertions(+), 25 deletions(-) diff --git a/src/client/index.test.ts b/src/client/index.test.ts index 4f4c96f..67956da 100644 --- a/src/client/index.test.ts +++ b/src/client/index.test.ts @@ -1,4 +1,4 @@ -import { describe, test } from "bun:test"; +import { describe, test, expect } from "bun:test"; import { MOCK_QSTASH_SERVER_URL, mockQStashServer, WORKFLOW_ENDPOINT } from "../test-utils"; import { Client } from "."; import { getWorkflowRunId, nanoid } from "../utils"; @@ -7,21 +7,146 @@ describe("workflow client", () => { const token = nanoid(); const client = new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token }); - test("should send cancel", async () => { - const workflowRunId = `wfr-${nanoid()}`; - await mockQStashServer({ - execute: async () => { - await client.cancel({ workflowRunId }); - }, - responseFields: { - status: 200, - body: "msgId", - }, - receivesRequest: { - method: "DELETE", - url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs/${workflowRunId}?cancel=true`, - token, - }, + describe("cancel - mocked", () => { + test("should cancel single workflow run id", async () => { + const workflowRunId = `wfr-${nanoid()}`; + await mockQStashServer({ + execute: async () => { + await client.cancel({ workflowRunId }); + }, + responseFields: { + status: 200, + body: "msgId", + }, + receivesRequest: { + method: "DELETE", + url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs`, + token, + body: { workflowRunIds: [workflowRunId] }, + }, + }); + }); + + test("should cancel multiple workflow run ids", async () => { + const workflowRunId = [`wfr-${nanoid()}`, `wfr-${nanoid()}`]; + await mockQStashServer({ + execute: async () => { + await client.cancel({ workflowRunId }); + }, + responseFields: { + status: 200, + body: "msgId", + }, + receivesRequest: { + method: "DELETE", + url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs`, + token, + body: { workflowRunIds: workflowRunId }, + }, + }); + }); + + test("should cancel workflowUrl", async () => { + const workflowUrl = "http://workflow-endpoint.com"; + await mockQStashServer({ + execute: async () => { + await client.cancel({ workflowUrl }); + }, + responseFields: { + status: 200, + body: "msgId", + }, + receivesRequest: { + method: "DELETE", + url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs`, + token, + body: { workflowUrl }, + }, + }); + }); + + test("should cancel all", async () => { + await mockQStashServer({ + execute: async () => { + await client.cancel(); + }, + responseFields: { + status: 200, + body: "msgId", + }, + receivesRequest: { + method: "DELETE", + url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs`, + token, + body: {}, + }, + }); + }); + }); + + describe("cancel - live", () => { + const liveClient = new Client({ + baseUrl: process.env.QSTASH_URL, + token: process.env.QSTASH_TOKEN!, + }); + + test("should cancel single workflow run id", async () => { + const { workflowRunId } = await liveClient.trigger({ + url: "http://requestcatcher.com", + }); + + const cancel = await liveClient.cancel({ + workflowRunId, + }); + expect(cancel).toEqual({ cancelled: 1 }); + + const throws = () => liveClient.cancel({ workflowRunId }); + expect(throws).toThrow(`{"error":"workflowRun ${workflowRunId} not found"}`); + }); + + test("should cancel multiple workflow run ids", async () => { + const { workflowRunId: workflowRunIdOne } = await liveClient.trigger({ + url: "http://requestcatcher.com", + }); + const { workflowRunId: workflowRunIdTwo } = await liveClient.trigger({ + url: "http://requestcatcher.com", + }); + + const throws = async () => + await liveClient.cancel({ + workflowRunId: [workflowRunIdOne, workflowRunIdTwo, "non-existent"], + }); + + // if there is any workflow which doesn't exist, we throw + expect(throws).toThrow(`{"error":"workflowRun non-existent not found"}`); + + // trying to cancel the workflows one by one gives error, as they were canceled above + const throwsFirst = async () => await liveClient.cancel({ workflowRunId: workflowRunIdOne }); + expect(throwsFirst).toThrow(`{"error":"workflowRun ${workflowRunIdOne} not found"}`); + + // trying to cancel the workflows one by one gives error, as they were canceled above + const throwsSecond = async () => await liveClient.cancel({ workflowRunId: workflowRunIdTwo }); + expect(throwsSecond).toThrow(`{"error":"workflowRun ${workflowRunIdTwo} not found"}`); + }); + + test("should cancel workflowUrl", async () => { + await liveClient.trigger({ + url: "http://requestcatcher.com/first", + }); + await liveClient.trigger({ + url: "http://requestcatcher.com/second", + }); + + const cancel = await liveClient.cancel({ + workflowUrl: "http://requestcatcher.com", + }); + + expect(cancel).toEqual({ cancelled: 2 }); + }); + + test.skip("should cancel all", async () => { + // intentionally didn't write a test for cancel.all, + // because it may break apps running on the same QStash user. }); }); diff --git a/src/client/index.ts b/src/client/index.ts index 3d56a84..d3cb821 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -7,6 +7,10 @@ import { WorkflowContext } from "../context"; import { DEFAULT_RETRIES } from "../constants"; type ClientConfig = ConstructorParameters[0]; +type CancelOptions = { + workflowRunId?: string | string[]; + workflowUrl?: string; +}; /** * Workflow client for canceling & notifying workflows and getting waiters of an @@ -30,23 +34,74 @@ export class Client { /** * Cancel an ongoing workflow * - * ```ts - * import { Client } from "@upstash/workflow"; + * There are multiple ways you can cancel workflows: + * - pass one or more workflow run ids to cancel them + * - pass a workflow url to cancel all runs starting with this url + * - don't pass any options. in this case, all workflow will be canceled * - * const client = new Client({ token: "" }) + * ### Cancel a set of workflow runs + * + * ```ts + * // cancel a single workflow * await client.cancel({ workflowRunId: "" }) + * + * // cancel a set of workflow runs + * await client.cancel({ workflowRunId: [ + * "", + * "", + * ]}) + * ``` + * + * ### Cancel workflows starting with a url + * + * If you have an endpoint called `https://your-endpoint.com` and you + * want to cancel all workflow runs on it, you can use `workflowUrl`. + * + * Note that this will cancel workflows in all endpoints under + * `https://your-endpoint.com`. + * + * ```ts + * await client.cancel({ workflowUrl: "https://your-endpoint.com" }) + * ``` + * + * ### Cancel *all* workflows + * + * If you want to cancel all workflows under your user, you can + * do it like this: + * + * ```ts + * await client.cancel() * ``` * * @param workflowRunId run id of the workflow to delete + * @param workflowUrl cancel workflows starting with this url. Will be ignored + * if `workflowRunId` parameter is set. * @returns true if workflow is succesfully deleted. Otherwise throws QStashError */ - public async cancel({ workflowRunId }: { workflowRunId: string }) { - const result = (await this.client.http.request({ - path: ["v2", "workflows", "runs", `${workflowRunId}?cancel=true`], + public async cancel(options?: CancelOptions) { + const { workflowRunId, workflowUrl } = options ?? {}; + + let body: string; + if (workflowRunId) { + const runIdArray = typeof workflowRunId === "string" ? [workflowRunId] : workflowRunId; + + body = JSON.stringify({ workflowRunIds: runIdArray }); + } else if (workflowUrl) { + body = JSON.stringify({ workflowUrl }); + } else { + body = "{}"; + } + + const result = await this.client.http.request<{ cancelled: number }>({ + path: ["v2", "workflows", "runs"], method: "DELETE", - parseResponseAsJson: false, - })) as { error: string } | undefined; - return result ?? true; + body, + headers: { + "Content-Type": "application/json", + }, + }); + + return result; } /** From 9c0e139cb4d795a26990699c172dc98eedb57da2 Mon Sep 17 00:00:00 2001 From: CahidArda Date: Fri, 15 Nov 2024 16:54:16 +0300 Subject: [PATCH 2/5] fix: add all option --- src/client/index.test.ts | 7 ++++++- src/client/index.ts | 24 +++++++++++++++--------- 2 files changed, 21 insertions(+), 10 deletions(-) diff --git a/src/client/index.test.ts b/src/client/index.test.ts index 67956da..b1f212a 100644 --- a/src/client/index.test.ts +++ b/src/client/index.test.ts @@ -68,7 +68,7 @@ describe("workflow client", () => { test("should cancel all", async () => { await mockQStashServer({ execute: async () => { - await client.cancel(); + await client.cancel({ all: true }); }, responseFields: { status: 200, @@ -82,6 +82,11 @@ describe("workflow client", () => { }, }); }); + + test("should throw if no option", async () => { + const throws = () => client.cancel({}); + expect(throws).toThrow("The `cancel` method cannot be called without any options."); + }); }); describe("cancel - live", () => { diff --git a/src/client/index.ts b/src/client/index.ts index d3cb821..c6e0ae2 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -7,10 +7,6 @@ import { WorkflowContext } from "../context"; import { DEFAULT_RETRIES } from "../constants"; type ClientConfig = ConstructorParameters[0]; -type CancelOptions = { - workflowRunId?: string | string[]; - workflowUrl?: string; -}; /** * Workflow client for canceling & notifying workflows and getting waiters of an @@ -70,17 +66,25 @@ export class Client { * do it like this: * * ```ts - * await client.cancel() + * await client.cancel({ all: true }) * ``` * * @param workflowRunId run id of the workflow to delete * @param workflowUrl cancel workflows starting with this url. Will be ignored * if `workflowRunId` parameter is set. + * @param all set to true in order to cancel all workflows. Will be ignored + * if `workflowRunId` or `workflowUrl` parameters are set. * @returns true if workflow is succesfully deleted. Otherwise throws QStashError */ - public async cancel(options?: CancelOptions) { - const { workflowRunId, workflowUrl } = options ?? {}; - + public async cancel({ + workflowRunId, + workflowUrl, + all, + }: { + workflowRunId?: string | string[]; + workflowUrl?: string; + all?: true; + }) { let body: string; if (workflowRunId) { const runIdArray = typeof workflowRunId === "string" ? [workflowRunId] : workflowRunId; @@ -88,8 +92,10 @@ export class Client { body = JSON.stringify({ workflowRunIds: runIdArray }); } else if (workflowUrl) { body = JSON.stringify({ workflowUrl }); - } else { + } else if (all) { body = "{}"; + } else { + throw new TypeError("The `cancel` method cannot be called without any options."); } const result = await this.client.http.request<{ cancelled: number }>({ From f8b325e690c7dddb987b91f1bfa0df1430b8450e Mon Sep 17 00:00:00 2001 From: CahidArda Date: Fri, 15 Nov 2024 17:23:32 +0300 Subject: [PATCH 3/5] fix: update parameters --- src/client/index.test.ts | 30 +++++++++++++++--------------- src/client/index.ts | 32 ++++++++++++++++---------------- 2 files changed, 31 insertions(+), 31 deletions(-) diff --git a/src/client/index.test.ts b/src/client/index.test.ts index b1f212a..3af4f71 100644 --- a/src/client/index.test.ts +++ b/src/client/index.test.ts @@ -9,10 +9,10 @@ describe("workflow client", () => { describe("cancel - mocked", () => { test("should cancel single workflow run id", async () => { - const workflowRunId = `wfr-${nanoid()}`; + const ids = `wfr-${nanoid()}`; await mockQStashServer({ execute: async () => { - await client.cancel({ workflowRunId }); + await client.cancel({ ids }); }, responseFields: { status: 200, @@ -22,16 +22,16 @@ describe("workflow client", () => { method: "DELETE", url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs`, token, - body: { workflowRunIds: [workflowRunId] }, + body: { workflowRunIds: [ids] }, }, }); }); test("should cancel multiple workflow run ids", async () => { - const workflowRunId = [`wfr-${nanoid()}`, `wfr-${nanoid()}`]; + const ids = [`wfr-${nanoid()}`, `wfr-${nanoid()}`]; await mockQStashServer({ execute: async () => { - await client.cancel({ workflowRunId }); + await client.cancel({ ids }); }, responseFields: { status: 200, @@ -41,16 +41,16 @@ describe("workflow client", () => { method: "DELETE", url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs`, token, - body: { workflowRunIds: workflowRunId }, + body: { workflowRunIds: ids }, }, }); }); test("should cancel workflowUrl", async () => { - const workflowUrl = "http://workflow-endpoint.com"; + const urlStartingWith = "http://workflow-endpoint.com"; await mockQStashServer({ execute: async () => { - await client.cancel({ workflowUrl }); + await client.cancel({ urlStartingWith }); }, responseFields: { status: 200, @@ -60,7 +60,7 @@ describe("workflow client", () => { method: "DELETE", url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs`, token, - body: { workflowUrl }, + body: { workflowUrl: urlStartingWith }, }, }); }); @@ -101,11 +101,11 @@ describe("workflow client", () => { }); const cancel = await liveClient.cancel({ - workflowRunId, + ids: workflowRunId, }); expect(cancel).toEqual({ cancelled: 1 }); - const throws = () => liveClient.cancel({ workflowRunId }); + const throws = () => liveClient.cancel({ ids: workflowRunId }); expect(throws).toThrow(`{"error":"workflowRun ${workflowRunId} not found"}`); }); @@ -119,18 +119,18 @@ describe("workflow client", () => { const throws = async () => await liveClient.cancel({ - workflowRunId: [workflowRunIdOne, workflowRunIdTwo, "non-existent"], + ids: [workflowRunIdOne, workflowRunIdTwo, "non-existent"], }); // if there is any workflow which doesn't exist, we throw expect(throws).toThrow(`{"error":"workflowRun non-existent not found"}`); // trying to cancel the workflows one by one gives error, as they were canceled above - const throwsFirst = async () => await liveClient.cancel({ workflowRunId: workflowRunIdOne }); + const throwsFirst = async () => await liveClient.cancel({ ids: workflowRunIdOne }); expect(throwsFirst).toThrow(`{"error":"workflowRun ${workflowRunIdOne} not found"}`); // trying to cancel the workflows one by one gives error, as they were canceled above - const throwsSecond = async () => await liveClient.cancel({ workflowRunId: workflowRunIdTwo }); + const throwsSecond = async () => await liveClient.cancel({ ids: workflowRunIdTwo }); expect(throwsSecond).toThrow(`{"error":"workflowRun ${workflowRunIdTwo} not found"}`); }); @@ -143,7 +143,7 @@ describe("workflow client", () => { }); const cancel = await liveClient.cancel({ - workflowUrl: "http://requestcatcher.com", + urlStartingWith: "http://requestcatcher.com", }); expect(cancel).toEqual({ cancelled: 2 }); diff --git a/src/client/index.ts b/src/client/index.ts index c6e0ae2..757d011 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -39,10 +39,10 @@ export class Client { * * ```ts * // cancel a single workflow - * await client.cancel({ workflowRunId: "" }) + * await client.cancel({ ids: "" }) * * // cancel a set of workflow runs - * await client.cancel({ workflowRunId: [ + * await client.cancel({ ids: [ * "", * "", * ]}) @@ -51,13 +51,13 @@ export class Client { * ### Cancel workflows starting with a url * * If you have an endpoint called `https://your-endpoint.com` and you - * want to cancel all workflow runs on it, you can use `workflowUrl`. + * want to cancel all workflow runs on it, you can use `urlStartingWith`. * * Note that this will cancel workflows in all endpoints under * `https://your-endpoint.com`. * * ```ts - * await client.cancel({ workflowUrl: "https://your-endpoint.com" }) + * await client.cancel({ urlStartingWith: "https://your-endpoint.com" }) * ``` * * ### Cancel *all* workflows @@ -69,29 +69,29 @@ export class Client { * await client.cancel({ all: true }) * ``` * - * @param workflowRunId run id of the workflow to delete - * @param workflowUrl cancel workflows starting with this url. Will be ignored - * if `workflowRunId` parameter is set. + * @param ids run id of the workflow to delete + * @param urlStartingWith cancel workflows starting with this url. Will be ignored + * if `ids` parameter is set. * @param all set to true in order to cancel all workflows. Will be ignored - * if `workflowRunId` or `workflowUrl` parameters are set. + * if `ids` or `urlStartingWith` parameters are set. * @returns true if workflow is succesfully deleted. Otherwise throws QStashError */ public async cancel({ - workflowRunId, - workflowUrl, + ids, + urlStartingWith, all, }: { - workflowRunId?: string | string[]; - workflowUrl?: string; + ids?: string | string[]; + urlStartingWith?: string; all?: true; }) { let body: string; - if (workflowRunId) { - const runIdArray = typeof workflowRunId === "string" ? [workflowRunId] : workflowRunId; + if (ids) { + const runIdArray = typeof ids === "string" ? [ids] : ids; body = JSON.stringify({ workflowRunIds: runIdArray }); - } else if (workflowUrl) { - body = JSON.stringify({ workflowUrl }); + } else if (urlStartingWith) { + body = JSON.stringify({ workflowUrl: urlStartingWith }); } else if (all) { body = "{}"; } else { From 265d4a3c542994b63636bc45a1d9c5723ca264fa Mon Sep 17 00:00:00 2001 From: CahidArda Date: Fri, 15 Nov 2024 17:54:00 +0300 Subject: [PATCH 4/5] fix: docstring --- src/client/index.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/client/index.ts b/src/client/index.ts index 757d011..4c13697 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -33,7 +33,7 @@ export class Client { * There are multiple ways you can cancel workflows: * - pass one or more workflow run ids to cancel them * - pass a workflow url to cancel all runs starting with this url - * - don't pass any options. in this case, all workflow will be canceled + * - cancel all pending or active workflow runs * * ### Cancel a set of workflow runs * @@ -62,7 +62,7 @@ export class Client { * * ### Cancel *all* workflows * - * If you want to cancel all workflows under your user, you can + * To cancel all pending and currently running workflows, you can * do it like this: * * ```ts From e1376a22a84af33140551d290b6c499328f36a88 Mon Sep 17 00:00:00 2001 From: CahidArda Date: Mon, 18 Nov 2024 17:07:07 +0300 Subject: [PATCH 5/5] fix: trigger docstring --- src/client/index.ts | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/client/index.ts b/src/client/index.ts index 4c13697..83f3907 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -158,13 +158,13 @@ export class Client { * Trigger new workflow run and returns the workflow run id * * ```ts - * const { workflowRunId } await client.trigger({ + * const { workflowRunId } = await client.trigger({ * url: "https://workflow-endpoint.com", - * body: "hello there!", // optional body - * headers: { ... }, // optional headers - * workflowRunId: "my-workflow", // optional workflow run id - * retries: 3 // optional retries in the initial request - * }) + * body: "hello there!", // Optional body + * headers: { ... }, // Optional headers + * workflowRunId: "my-workflow", // Optional workflow run ID + * retries: 3 // Optional retries for the initial request + * }); * * console.log(workflowRunId) * // wfr_my-workflow