Some improvements to the resume tokens

Author: matt-aitken

apps/webapp/app/routes/api.v1.waitpoints.ts

@@ -1,5 +1,6 @@
 import { json } from "@remix-run/server-runtime";
 import { CreateWaitpointRequestBody, CreateWaitpointResponseBody } from "@trigger.dev/core/v3";
+import { ResumeTokenId, WaitpointId } from "@trigger.dev/core/v3/apps";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { parseDelay } from "~/utils/delays";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
@@ -18,17 +19,21 @@ const { action } = createActionApiRoute(
 
     const timeout = await parseDelay(body.timeout);
 
-    const waitpoint = await engine.createManualWaitpoint({
+    const result = await engine.createManualWaitpoint({
       environmentId: authentication.environment.id,
       projectId: authentication.environment.projectId,
       idempotencyKey: body.idempotencyKey,
       idempotencyKeyExpiresAt,
       timeout,
     });
 
+    //result is a waitpoint but we want to make it look like a resume token
+    const resumeTokenFriendlyId = ResumeTokenId.toFriendlyId(result.waitpoint.id);
+
     return json<CreateWaitpointResponseBody>(
       {
-        id: waitpoint.friendlyId,
+        id: resumeTokenFriendlyId,
+        isCached: result.isCached,
       },
       { status: 200 }
     );

internal-packages/run-engine/src/engine/index.ts

@@ -1744,7 +1744,7 @@ export class RunEngine {
     idempotencyKey?: string;
     idempotencyKeyExpiresAt?: Date;
     timeout?: Date;
-  }): Promise<Waitpoint> {
+  }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> {
     const existingWaitpoint = idempotencyKey
       ? await this.prisma.waitpoint.findUnique({
           where: {
@@ -1775,7 +1775,7 @@ export class RunEngine {
 
         //let it fall through to create a new waitpoint
       } else {
-        return existingWaitpoint;
+        return { waitpoint: existingWaitpoint, isCached: true };
       }
     }
 
@@ -1815,7 +1815,7 @@ export class RunEngine {
       });
     }
 
-    return waitpoint;
+    return { waitpoint, isCached: false };
   }
 
   /** This block a run with a BATCH waitpoint.

packages/core/src/v3/apiClient/core.ts

@@ -32,7 +32,7 @@ export type ZodFetchOptions<T = unknown> = {
   name?: string;
   attributes?: Attributes;
   icon?: string;
-  onResponseBody?: (body: unknown, span: Span) => void;
+  onResponseBody?: (body: T, span: Span) => void;
   prepareData?: (data: T) => Promise<T> | T;
 };
 

packages/core/src/v3/apps/friendlyId.ts

@@ -90,3 +90,4 @@ export const RunId = new IdUtil("run");
 export const SnapshotId = new IdUtil("snapshot");
 export const WaitpointId = new IdUtil("waitpoint");
 export const BatchId = new IdUtil("batch");
+export const ResumeTokenId = new IdUtil("resumetoken");

packages/core/src/v3/schemas/api.ts

@@ -914,6 +914,7 @@ export type CreateWaitpointRequestBody = z.infer<typeof CreateWaitpointRequestBo
 
 export const CreateWaitpointResponseBody = z.object({
   id: z.string(),
+  isCached: z.boolean(),
 });
 export type CreateWaitpointResponseBody = z.infer<typeof CreateWaitpointResponseBody>;
 

packages/trigger-sdk/src/v3/resumeTokens.ts

@@ -19,6 +19,19 @@ function createResumeToken(
       tracer,
       name: "resumeTokens.create()",
       icon: "wait",
+      attributes: {
+        idempotencyKey: options?.idempotencyKey,
+        idempotencyKeyTTL: options?.idempotencyKeyTTL,
+        timeout: options?.timeout
+          ? typeof options.timeout === "string"
+            ? options.timeout
+            : options.timeout.toISOString()
+          : undefined,
+      },
+      onResponseBody: (body: CreateWaitpointResponseBody, span) => {
+        span.setAttribute("id", body.id);
+        span.setAttribute("isCached", body.isCached);
+      },
     },
     requestOptions
   );

references/hello-world/src/trigger/resume-tokens.ts

@@ -5,7 +5,15 @@ export const resumeToken = task({
   run: async (payload: any, { ctx }) => {
     logger.log("Hello, world", { payload });
 
-    const token = await resumeTokens.create();
+    const idempotencyKey = "a";
+
+    const token = await resumeTokens.create({
+      idempotencyKey,
+      timeout: new Date(Date.now() + 5_000),
+    });
     logger.log("Token", token);
+
+    const token2 = await resumeTokens.create({ idempotencyKey, timeout: "10s" });
+    logger.log("Token2", token2);
   },
 });