Waitpoint token callback URLs

.changeset/curvy-dogs-share.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Added wait.createHttpCallback() to allow offloading work to 3rd party APIs

apps/webapp/app/components/navigation/SideMenu.tsx

@@ -80,6 +80,7 @@ import { HelpAndFeedback } from "./HelpAndFeedbackPopover";
 import { SideMenuHeader } from "./SideMenuHeader";
 import { SideMenuItem } from "./SideMenuItem";
 import { SideMenuSection } from "./SideMenuSection";
+import { HttpCallbackIcon } from "~/assets/icons/HttpCallbackIcon";
 
 type SideMenuUser = Pick<User, "email" | "admin"> & { isImpersonating: boolean };
 export type SideMenuProject = Pick<

apps/webapp/app/components/runs/v3/RunIcon.tsx

@@ -19,6 +19,7 @@ import { FunctionIcon } from "~/assets/icons/FunctionIcon";
 import { TriggerIcon } from "~/assets/icons/TriggerIcon";
 import { PythonLogoIcon } from "~/assets/icons/PythonLogoIcon";
 import { TraceIcon } from "~/assets/icons/TraceIcon";
+import { WaitpointTokenIcon } from "~/assets/icons/WaitpointTokenIcon";
 
 type TaskIconProps = {
   name: string | undefined;
@@ -75,6 +76,10 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
       return <TriggerIcon className={cn(className, "text-orange-500")} />;
     case "python":
       return <PythonLogoIcon className={className} />;
+    case "wait-token":
+      return <WaitpointTokenIcon className={cn(className, "text-sky-500")} />;
+    case "function":
+      return <FunctionIcon className={cn(className, "text-text-dimmed")} />;
     //log levels
     case "debug":
     case "log":

apps/webapp/app/components/runs/v3/WaitpointDetails.tsx

@@ -11,6 +11,7 @@ import { v3WaitpointTokenPath, v3WaitpointTokensPath } from "~/utils/pathBuilder
 import { PacketDisplay } from "./PacketDisplay";
 import { WaitpointStatusCombo } from "./WaitpointStatus";
 import { RunTag } from "./RunTag";
+import { ClipboardField } from "~/components/primitives/ClipboardField";
 
 export function WaitpointDetailTable({
   waitpoint,
@@ -50,6 +51,12 @@ export function WaitpointDetailTable({
           )}
         </Property.Value>
       </Property.Item>
+      <Property.Item>
+        <Property.Label>Callback URL</Property.Label>
+        <Property.Value className="my-1">
+          <ClipboardField value={waitpoint.callbackUrl} variant={"secondary/small"} />
+        </Property.Value>
+      </Property.Item>
       <Property.Item>
         <Property.Label>Idempotency key</Property.Label>
         <Property.Value>

apps/webapp/app/presenters/v3/ApiWaitpointTokenListPresenter.server.ts → apps/webapp/app/presenters/v3/ApiWaitpointListPresenter.server.ts

@@ -1,16 +1,12 @@
-import { RuntimeEnvironmentType, WaitpointTokenStatus } from "@trigger.dev/core/v3";
+import { type RuntimeEnvironmentType, WaitpointTokenStatus } from "@trigger.dev/core/v3";
+import { type RunEngineVersion, type WaitpointResolver } from "@trigger.dev/database";
 import { z } from "zod";
-import { BasePresenter } from "./basePresenter.server";
 import { CoercedDate } from "~/utils/zod";
-import { AuthenticatedEnvironment } from "@internal/run-engine";
-import {
-  WaitpointTokenListOptions,
-  WaitpointTokenListPresenter,
-} from "./WaitpointTokenListPresenter.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
-import { RunEngineVersion } from "@trigger.dev/database";
+import { BasePresenter } from "./basePresenter.server";
+import { type WaitpointListOptions, WaitpointListPresenter } from "./WaitpointListPresenter.server";
 
-export const ApiWaitpointTokenListSearchParams = z.object({
+export const ApiWaitpointListSearchParams = z.object({
   "page[size]": z.coerce.number().int().positive().min(1).max(100).optional(),
   "page[after]": z.string().optional(),
   "page[before]": z.string().optional(),
@@ -61,9 +57,9 @@ export const ApiWaitpointTokenListSearchParams = z.object({
   "filter[createdAt][to]": CoercedDate,
 });
 
-type ApiWaitpointTokenListSearchParams = z.infer<typeof ApiWaitpointTokenListSearchParams>;
+type ApiWaitpointListSearchParams = z.infer<typeof ApiWaitpointListSearchParams>;
 
-export class ApiWaitpointTokenListPresenter extends BasePresenter {
+export class ApiWaitpointListPresenter extends BasePresenter {
   public async call(
     environment: {
       id: string;
@@ -72,11 +68,12 @@ export class ApiWaitpointTokenListPresenter extends BasePresenter {
         id: string;
         engine: RunEngineVersion;
       };
+      apiKey: string;
     },
-    searchParams: ApiWaitpointTokenListSearchParams
+    searchParams: ApiWaitpointListSearchParams
   ) {
     return this.trace("call", async (span) => {
-      const options: WaitpointTokenListOptions = {
+      const options: WaitpointListOptions = {
         environment,
       };
 
@@ -118,7 +115,7 @@ export class ApiWaitpointTokenListPresenter extends BasePresenter {
         options.to = searchParams["filter[createdAt][to]"].getTime();
       }
 
-      const presenter = new WaitpointTokenListPresenter();
+      const presenter = new WaitpointListPresenter();
       const result = await presenter.call(options);
 
       if (!result.success) {

apps/webapp/app/presenters/v3/ApiWaitpointPresenter.server.ts

@@ -3,7 +3,7 @@ import { type RunEngineVersion } from "@trigger.dev/database";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { BasePresenter } from "./basePresenter.server";
 import { WaitpointPresenter } from "./WaitpointPresenter.server";
-import { waitpointStatusToApiStatus } from "./WaitpointTokenListPresenter.server";
+import { waitpointStatusToApiStatus } from "./WaitpointListPresenter.server";
 
 export class ApiWaitpointPresenter extends BasePresenter {
   public async call(

apps/webapp/app/presenters/v3/WaitpointTokenListPresenter.server.ts → apps/webapp/app/presenters/v3/WaitpointListPresenter.server.ts

@@ -1,6 +1,7 @@
 import parse from "parse-duration";
 import {
   Prisma,
+  type WaitpointResolver,
   type RunEngineVersion,
   type RuntimeEnvironmentType,
   type WaitpointStatus,
@@ -11,17 +12,19 @@ import { BasePresenter } from "./basePresenter.server";
 import { type WaitpointSearchParams } from "~/components/runs/v3/WaitpointTokenFilters";
 import { determineEngineVersion } from "~/v3/engineVersion.server";
 import { type WaitpointTokenStatus, type WaitpointTokenItem } from "@trigger.dev/core/v3";
+import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
 
 const DEFAULT_PAGE_SIZE = 25;
 
-export type WaitpointTokenListOptions = {
+export type WaitpointListOptions = {
   environment: {
     id: string;
     type: RuntimeEnvironmentType;
     project: {
       id: string;
       engine: RunEngineVersion;
     };
+    apiKey: string;
   };
   // filters
   id?: string;
@@ -40,7 +43,7 @@ export type WaitpointTokenListOptions = {
 type Result =
   | {
       success: true;
-      tokens: WaitpointTokenItem[];
+      tokens: (WaitpointTokenItem & { callbackUrl: string })[];
       pagination: {
         next: string | undefined;
         previous: string | undefined;
@@ -63,7 +66,7 @@ type Result =
       filters: undefined;
     };
 
-export class WaitpointTokenListPresenter extends BasePresenter {
+export class WaitpointListPresenter extends BasePresenter {
   public async call({
     environment,
     id,
@@ -76,7 +79,7 @@ export class WaitpointTokenListPresenter extends BasePresenter {
     direction = "forward",
     cursor,
     pageSize = DEFAULT_PAGE_SIZE,
-  }: WaitpointTokenListOptions): Promise<Result> {
+  }: WaitpointListOptions): Promise<Result> {
     const engineVersion = await determineEngineVersion({ environment });
     if (engineVersion === "V1") {
       return {
@@ -165,8 +168,8 @@ export class WaitpointTokenListPresenter extends BasePresenter {
       ${sqlDatabaseSchema}."Waitpoint" w
     WHERE
       w."environmentId" = ${environment.id}
-      AND w.type = 'MANUAL'
-      -- cursor
+    AND w.type = 'MANUAL'      
+    -- cursor
       ${
         cursor
           ? direction === "forward"
@@ -263,6 +266,7 @@ export class WaitpointTokenListPresenter extends BasePresenter {
       success: true,
       tokens: tokensToReturn.map((token) => ({
         id: token.friendlyId,
+        callbackUrl: generateHttpCallbackUrl(token.id, environment.apiKey),
         status: waitpointStatusToApiStatus(token.status, token.outputIsError),
         completedAt: token.completedAt ?? undefined,
         timeoutAt: token.completedAfter ?? undefined,

apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts

@@ -2,7 +2,10 @@ import { isWaitpointOutputTimeout, prettyPrintPacket } from "@trigger.dev/core/v
 import { logger } from "~/services/logger.server";
 import { BasePresenter } from "./basePresenter.server";
 import { type RunListItem, RunListPresenter } from "./RunListPresenter.server";
-import { waitpointStatusToApiStatus } from "./WaitpointTokenListPresenter.server";
+import { waitpointStatusToApiStatus } from "./WaitpointListPresenter.server";
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import { env } from "~/env.server";
+import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
 
 export type WaitpointDetail = NonNullable<Awaited<ReturnType<WaitpointPresenter["call"]>>>;
 
@@ -22,6 +25,7 @@ export class WaitpointPresenter extends BasePresenter {
         environmentId,
       },
       select: {
+        id: true,
         friendlyId: true,
         type: true,
         status: true,
@@ -42,6 +46,11 @@ export class WaitpointPresenter extends BasePresenter {
           take: 5,
         },
         tags: true,
+        environment: {
+          select: {
+            apiKey: true,
+          },
+        },
       },
     });
 
@@ -83,6 +92,7 @@ export class WaitpointPresenter extends BasePresenter {
     return {
       id: waitpoint.friendlyId,
       type: waitpoint.type,
+      callbackUrl: generateHttpCallbackUrl(waitpoint.id, waitpoint.environment.apiKey),
       status: waitpointStatusToApiStatus(waitpoint.status, waitpoint.outputIsError),
       idempotencyKey: waitpoint.idempotencyKey,
       userProvidedIdempotencyKey: waitpoint.userProvidedIdempotencyKey,

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.tokens/route.tsx

@@ -7,6 +7,7 @@ import { NoWaitpointTokens } from "~/components/BlankStatePanels";
 import { MainCenteredContainer, PageBody, PageContainer } from "~/components/layout/AppLayout";
 import { ListPagination } from "~/components/ListPagination";
 import { LinkButton } from "~/components/primitives/Buttons";
+import { ClipboardField } from "~/components/primitives/ClipboardField";
 import { CopyableText } from "~/components/primitives/CopyableText";
 import { DateTime } from "~/components/primitives/DateTime";
 import { NavBar, PageAccessories, PageTitle } from "~/components/primitives/PageHeader";
@@ -36,7 +37,7 @@ import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
-import { WaitpointTokenListPresenter } from "~/presenters/v3/WaitpointTokenListPresenter.server";
+import { WaitpointListPresenter } from "~/presenters/v3/WaitpointListPresenter.server";
 import { requireUserId } from "~/services/session.server";
 import { docsPath, EnvironmentParamSchema, v3WaitpointTokenPath } from "~/utils/pathBuilder";
 
@@ -84,7 +85,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   }
 
   try {
-    const presenter = new WaitpointTokenListPresenter();
+    const presenter = new WaitpointListPresenter();
     const result = await presenter.call({
       environment,
       ...searchParams,
@@ -143,6 +144,7 @@ export default function Page() {
                       <TableRow>
                         <TableHeaderCell className="w-[1%]">Created</TableHeaderCell>
                         <TableHeaderCell className="w-[20%]">ID</TableHeaderCell>
+                        <TableHeaderCell className="w-[20%]">Callback URL</TableHeaderCell>
                         <TableHeaderCell className="w-[20%]">Status</TableHeaderCell>
                         <TableHeaderCell className="w-[20%]">Completed</TableHeaderCell>
                         <TableHeaderCell className="w-[20%]">Idempotency Key</TableHeaderCell>
@@ -178,6 +180,12 @@ export default function Page() {
                               <TableCell to={path}>
                                 <CopyableText value={token.id} className="font-mono" />
                               </TableCell>
+                              <TableCell to={path}>
+                                <ClipboardField
+                                  value={token.callbackUrl}
+                                  variant={"secondary/small"}
+                                />
+                              </TableCell>
                               <TableCell to={path}>
                                 <WaitpointStatusCombo status={token.status} className="text-xs" />
                               </TableCell>

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

@@ -0,0 +1,91 @@
+import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import {
+  type CompleteWaitpointTokenResponseBody,
+  conditionallyExportPacket,
+  stringifyIO,
+} from "@trigger.dev/core/v3";
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { env } from "~/env.server";
+import { verifyHttpCallbackHash } from "~/services/httpCallback.server";
+import { logger } from "~/services/logger.server";
+import { engine } from "~/v3/runEngine.server";
+
+const paramsSchema = z.object({
+  waitpointFriendlyId: z.string(),
+  hash: z.string(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  if (request.method.toUpperCase() !== "POST") {
+    return json({ error: "Method not allowed" }, { status: 405, headers: { Allow: "POST" } });
+  }
+
+  const contentLength = request.headers.get("content-length");
+  if (!contentLength) {
+    return json({ error: "Content-Length header is required" }, { status: 411 });
+  }
+
+  if (parseInt(contentLength) > env.TASK_PAYLOAD_MAXIMUM_SIZE) {
+    return json({ error: "Request body too large" }, { status: 413 });
+  }
+
+  const { waitpointFriendlyId, hash } = paramsSchema.parse(params);
+  const waitpointId = WaitpointId.toId(waitpointFriendlyId);
+
+  try {
+    const waitpoint = await $replica.waitpoint.findFirst({
+      where: {
+        id: waitpointId,
+      },
+      include: {
+        environment: {
+          select: {
+            apiKey: true,
+          },
+        },
+      },
+    });
+
+    if (!waitpoint) {
+      return json({ error: "Waitpoint not found" }, { status: 404 });
+    }
+
+    if (!verifyHttpCallbackHash(waitpoint.id, hash, waitpoint.environment.apiKey)) {
+      return json({ error: "Invalid URL, hash doesn't match" }, { status: 401 });
+    }
+
+    if (waitpoint.status === "COMPLETED") {
+      return json<CompleteWaitpointTokenResponseBody>({
+        success: true,
+      });
+    }
+
+    // If the request body is not valid JSON, return an empty object
+    const body = await request.json().catch(() => ({}));
+
+    const stringifiedData = await stringifyIO(body);
+    const finalData = await conditionallyExportPacket(
+      stringifiedData,
+      `${waitpointId}/waitpoint/http-callback`
+    );
+
+    const result = await engine.completeWaitpoint({
+      id: waitpointId,
+      output: finalData.data
+        ? { type: finalData.dataType, value: finalData.data, isError: false }
+        : undefined,
+    });
+
+    return json<CompleteWaitpointTokenResponseBody>(
+      {
+        success: true,
+      },
+      { status: 200 }
+    );
+  } catch (error) {
+    logger.error("Failed to complete HTTP callback", { error });
+    throw json({ error: "Failed to complete HTTP callback" }, { status: 500 });
+  }
+}