feat: add delay to client.trigger (#100)

Author: CahidArda

src/client/index.test.ts

@@ -209,6 +209,7 @@ describe("workflow client", () => {
           headers: { "user-header": "user-header-value" },
           workflowRunId: myWorkflowRunId,
           retries: 15,
+          delay: 1,
         });
       },
       responseFields: {
@@ -228,6 +229,7 @@ describe("workflow client", () => {
           "upstash-workflow-init": "true",
           "upstash-workflow-runid": `wfr_${myWorkflowRunId}`,
           "upstash-workflow-url": "https://requestcatcher.com/api",
+          "upstash-delay": "1s",
         },
       },
     });

src/client/index.ts

@@ -1,5 +1,5 @@
 import { NotifyResponse, Waiter } from "../types";
-import { FlowControl, Client as QStashClient } from "@upstash/qstash";
+import { FlowControl, PublishRequest, Client as QStashClient } from "@upstash/qstash";
 import { makeGetWaitersRequest, makeNotifyRequest } from "./utils";
 import { getWorkflowRunId } from "../utils";
 import { triggerFirstInvocation } from "../workflow-requests";
@@ -190,6 +190,11 @@ export class Client {
    *   with `wfr_`.
    * @param retries retry to use in the initial request. in the rest of
    *   the workflow, `retries` option of the `serve` will be used.
+   * @param flowControl Settings for controlling the number of active requests
+   *   and number of requests per second with the same key.
+   * @param delay Delay for the workflow run. This is used to delay the
+   *   execution of the workflow run. The delay is in seconds or can be passed
+   *   as a string with a time unit (e.g. "1h", "30m", "15s").
    * @returns workflow run id
    */
   public async trigger({
@@ -199,13 +204,15 @@ export class Client {
     workflowRunId,
     retries,
     flowControl,
+    delay,
   }: {
     url: string;
     body?: unknown;
     headers?: Record<string, string>;
     workflowRunId?: string;
     retries?: number;
     flowControl?: FlowControl;
+    delay?: PublishRequest["delay"];
   }): Promise<{ workflowRunId: string }> {
     const finalWorkflowRunId = getWorkflowRunId(workflowRunId);
     const context = new WorkflowContext({
@@ -223,6 +230,7 @@ export class Client {
     const result = await triggerFirstInvocation({
       workflowContext: context,
       telemetry: undefined, // can't know workflow telemetry here
+      delay,
     });
     if (result.isOk()) {
       return { workflowRunId: finalWorkflowRunId };

src/workflow-requests.ts

@@ -20,7 +20,7 @@ import type {
 } from "./types";
 import { StepTypes } from "./types";
 import type { WorkflowLogger } from "./logger";
-import { FlowControl, QstashError } from "@upstash/qstash";
+import { FlowControl, PublishRequest, QstashError } from "@upstash/qstash";
 import { getSteps } from "./client/utils";
 import { getHeaders } from "./qstash/headers";
 
@@ -30,12 +30,14 @@ export const triggerFirstInvocation = async <TInitialPayload>({
   telemetry,
   debug,
   invokeCount,
+  delay,
 }: {
   workflowContext: WorkflowContext<TInitialPayload>;
   useJSONContent?: boolean;
   telemetry?: Telemetry;
   debug?: WorkflowLogger;
   invokeCount?: number;
+  delay?: PublishRequest["delay"];
 }): Promise<Ok<"success" | "workflow-run-already-exists", never> | Err<never, Error>> => {
   const { headers } = getHeaders({
     initHeaderValue: "true",
@@ -72,6 +74,7 @@ export const triggerFirstInvocation = async <TInitialPayload>({
       method: "POST",
       body,
       url: workflowContext.url,
+      delay,
     });
 
     if (result.deduplicated) {