Completing waitpoints using public access tokens

Author: ericallam

apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts

@@ -45,6 +45,12 @@ const { action, loader } = createActionApiRoute(
         throw json({ error: "Waitpoint not found" }, { status: 404 });
       }
 
+      // if (waitpoint.status === "COMPLETED") {
+      //   return json<CompleteWaitpointTokenResponseBody>({
+      //     success: true,
+      //   });
+      // }
+
       const stringifiedData = await stringifyIO(body.data);
       const finalData = await conditionallyExportPacket(
         stringifiedData,

internal-packages/run-engine/src/engine/systems/waitpointSystem.ts

@@ -1,4 +1,4 @@
-import { timeoutError } from "@trigger.dev/core/v3";
+import { timeoutError, tryCatch } from "@trigger.dev/core/v3";
 import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import {
   $transaction,
@@ -66,65 +66,51 @@ export class WaitpointSystem {
       isError: boolean;
     };
   }): Promise<Waitpoint> {
-    const result = await $transaction(
-      this.$.prisma,
-      async (tx) => {
-        // 1. Find the TaskRuns blocked by this waitpoint
-        const affectedTaskRuns = await tx.taskRunWaitpoint.findMany({
-          where: { waitpointId: id },
-          select: { taskRunId: true, spanIdToComplete: true, createdAt: true },
-        });
+    // 1. Find the TaskRuns blocked by this waitpoint
+    const affectedTaskRuns = await this.$.prisma.taskRunWaitpoint.findMany({
+      where: { waitpointId: id },
+      select: { taskRunId: true, spanIdToComplete: true, createdAt: true },
+    });
 
-        if (affectedTaskRuns.length === 0) {
-          this.$.logger.warn(`completeWaitpoint: No TaskRunWaitpoints found for waitpoint`, {
-            waitpointId: id,
-          });
-        }
+    if (affectedTaskRuns.length === 0) {
+      this.$.logger.debug(`completeWaitpoint: No TaskRunWaitpoints found for waitpoint`, {
+        waitpointId: id,
+      });
+    }
 
-        // 2. Update the waitpoint to completed (only if it's pending)
-        let waitpoint: Waitpoint | null = null;
-        try {
-          waitpoint = await tx.waitpoint.update({
-            where: { id, status: "PENDING" },
-            data: {
-              status: "COMPLETED",
-              completedAt: new Date(),
-              output: output?.value,
-              outputType: output?.type,
-              outputIsError: output?.isError,
-            },
-          });
-        } catch (error) {
-          if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2025") {
-            waitpoint = await tx.waitpoint.findFirst({
-              where: { id },
-            });
-          } else {
-            this.$.logger.log("completeWaitpoint: error updating waitpoint:", { error });
-            throw error;
-          }
-        }
+    let [waitpointError, waitpoint] = await tryCatch(
+      this.$.prisma.waitpoint.update({
+        where: { id, status: "PENDING" },
+        data: {
+          status: "COMPLETED",
+          completedAt: new Date(),
+          output: output?.value,
+          outputType: output?.type,
+          outputIsError: output?.isError,
+        },
+      })
+    );
 
-        return { waitpoint, affectedTaskRuns };
-      },
-      (error) => {
-        this.$.logger.error(`completeWaitpoint: Error completing waitpoint ${id}, retrying`, {
-          error,
+    if (waitpointError) {
+      if (
+        waitpointError instanceof Prisma.PrismaClientKnownRequestError &&
+        waitpointError.code === "P2025"
+      ) {
+        waitpoint = await this.$.prisma.waitpoint.findFirst({
+          where: { id },
         });
-        throw error;
+      } else {
+        this.$.logger.log("completeWaitpoint: error updating waitpoint:", { waitpointError });
+        throw waitpointError;
       }
-    );
-
-    if (!result) {
-      throw new Error(`Waitpoint couldn't be updated`);
     }
 
-    if (!result.waitpoint) {
+    if (!waitpoint) {
       throw new Error(`Waitpoint ${id} not found`);
     }
 
     //schedule trying to continue the runs
-    for (const run of result.affectedTaskRuns) {
+    for (const run of affectedTaskRuns) {
       await this.$.worker.enqueue({
         //this will debounce the call
         id: `continueRunIfUnblocked:${run.taskRunId}`,
@@ -148,7 +134,7 @@ export class WaitpointSystem {
       }
     }
 
-    return result.waitpoint;
+    return waitpoint;
   }
 
   /**

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

@@ -692,7 +692,7 @@ export class ApiClient {
               ...claims,
               scopes: [`write:waitpoints:${data.id}`],
             },
-            expirationTime: "1h",
+            expirationTime: "24h",
           });
 
           return {

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

@@ -53,14 +53,16 @@ type PublicTokenPermissionProperties = {
    * Grant access to specific batch runs
    */
   batch?: string | string[];
+
+  /**
+   * Grant access to specific waitpoints
+   */
+  waitpoints?: string | string[];
 };
 
 export type PublicTokenPermissions = {
   read?: PublicTokenPermissionProperties;
 
-  /**
-   * @deprecated use trigger instead
-   */
   write?: PublicTokenPermissionProperties;
 
   /**

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

@@ -24,6 +24,12 @@ import { tracer } from "./tracer.js";
 import { conditionallyImportAndParsePacket } from "@trigger.dev/core/v3/utils/ioSerialization";
 import { SpanStatusCode } from "@opentelemetry/api";
 
+export type CreateWaitpointTokenResponse = Prettify<
+  CreateWaitpointTokenResponseBody & {
+    publicAccessToken: string;
+  }
+>;
+
 /**
  * This creates a waitpoint token.
  * You can use this to pause a run until you complete the waitpoint (or it times out).
@@ -48,10 +54,6 @@ import { SpanStatusCode } from "@opentelemetry/api";
  * @param requestOptions - The request options for the waitpoint token.
  * @returns The waitpoint token.
  */
-export type CreateWaitpointTokenResponse = CreateWaitpointTokenResponseBody & {
-  publicAccessToken: string;
-};
-
 function createToken(
   options?: CreateWaitpointTokenRequestBody,
   requestOptions?: ApiRequestOptions

references/hello-world/src/trigger/waits.ts

@@ -1,4 +1,4 @@
-import { logger, wait, task, retry, idempotencyKeys } from "@trigger.dev/sdk/v3";
+import { logger, wait, task, retry, idempotencyKeys, auth } from "@trigger.dev/sdk/v3";
 
 type Token = {
   status: "approved" | "pending" | "rejected";
@@ -8,13 +8,15 @@ export const waitToken = task({
   id: "wait-token",
   run: async ({
     completeBeforeWaiting = false,
+    completeWithPublicToken = false,
     idempotencyKey,
     idempotencyKeyTTL,
     completionDelay,
     timeout,
     tags,
   }: {
     completeBeforeWaiting?: boolean;
+    completeWithPublicToken?: boolean;
     idempotencyKey?: string;
     idempotencyKeyTTL?: string;
     completionDelay?: number;
@@ -39,6 +41,28 @@ export const waitToken = task({
     });
     logger.log("Token2", token2);
 
+    const publicAccessToken = await auth.createPublicToken({
+      scopes: {
+        write: {
+          waitpoints: token.id,
+        },
+      },
+      expirationTime: "1h",
+    });
+
+    logger.log("Public access token", { publicAccessToken });
+
+    if (completeWithPublicToken) {
+      await auth.withAuth(
+        {
+          accessToken: token.publicAccessToken,
+        },
+        async () => {
+          await wait.completeToken<Token>(token.id, { status: "approved" });
+        }
+      );
+    }
+
     if (completeBeforeWaiting) {
       await wait.completeToken<Token>(token.id, { status: "approved" });
       await wait.for({ seconds: 5 });