feat/realtime-streams

.changeset/swift-glasses-mate.md

@@ -0,0 +1,10 @@
+---
+"@trigger.dev/react-hooks": patch
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/build": patch
+"@trigger.dev/core": patch
+"@trigger.dev/rsc": patch
+---
+
+Realtime streams

.vscode/launch.json

@@ -13,6 +13,15 @@
       "cwd": "${workspaceFolder}",
       "sourceMaps": true
     },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug realtimeStreams.test.ts",
+      "command": "pnpm run test -t RealtimeStreams",
+      "envFile": "${workspaceFolder}/.env",
+      "cwd": "${workspaceFolder}/apps/webapp",
+      "sourceMaps": true
+    },
     {
       "type": "chrome",
       "request": "launch",
@@ -36,6 +45,14 @@
       "cwd": "${workspaceFolder}/references/v3-catalog",
       "sourceMaps": true
     },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug Dev Next.js Realtime",
+      "command": "pnpm exec trigger dev",
+      "cwd": "${workspaceFolder}/references/nextjs-realtime",
+      "sourceMaps": true
+    },
     {
       "type": "node-terminal",
       "request": "launch",

apps/webapp/app/env.server.ts

@@ -32,6 +32,7 @@ const EnvironmentSchema = z.object({
   LOGIN_ORIGIN: z.string().default("http://localhost:3030"),
   APP_ORIGIN: z.string().default("http://localhost:3030"),
   API_ORIGIN: z.string().optional(),
+  STREAM_ORIGIN: z.string().optional(),
   ELECTRIC_ORIGIN: z.string().default("http://localhost:3060"),
   APP_ENV: z.string().default(process.env.NODE_ENV),
   SERVICE_NAME: z.string().default("trigger.dev webapp"),

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

@@ -210,7 +210,7 @@ export class SpanPresenter extends BasePresenter {
     const span = await eventRepository.getSpan(spanId, run.traceId);
 
     const metadata = run.metadata
-      ? await prettyPrintPacket(run.metadata, run.metadataType)
+      ? await prettyPrintPacket(run.metadata, run.metadataType, { filteredKeys: ["$$streams"] })
       : undefined;
 
     const context = {

apps/webapp/app/routes/api.v1.packets.$.ts

@@ -2,6 +2,7 @@ import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { generatePresignedUrl } from "~/v3/r2.server";
 
 const ParamsSchema = z.object({
@@ -39,28 +40,27 @@ export async function action({ request, params }: ActionFunctionArgs) {
   return json({ presignedUrl });
 }
 
-export async function loader({ request, params }: ActionFunctionArgs) {
-  // Next authenticate the request
-  const authenticationResult = await authenticateApiRequest(request);
+export const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    allowJWT: true,
+    corsStrategy: "all",
+  },
+  async ({ params, authentication }) => {
+    const filename = params["*"];
 
-  if (!authenticationResult) {
-    return json({ error: "Invalid or Missing API key" }, { status: 401 });
-  }
+    const presignedUrl = await generatePresignedUrl(
+      authentication.environment.project.externalRef,
+      authentication.environment.slug,
+      filename,
+      "GET"
+    );
 
-  const parsedParams = ParamsSchema.parse(params);
-  const filename = parsedParams["*"];
-
-  const presignedUrl = await generatePresignedUrl(
-    authenticationResult.environment.project.externalRef,
-    authenticationResult.environment.slug,
-    filename,
-    "GET"
-  );
+    if (!presignedUrl) {
+      return json({ error: "Failed to generate presigned URL" }, { status: 500 });
+    }
 
-  if (!presignedUrl) {
-    return json({ error: "Failed to generate presigned URL" }, { status: 500 });
+    // Caller can now use this URL to fetch that object.
+    return json({ presignedUrl });
   }
-
-  // Caller can now use this URL to fetch that object.
-  return json({ presignedUrl });
-}
+);

apps/webapp/app/routes/api.v1.projects.$projectRef.runs.ts

@@ -5,7 +5,7 @@ import {
   ApiRunListPresenter,
   ApiRunListSearchParams,
 } from "~/presenters/v3/ApiRunListPresenter.server";
-import { createLoaderPATApiRoute } from "~/services/routeBuiilders/apiBuilder.server";
+import { createLoaderPATApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
 const ParamsSchema = z.object({
   projectRef: z.string(),

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

@@ -3,7 +3,7 @@ import {
   ApiRunListPresenter,
   ApiRunListSearchParams,
 } from "~/presenters/v3/ApiRunListPresenter.server";
-import { createLoaderApiRoute } from "~/services/routeBuiilders/apiBuilder.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
 export const loader = createLoaderApiRoute(
   {

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -1,15 +1,13 @@
-import { fromZodError } from "zod-validation-error";
-import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { TriggerTaskRequestBody } from "@trigger.dev/core/v3";
+import { generateJWT as internal_generateJWT, TriggerTaskRequestBody } from "@trigger.dev/core/v3";
+import { TaskRun } from "@trigger.dev/database";
 import { z } from "zod";
 import { env } from "~/env.server";
-import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { parseRequestJsonAsync } from "~/utils/parseRequestJson.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { OutOfEntitlementError, TriggerTaskService } from "~/v3/services/triggerTask.server";
-import { startActiveSpan } from "~/v3/tracer.server";
 
 const ParamsSchema = z.object({
   taskId: z.string(),
@@ -20,115 +18,125 @@ export const HeadersSchema = z.object({
   "trigger-version": z.string().nullish(),
   "x-trigger-span-parent-as-link": z.coerce.number().nullish(),
   "x-trigger-worker": z.string().nullish(),
+  "x-trigger-client": z.string().nullish(),
   traceparent: z.string().optional(),
   tracestate: z.string().optional(),
 });
 
-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" };
-  }
-
-  logger.debug("TriggerTask action", { headers: Object.fromEntries(request.headers) });
-
-  // Next authenticate the request
-  const authenticationResult = await authenticateApiRequest(request);
-
-  if (!authenticationResult) {
-    return json({ error: "Invalid or Missing API key" }, { status: 401 });
-  }
-
-  const contentLength = request.headers.get("content-length");
-
-  if (!contentLength || parseInt(contentLength) > env.TASK_PAYLOAD_MAXIMUM_SIZE) {
-    return json({ error: "Request body too large" }, { status: 413 });
-  }
+const { action, loader } = createActionApiRoute(
+  {
+    headers: HeadersSchema,
+    params: ParamsSchema,
+    body: TriggerTaskRequestBody,
+    allowJWT: true,
+    maxContentLength: env.TASK_PAYLOAD_MAXIMUM_SIZE,
+    authorization: {
+      action: "write",
+      resource: (params) => ({ tasks: params.taskId }),
+      superScopes: ["write:tasks", "admin"],
+    },
+    corsStrategy: "all",
+  },
+  async ({ body, headers, params, authentication }) => {
+    const {
+      "idempotency-key": idempotencyKey,
+      "trigger-version": triggerVersion,
+      "x-trigger-span-parent-as-link": spanParentAsLink,
+      traceparent,
+      tracestate,
+      "x-trigger-worker": isFromWorker,
+      "x-trigger-client": triggerClient,
+    } = headers;
+
+    const service = new TriggerTaskService();
+
+    try {
+      const traceContext =
+        traceparent && isFromWorker /// If the request is from a worker, we should pass the trace context
+          ? { traceparent, tracestate }
+          : undefined;
+
+      logger.debug("Triggering task", {
+        taskId: params.taskId,
+        idempotencyKey,
+        triggerVersion,
+        headers,
+        options: body.options,
+        isFromWorker,
+        traceContext,
+      });
+
+      const run = await service.call(params.taskId, authentication.environment, body, {
+        idempotencyKey: idempotencyKey ?? undefined,
+        triggerVersion: triggerVersion ?? undefined,
+        traceContext,
+        spanParentAsLink: spanParentAsLink === 1,
+      });
+
+      if (!run) {
+        return json({ error: "Task not found" }, { status: 404 });
+      }
 
-  const rawHeaders = Object.fromEntries(request.headers);
+      const $responseHeaders = await responseHeaders(
+        run,
+        authentication.environment,
+        triggerClient
+      );
 
-  const headers = HeadersSchema.safeParse(rawHeaders);
+      return json(
+        {
+          id: run.friendlyId,
+        },
+        {
+          headers: $responseHeaders,
+        }
+      );
+    } catch (error) {
+      if (error instanceof ServiceValidationError) {
+        return json({ error: error.message }, { status: 422 });
+      } else if (error instanceof OutOfEntitlementError) {
+        return json({ error: error.message }, { status: 422 });
+      } else if (error instanceof Error) {
+        return json({ error: error.message }, { status: 500 });
+      }
 
-  if (!headers.success) {
-    return json({ error: "Invalid headers" }, { status: 400 });
+      return json({ error: "Something went wrong" }, { status: 500 });
+    }
   }
-
-  const {
-    "idempotency-key": idempotencyKey,
-    "trigger-version": triggerVersion,
-    "x-trigger-span-parent-as-link": spanParentAsLink,
-    traceparent,
-    tracestate,
-    "x-trigger-worker": isFromWorker,
-  } = headers.data;
-
-  const { taskId } = ParamsSchema.parse(params);
-
-  // Now parse the request body
-  const anyBody = await parseRequestJsonAsync(request, { taskId });
-
-  const body = await startActiveSpan("TriggerTaskRequestBody.safeParse()", async (span) => {
-    return TriggerTaskRequestBody.safeParse(anyBody);
+);
+
+async function responseHeaders(
+  run: TaskRun,
+  environment: AuthenticatedEnvironment,
+  triggerClient?: string | null
+): Promise<Record<string, string>> {
+  const claimsHeader = JSON.stringify({
+    sub: environment.id,
+    pub: true,
   });
 
-  if (!body.success) {
-    return json(
-      { error: fromZodError(body.error, { prefix: "Invalid trigger call" }).toString() },
-      { status: 400 }
-    );
-  }
-
-  const service = new TriggerTaskService();
-
-  try {
-    const traceContext =
-      traceparent && isFromWorker /// If the request is from a worker, we should pass the trace context
-        ? { traceparent, tracestate }
-        : undefined;
-
-    logger.debug("Triggering task", {
-      taskId,
-      idempotencyKey,
-      triggerVersion,
-      headers: Object.fromEntries(request.headers),
-      options: body.data.options,
-      isFromWorker,
-      traceContext,
+  if (triggerClient === "browser") {
+    const claims = {
+      sub: environment.id,
+      pub: true,
+      scopes: [`read:runs:${run.friendlyId}`],
+    };
+
+    const jwt = await internal_generateJWT({
+      secretKey: environment.apiKey,
+      payload: claims,
+      expirationTime: "1h",
     });
 
-    const run = await service.call(taskId, authenticationResult.environment, body.data, {
-      idempotencyKey: idempotencyKey ?? undefined,
-      triggerVersion: triggerVersion ?? undefined,
-      traceContext,
-      spanParentAsLink: spanParentAsLink === 1,
-    });
-
-    if (!run) {
-      return json({ error: "Task not found" }, { status: 404 });
-    }
-
-    return json(
-      {
-        id: run.friendlyId,
-      },
-      {
-        headers: {
-          "x-trigger-jwt-claims": JSON.stringify({
-            sub: authenticationResult.environment.id,
-            pub: true,
-          }),
-        },
-      }
-    );
-  } catch (error) {
-    if (error instanceof ServiceValidationError) {
-      return json({ error: error.message }, { status: 422 });
-    } else if (error instanceof OutOfEntitlementError) {
-      return json({ error: error.message }, { status: 422 });
-    } else if (error instanceof Error) {
-      return json({ error: error.message }, { status: 400 });
-    }
-
-    return json({ error: "Something went wrong" }, { status: 500 });
+    return {
+      "x-trigger-jwt-claims": claimsHeader,
+      "x-trigger-jwt": jwt,
+    };
   }
+
+  return {
+    "x-trigger-jwt-claims": claimsHeader,
+  };
 }
+
+export { action, loader };

apps/webapp/app/routes/api.v3.runs.$runId.ts

@@ -1,7 +1,7 @@
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { ApiRetrieveRunPresenter } from "~/presenters/v3/ApiRetrieveRunPresenter.server";
-import { createLoaderApiRoute } from "~/services/routeBuiilders/apiBuilder.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),