Initial implentation ofr wait.forHttpCallback()

Author: matt-aitken

apps/webapp/app/routes/engine.v1.waitpoints.http-callback.ts

@@ -0,0 +1,107 @@
+import { json } from "@remix-run/server-runtime";
+import {
+  type CreateWaitpointHttpCallbackResponseBody,
+  CreateWaitpointTokenRequestBody,
+} from "@trigger.dev/core/v3";
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import { env } from "~/env.server";
+import { createWaitpointTag, MAX_TAGS_PER_WAITPOINT } from "~/models/waitpointTag.server";
+import {
+  ApiWaitpointListPresenter,
+  ApiWaitpointListSearchParams,
+} from "~/presenters/v3/ApiWaitpointListPresenter.server";
+import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import {
+  createActionApiRoute,
+  createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
+import { parseDelay } from "~/utils/delays";
+import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
+import { engine } from "~/v3/runEngine.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
+
+export const loader = createLoaderApiRoute(
+  {
+    searchParams: ApiWaitpointListSearchParams,
+    findResource: async () => 1, // This is a dummy function, we don't need to find a resource
+  },
+  async ({ searchParams, authentication }) => {
+    const presenter = new ApiWaitpointListPresenter();
+    const result = await presenter.call(authentication.environment, "HTTP_CALLBACK", searchParams);
+
+    return json(result);
+  }
+);
+
+const { action } = createActionApiRoute(
+  {
+    body: CreateWaitpointTokenRequestBody,
+    maxContentLength: 1024 * 10, // 10KB
+    method: "POST",
+  },
+  async ({ authentication, body }) => {
+    try {
+      const idempotencyKeyExpiresAt = body.idempotencyKeyTTL
+        ? resolveIdempotencyKeyTTL(body.idempotencyKeyTTL)
+        : undefined;
+
+      const timeout = await parseDelay(body.timeout);
+
+      //upsert tags
+      let tags: { id: string; name: string }[] = [];
+      const bodyTags = typeof body.tags === "string" ? [body.tags] : body.tags;
+
+      if (bodyTags && bodyTags.length > MAX_TAGS_PER_WAITPOINT) {
+        throw new ServiceValidationError(
+          `Waitpoints can only have ${MAX_TAGS_PER_WAITPOINT} tags, you're trying to set ${bodyTags.length}.`
+        );
+      }
+
+      if (bodyTags && bodyTags.length > 0) {
+        for (const tag of bodyTags) {
+          const tagRecord = await createWaitpointTag({
+            tag,
+            environmentId: authentication.environment.id,
+            projectId: authentication.environment.projectId,
+          });
+          if (tagRecord) {
+            tags.push(tagRecord);
+          }
+        }
+      }
+
+      const result = await engine.createManualWaitpoint({
+        environmentId: authentication.environment.id,
+        projectId: authentication.environment.projectId,
+        idempotencyKey: body.idempotencyKey,
+        idempotencyKeyExpiresAt,
+        timeout,
+        resolver: "HTTP_CALLBACK",
+        tags: bodyTags,
+      });
+
+      return json<CreateWaitpointHttpCallbackResponseBody>(
+        {
+          id: WaitpointId.toFriendlyId(result.waitpoint.id),
+          url: `${
+            env.API_ORIGIN ?? env.APP_ORIGIN
+          }/api/v1/waitpoints/http-callback/${WaitpointId.toFriendlyId(
+            result.waitpoint.id
+          )}/callback`,
+          isCached: result.isCached,
+        },
+        { status: 200 }
+      );
+    } catch (error) {
+      if (error instanceof ServiceValidationError) {
+        return json({ error: error.message }, { status: 422 });
+      } else if (error instanceof Error) {
+        return json({ error: error.message }, { status: 500 });
+      }
+
+      return json({ error: "Something went wrong" }, { status: 500 });
+    }
+  }
+);
+
+export { action };

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

@@ -12,6 +12,7 @@ import {
   CreateEnvironmentVariableRequestBody,
   CreateScheduleOptions,
   CreateUploadPayloadUrlResponseBody,
+  CreateWaitpointHttpCallbackResponseBody,
   CreateWaitpointTokenRequestBody,
   CreateWaitpointTokenResponseBody,
   DeletedScheduleObject,
@@ -790,6 +791,24 @@ export class ApiClient {
     );
   }
 
+  createWaitpointHttpCallback(
+    options: CreateWaitpointTokenRequestBody,
+    requestOptions?: ZodFetchOptions
+  ) {
+    return zodfetch(
+      CreateWaitpointHttpCallbackResponseBody,
+      `${this.baseUrl}/engine/v1/waitpoints/http-callback`,
+      {
+        method: "POST",
+        headers: this.#getHeaders(false),
+        body: JSON.stringify(options),
+      },
+      {
+        ...mergeRequestOptions(this.defaultRequestOptions, requestOptions),
+      }
+    ) as ApiPromise<CreateWaitpointHttpCallbackResponseBody>;
+  }
+
   async waitForDuration(
     runId: string,
     body: WaitForDurationRequestBody,

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

@@ -1010,6 +1010,15 @@ export const WaitForWaitpointTokenResponseBody = z.object({
 });
 export type WaitForWaitpointTokenResponseBody = z.infer<typeof WaitForWaitpointTokenResponseBody>;
 
+export const CreateWaitpointHttpCallbackResponseBody = z.object({
+  id: z.string(),
+  url: z.string(),
+  isCached: z.boolean(),
+});
+export type CreateWaitpointHttpCallbackResponseBody = z.infer<
+  typeof CreateWaitpointHttpCallbackResponseBody
+>;
+
 export const WaitForDurationRequestBody = z.object({
   /**
    * An optional idempotency key for the waitpoint.

packages/core/src/v3/types/waitpoints.ts

@@ -6,7 +6,7 @@ export type HttpCallbackResultTypeFromSchema<TSchema extends HttpCallbackSchema>
 export type HttpCallbackResult<TResult> =
   | {
       ok: true;
-      result: TResult;
+      output: TResult;
     }
   | {
       ok: false;

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

@@ -23,6 +23,7 @@ import {
   HttpCallbackSchema,
   HttpCallbackResultTypeFromSchema,
   HttpCallbackResult,
+  tryCatch,
 } from "@trigger.dev/core/v3";
 import { tracer } from "./tracer.js";
 import { conditionallyImportAndParsePacket } from "@trigger.dev/core/v3/utils/ioSerialization";
@@ -312,47 +313,6 @@ async function completeToken<T>(
   return apiClient.completeWaitpointToken(tokenId, { data }, $requestOptions);
 }
 
-async function forHttpCallback<TResult>(
-  callback: (url: string) => Promise<void>,
-  options?: {
-    timeout?: string | Date | undefined;
-  }
-): Promise<HttpCallbackResult<TResult>> {
-  //TODO:
-  // Support a schema passed in, infer the type, or a generic supplied type
-  // Support a timeout passed in
-  // 1. Make an API call to engine.trigger.dev/v1/waitpoints/http-callback/create. New Waitpoint type "HTTPCallback"
-  // 2. Return the url and a waitpoint id (but don't block the run yet)
-  // 3. Create a span for the main call
-  // 4. Set the url and waitpoint entity type and id as attributes on the parent span
-  // 5. Create a span around the callback
-  // 6. Deal with errors thrown in the callback use `tryCatch()`
-  // 7. If that callback is successfully called, wait for the waitpoint with an API call to engine.trigger.dev/v1/waitpoints/http-callback/{waitpointId}/block
-  // 8. Wait for the waitpoint in the runtime
-  // 9. On the backend when the API is hit, complete the waitpoint with the result api.trigger.dev/v1/waitpoints/http-callback/{waitpointId}/callback
-  // 10. Receive the result here and import the packet, then get the result in the right format
-  // 11. Make unwrap work
-
-  const url = "https://trigger.dev/wait/success";
-
-  const result = await callback(url);
-
-  return {
-    ok: true,
-    output: result,
-  } as any;
-}
-
-async function forHttpCallbackWithSchema<TSchema extends HttpCallbackSchema>(
-  schema: TSchema,
-  callback: (successUrl: string, failureUrl: string) => Promise<void>,
-  options?: {
-    timeout?: string | Date | undefined;
-  }
-): Promise<HttpCallbackResult<HttpCallbackResultTypeFromSchema<TSchema>>> {
-  return {} as any;
-}
-
 export type CommonWaitOptions = {
   /**
    * An optional idempotency key for the waitpoint.
@@ -690,6 +650,116 @@ export const wait = {
       }
     );
   },
+  async forHttpCallback<TResult>(
+    callback: (url: string) => Promise<void>,
+    options?: CreateWaitpointTokenRequestBody & {
+      releaseConcurrency?: boolean;
+    },
+    requestOptions?: ApiRequestOptions
+  ): Promise<HttpCallbackResult<TResult>> {
+    const ctx = taskContext.ctx;
+
+    if (!ctx) {
+      throw new Error("wait.forHttpCallback can only be used from inside a task.run()");
+    }
+
+    const apiClient = apiClientManager.clientOrThrow();
+
+    const waitpoint = await apiClient.createWaitpointHttpCallback(options ?? {}, requestOptions);
+
+    return tracer.startActiveSpan(
+      `wait.forHttpCallback()`,
+      async (span) => {
+        const [error] = await tryCatch(callback(waitpoint.url));
+
+        if (error) {
+          throw new Error(`You threw an error in your callback: ${error.message}`, {
+            cause: error,
+          });
+        }
+
+        const response = await apiClient.waitForWaitpointToken({
+          runFriendlyId: ctx.run.id,
+          waitpointFriendlyId: waitpoint.id,
+          releaseConcurrency: options?.releaseConcurrency,
+        });
+
+        if (!response.success) {
+          throw new Error(`Failed to wait for wait for HTTP callback ${waitpoint.id}`);
+        }
+
+        const result = await runtime.waitUntil(waitpoint.id);
+
+        const data = result.output
+          ? await conditionallyImportAndParsePacket(
+              { data: result.output, dataType: result.outputType ?? "application/json" },
+              apiClient
+            )
+          : undefined;
+
+        if (result.ok) {
+          return {
+            ok: result.ok,
+            output: data,
+          };
+        } else {
+          const error = new WaitpointTimeoutError(data.message);
+
+          span.recordException(error);
+          span.setStatus({
+            code: SpanStatusCode.ERROR,
+          });
+
+          return {
+            ok: result.ok,
+            error,
+          };
+        }
+      },
+      {
+        attributes: {
+          [SemanticInternalAttributes.STYLE_ICON]: "wait-http-callback",
+          [SemanticInternalAttributes.ENTITY_TYPE]: "waitpoint",
+          [SemanticInternalAttributes.ENTITY_ID]: waitpoint.id,
+          ...accessoryAttributes({
+            items: [
+              {
+                text: waitpoint.id,
+                variant: "normal",
+              },
+            ],
+            style: "codepath",
+          }),
+          id: waitpoint.id,
+          isCached: waitpoint.isCached,
+          idempotencyKey: options?.idempotencyKey,
+          idempotencyKeyTTL: options?.idempotencyKeyTTL,
+          timeout: options?.timeout
+            ? typeof options.timeout === "string"
+              ? options.timeout
+              : options.timeout.toISOString()
+            : undefined,
+          tags: options?.tags,
+          url: waitpoint.url,
+        },
+      }
+    );
+
+    //TODO:
+    // Support a schema passed in, infer the type, or a generic supplied type
+    // Support a timeout passed in
+    // 1. Make an API call to engine.trigger.dev/v1/waitpoints/http-callback/create. New Waitpoint type "HTTPCallback"
+    // 2. Return the url and a waitpoint id (but don't block the run yet)
+    // 3. Create a span for the main call
+    // 4. Set the url and waitpoint entity type and id as attributes on the parent span
+    // 5. Create a span around the callback
+    // 6. Deal with errors thrown in the callback use `tryCatch()`
+    // 7. If that callback is successfully called, wait for the waitpoint with an API call to engine.trigger.dev/v1/waitpoints/http-callback/{waitpointId}/block
+    // 8. Wait for the waitpoint in the runtime
+    // 9. On the backend when the API is hit, complete the waitpoint with the result api.trigger.dev/v1/waitpoints/http-callback/{waitpointId}/callback
+    // 10. Receive the result here and import the packet, then get the result in the right format
+    // 11. Make unwrap work
+  },
 };
 
 function nameForWaitOptions(options: WaitForOptions): string {

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

@@ -140,3 +140,19 @@ export const waitForDuration = task({
     );
   },
 });
+
+export const waitHttpCallback = task({
+  id: "wait-http-callback",
+  run: async () => {
+    const result = await wait.forHttpCallback<{ foo: string }>(
+      async (url) => {
+        logger.log(`Wait for HTTP callback ${url}`);
+      },
+      {
+        timeout: "60s",
+      }
+    );
+
+    logger.log("Wait for HTTP callback completed", result);
+  },
+});