github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Added waitForRequest built-in tasks (#725)

Browse Source 
* Added waitForRequest built-in tasks

* Create grumpy-buttons-thank.md
This commit is contained in:
Eric Allam
2023-11-06 18:30:49 +00:00
committed by GitHub
parent 42185bc36b
commit a74716a1b9
8 changed files with 242 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/grumpy-buttons-thank.md
+5
View File
@@ -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
+2 -124
View File
@@ -1,125 +1,3 @@
import type { ActionFunctionArgs } from "@remix-run/server-runtime";
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";
import { action } from "./api.v1.tasks.$id.callback.$secret";
const ParamsSchema = z.object({
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,
},
},
},
});
}
export { action };
apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts
+1 -1
View File
@@ -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
+121
View File
@@ -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
+59
View File
@@ -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
+1 -2
View File
@@ -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/random.ts → references/job-catalog/src/built-ins.ts
+53 -1
View File
@@ -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
-36
View File
@@ -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);