Added waitForRequest built-in tasks (#725)

* Added waitForRequest built-in tasks

* Create grumpy-buttons-thank.md

.changeset/grumpy-buttons-thank.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Added waitForRequest built-in tasks

apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.callback.$secret.ts

@@ -1,125 +1,3 @@
-import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { action } from "./api.v1.tasks.$id.callback.$secret";
-import { json } from "@remix-run/server-runtime";
-import { RuntimeEnvironmentType } from "@trigger.dev/database";
-import { z } from "zod";
-import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
-import { logger } from "~/services/logger.server";
-import { ResumeTaskService } from "~/services/tasks/resumeTask.server";
-import { workerQueue } from "~/services/worker.server";
 
-const ParamsSchema = z.object({
+export { action };
-  runId: z.string(),
-  id: z.string(),
-  secret: z.string(),
-});
-
-export async function action({ request, params }: ActionFunctionArgs) {
-  // Ensure this is a POST request
-  if (request.method.toUpperCase() !== "POST") {
-    return { status: 405, body: "Method Not Allowed" };
-  }
-
-  const { runId, id } = ParamsSchema.parse(params);
-
-  // Parse body as JSON (no schema parsing)
-  const body = await request.json();
-
-  const service = new CallbackRunTaskService();
-
-  try {
-    // Complete task with request body as output
-    await service.call(runId, id, body, request.url);
-
-    return json({ success: true });
-  } catch (error) {
-    if (error instanceof Error) {
-      logger.error("Error while processing task callback:", { error });
-    }
-
-    return json({ error: "Something went wrong" }, { status: 500 });
-  }
-}
-
-export class CallbackRunTaskService {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
-
-  public async call(runId: string, id: string, taskBody: any, callbackUrl: string): Promise<void> {
-    const task = await findTask(prisma, id);
-
-    if (!task) {
-      return;
-    }
-
-    if (task.runId !== runId) {
-      return;
-    }
-
-    if (task.status !== "WAITING") {
-      return;
-    }
-
-    if (!task.callbackUrl) {
-      return;
-    }
-
-    if (new URL(task.callbackUrl).pathname !== new URL(callbackUrl).pathname) {
-      logger.error("Callback URLs don't match", { runId, taskId: id, callbackUrl });
-      return;
-    }
-
-    logger.debug("CallbackRunTaskService.call()", { task });
-
-    await this.#resumeTask(task, taskBody);
-  }
-
-  async #resumeTask(task: NonNullable<FoundTask>, output: any) {
-    await $transaction(this.#prismaClient, async (tx) => {
-      await tx.taskAttempt.updateMany({
-        where: {
-          taskId: task.id,
-          status: "PENDING",
-        },
-        data: {
-          status: "COMPLETED",
-        },
-      });
-
-      await tx.task.update({
-        where: { id: task.id },
-        data: {
-          status: "COMPLETED",
-          completedAt: new Date(),
-          output: output ? output : undefined,
-        },
-      });
-
-      await workerQueue.dequeue(`process-callback:${task.id}`, { tx });
-
-      await this.#resumeRunExecution(task, tx);
-    });
-  }
-
-  async #resumeRunExecution(task: NonNullable<FoundTask>, prisma: PrismaClientOrTransaction) {
-    await ResumeTaskService.enqueue(task.id, undefined, prisma);
-  }
-}
-
-type FoundTask = Awaited<ReturnType<typeof findTask>>;
-
-async function findTask(prisma: PrismaClientOrTransaction, id: string) {
-  return prisma.task.findUnique({
-    where: { id },
-    include: {
-      run: {
-        include: {
-          environment: true,
-          queue: true,
-        },
-      },
-    },
-  });
-}

apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts

@@ -248,7 +248,7 @@ export class RunTaskService {
 
       const taskId = ulid();
       const callbackUrl = callbackEnabled
-        ? `${env.APP_ORIGIN}/api/v1/runs/${runId}/tasks/${taskId}/callback/${generateSecret(12)}`
+        ? `${env.APP_ORIGIN}/api/v1/tasks/${taskId}/callback/${generateSecret(12)}`
         : undefined;
 
       const task = await tx.task.create({

apps/webapp/app/routes/api.v1.tasks.$id.callback.$secret.ts

@@ -0,0 +1,121 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { ResumeTaskService } from "~/services/tasks/resumeTask.server";
+import { workerQueue } from "~/services/worker.server";
+
+const ParamsSchema = z.object({
+  id: z.string(),
+  secret: z.string(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  const { id } = ParamsSchema.parse(params);
+
+  // Parse body as JSON (no schema parsing)
+  const body = await request.json();
+
+  const service = new CallbackRunTaskService();
+
+  try {
+    // Complete task with request body as output
+    await service.call(id, body, request.url);
+
+    return json({ success: true });
+  } catch (error) {
+    if (error instanceof Error) {
+      logger.error("Error while processing task callback:", { error });
+
+      return json({ error: `Something went wrong: ${error.message}` }, { status: 500 });
+    }
+    return json({ error: "Something went wrong" }, { status: 500 });
+  }
+}
+
+export class CallbackRunTaskService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(id: string, taskBody: any, callbackUrl: string): Promise<void> {
+    const task = await findTask(prisma, id);
+
+    if (!task) {
+      return;
+    }
+
+    if (task.status !== "WAITING") {
+      return;
+    }
+
+    if (!task.callbackUrl) {
+      throw new Error("Task doesn't have a callback URL");
+    }
+
+    if (new URL(task.callbackUrl).pathname !== new URL(callbackUrl).pathname) {
+      logger.debug("Callback URLs don't match", { taskId: id, callbackUrl });
+
+      throw new Error("Callback URLs don't match");
+    }
+
+    logger.debug("CallbackRunTaskService.call()", { task });
+
+    await this.#resumeTask(task, taskBody);
+  }
+
+  async #resumeTask(task: NonNullable<FoundTask>, output: any) {
+    await $transaction(this.#prismaClient, async (tx) => {
+      await tx.taskAttempt.updateMany({
+        where: {
+          taskId: task.id,
+          status: "PENDING",
+        },
+        data: {
+          status: "COMPLETED",
+        },
+      });
+
+      await tx.task.update({
+        where: { id: task.id },
+        data: {
+          status: "COMPLETED",
+          completedAt: new Date(),
+          output: output ? output : undefined,
+        },
+      });
+
+      await workerQueue.dequeue(`process-callback:${task.id}`, { tx });
+
+      await this.#resumeRunExecution(task, tx);
+    });
+  }
+
+  async #resumeRunExecution(task: NonNullable<FoundTask>, prisma: PrismaClientOrTransaction) {
+    await ResumeTaskService.enqueue(task.id, undefined, prisma);
+  }
+}
+
+type FoundTask = Awaited<ReturnType<typeof findTask>>;
+
+async function findTask(prisma: PrismaClientOrTransaction, id: string) {
+  return prisma.task.findUnique({
+    where: { id },
+    include: {
+      run: {
+        include: {
+          environment: true,
+          queue: true,
+        },
+      },
+    },
+  });
+}

packages/trigger-sdk/src/io.ts

@@ -331,6 +331,65 @@ export class IO {
     });
   }
 
+  /** `io.waitForRequest()` allows you to pause the execution of a run until the url provided in the callback is POSTed to.
+   *  This is useful for integrating with external services that require a callback URL to be provided, or if you want to be able to wait until an action is performed somewhere else in your system.
+   *  @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+   *  @param callback A callback function that will provide the unique URL to POST to.
+   *  @param options Options for the callback.
+   *  @param options.timeoutInSeconds How long to wait for the request to be POSTed to the callback URL before timing out. Defaults to 1hr.
+   *  @returns The POSTed request JSON body.
+   *  @example
+   * ```ts
+    const result = await io.waitForRequest<{ message: string }>(
+      "wait-for-request",
+      async (url, task) => {
+        // Save the URL somewhere so you can POST to it later
+        // Or send it to an external service that will POST to it
+      },
+      { timeoutInSeconds: 60 } // wait 60 seconds
+    );
+    * ```
+   */
+  async waitForRequest<T extends Json<T> | unknown = unknown>(
+    cacheKey: string | any[],
+    callback: (url: string) => Promise<unknown>,
+    options?: { timeoutInSeconds?: number }
+  ): Promise<T> {
+    const timeoutInSeconds = options?.timeoutInSeconds ?? 60 * 60;
+
+    return (await this.runTask(
+      cacheKey,
+      async (task, io) => {
+        if (!task.callbackUrl) {
+          throw new Error("No callbackUrl found on task");
+        }
+
+        task.outputProperties = [
+          {
+            label: "Callback URL",
+            text: task.callbackUrl,
+          },
+        ];
+
+        return callback(task.callbackUrl) as Promise<{}>;
+      },
+      {
+        name: "Wait for Request",
+        icon: "clock",
+        callback: {
+          enabled: true,
+          timeoutInSeconds: options?.timeoutInSeconds,
+        },
+        properties: [
+          {
+            label: "Timeout",
+            text: `${timeoutInSeconds}s`,
+          },
+        ],
+      }
+    )) as T;
+  }
+
   /** `io.createStatus()` allows you to set a status with associated data during the Run. Statuses can be used by your UI using the react package 
    * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
    * @param initialStatus The initial status you want this status to have. You can update it during the rub using the returned object.

references/job-catalog/package.json

@@ -12,7 +12,6 @@
     "events": "nodemon --watch src/events.ts -r tsconfig-paths/register -r dotenv/config src/events.ts",
     "schedules": "nodemon --watch src/schedules.ts -r tsconfig-paths/register -r dotenv/config src/schedules.ts",
     "stressTest": "nodemon --watch src/stressTest.ts -r tsconfig-paths/register -r dotenv/config src/stressTest.ts",
-    "delays": "nodemon --watch src/delays.ts -r tsconfig-paths/register -r dotenv/config src/delays.ts",
     "airtable": "nodemon --watch src/airtable.ts -r tsconfig-paths/register -r dotenv/config src/airtable.ts",
     "resend": "nodemon --watch src/resend.ts -r tsconfig-paths/register -r dotenv/config src/resend.ts",
     "github": "nodemon --watch src/github.ts -r tsconfig-paths/register -r dotenv/config src/github.ts",
@@ -30,8 +29,8 @@
     "auto-yield": "nodemon --watch src/auto-yield.ts -r tsconfig-paths/register -r dotenv/config src/auto-yield.ts",
     "httptrigger": "nodemon --watch src/httpTrigger.ts -r tsconfig-paths/register -r dotenv/config src/httpTrigger.ts",
     "cli-example": "nodemon --watch src/cli-example.ts -r tsconfig-paths/register -r dotenv/config src/cli-example.ts",
-    "random": "nodemon --watch src/random.ts -r tsconfig-paths/register -r dotenv/config src/random.ts",
     "invoke": "nodemon --watch src/invoke.ts -r tsconfig-paths/register -r dotenv/config src/invoke.ts",
+    "built-ins": "nodemon --watch src/built-ins.ts -r tsconfig-paths/register -r dotenv/config src/built-ins.ts",
     "dev:trigger": "trigger-cli dev --port 8080"
   },
   "dependencies": {

references/job-catalog/src/built-ins.ts

@@ -1,5 +1,5 @@
 import { createExpressServer } from "@trigger.dev/express";
-import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { TriggerClient, eventTrigger, invokeTrigger } from "@trigger.dev/sdk";
 
 export const client = new TriggerClient({
   id: "job-catalog",
@@ -48,4 +48,56 @@ client.defineJob({
   },
 });
 
+client.defineJob({
+  id: "delays-example-1",
+  name: "Delays Example 1",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "delays.example",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.wait("wait-1", 60);
+  },
+});
+
+client.defineJob({
+  id: "delays-example-2",
+  name: "Delays Example 2 - Long Delay",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "delays.example.long",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.wait("wait-1", 60 * 30);
+  },
+});
+
+client.defineJob({
+  id: "wait-for-request-example",
+  name: "Wait for Request Example",
+  version: "1.0.0",
+  trigger: invokeTrigger(),
+  run: async (payload, io, ctx) => {
+    const result = await io.waitForRequest<{ message: string }>(
+      "wait-for-request",
+      async (url) => {
+        console.log("Waiting for request", url);
+      },
+      {
+        timeoutInSeconds: 60,
+      }
+    );
+
+    const result2 = await io.waitForRequest(
+      "wait-for-request-2",
+      async (url) => {
+        console.log("Waiting for request 2", url);
+      },
+      {
+        timeoutInSeconds: 10,
+      }
+    );
+  },
+});
+
 createExpressServer(client);

references/job-catalog/src/delays.ts

@@ -1,36 +0,0 @@
-import { createExpressServer } from "@trigger.dev/express";
-import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
-
-export const client = new TriggerClient({
-  id: "job-catalog",
-  apiKey: process.env["TRIGGER_API_KEY"],
-  apiUrl: process.env["TRIGGER_API_URL"],
-  verbose: false,
-  ioLogLocalEnabled: true,
-});
-
-client.defineJob({
-  id: "delays-example-1",
-  name: "Delays Example 1",
-  version: "1.0.0",
-  trigger: eventTrigger({
-    name: "delays.example",
-  }),
-  run: async (payload, io, ctx) => {
-    await io.wait("wait-1", 60);
-  },
-});
-
-client.defineJob({
-  id: "delays-example-2",
-  name: "Delays Example 2 - Long Delay",
-  version: "1.0.0",
-  trigger: eventTrigger({
-    name: "delays.example.long",
-  }),
-  run: async (payload, io, ctx) => {
-    await io.wait("wait-1", 60 * 30);
-  },
-});
-
-createExpressServer(client);