deploy v2 streaming WIP

Author: ericallam

apps/webapp/app/routes/api.v2.deployments.$deploymentId.finalize.ts

@@ -4,7 +4,7 @@ import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
-import { FinalizeDeploymentV2Service } from "~/v3/services/finalizeDeploymentV2";
+import { FinalizeDeploymentV2Service } from "~/v3/services/finalizeDeploymentV2.server";
 
 const ParamsSchema = z.object({
   deploymentId: z.string(),

apps/webapp/app/routes/api.v3.deployments.$deploymentId.finalize.ts

@@ -0,0 +1,96 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { FinalizeDeploymentRequestBody } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { FinalizeDeploymentV2Service } from "~/v3/services/finalizeDeploymentV2.server";
+
+const ParamsSchema = z.object({
+  deploymentId: z.string(),
+});
+
+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" };
+  }
+
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    return json({ error: "Invalid params" }, { status: 400 });
+  }
+
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    logger.info("Invalid or missing api key", { url: request.url });
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const authenticatedEnv = authenticationResult.environment;
+
+  const { deploymentId } = parsedParams.data;
+
+  const rawBody = await request.json();
+  const body = FinalizeDeploymentRequestBody.safeParse(rawBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid body", issues: body.error.issues }, { status: 400 });
+  }
+
+  try {
+    // Create a text stream chain
+    const stream = new TransformStream();
+    const encoder = new TextEncoderStream();
+    const writer = stream.writable.getWriter();
+
+    const service = new FinalizeDeploymentV2Service();
+
+    // Chain the streams: stream -> encoder -> response
+    const response = new Response(stream.readable.pipeThrough(encoder), {
+      headers: {
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache",
+        Connection: "keep-alive",
+      },
+    });
+
+    service
+      .call(authenticatedEnv, deploymentId, body.data, writer)
+      .then(async () => {
+        await writer.write(`event: complete\ndata: ${JSON.stringify({ id: deploymentId })}\n\n`);
+        await writer.close();
+      })
+      .catch(async (error) => {
+        let errorMessage;
+
+        if (error instanceof ServiceValidationError) {
+          errorMessage = { error: error.message };
+        } else if (error instanceof Error) {
+          logger.error("Error finalizing deployment", { error: error.message });
+          errorMessage = { error: `Internal server error: ${error.message}` };
+        } else {
+          logger.error("Error finalizing deployment", { error: String(error) });
+          errorMessage = { error: "Internal server error" };
+        }
+
+        await writer.write(`event: error\ndata: ${JSON.stringify(errorMessage)}\n\n`);
+        await writer.close();
+      });
+
+    return response;
+  } catch (error) {
+    if (error instanceof ServiceValidationError) {
+      return json({ error: error.message }, { status: 400 });
+    } else if (error instanceof Error) {
+      logger.error("Error finalizing deployment", { error: error.message });
+      return json({ error: `Internal server error: ${error.message}` }, { status: 500 });
+    } else {
+      logger.error("Error finalizing deployment", { error: String(error) });
+      return json({ error: "Internal server error" }, { status: 500 });
+    }
+  }
+}

apps/webapp/app/v3/services/finalizeDeploymentV2.ts renamed to apps/webapp/app/v3/services/finalizeDeploymentV2.server.ts

@@ -13,7 +13,8 @@ export class FinalizeDeploymentV2Service extends BaseService {
   public async call(
     authenticatedEnv: AuthenticatedEnvironment,
     id: string,
-    body: FinalizeDeploymentRequestBody
+    body: FinalizeDeploymentRequestBody,
+    writer?: WritableStreamDefaultWriter
   ) {
     // if it's self hosted, lets just use the v1 finalize deployment service
     if (body.selfHosted) {
@@ -148,11 +149,10 @@ type ExecutePushResult =
       logs: string;
     };
 
-async function executePushToRegistry({
-  depot,
-  registry,
-  deployment,
-}: ExecutePushToRegistryOptions): Promise<ExecutePushResult> {
+async function executePushToRegistry(
+  { depot, registry, deployment }: ExecutePushToRegistryOptions,
+  writer?: WritableStreamDefaultWriter
+): Promise<ExecutePushResult> {
   // Step 1: We need to "login" to the digital ocean registry
   const configDir = await ensureLoggedIntoDockerRegistry(registry.host, {
     username: registry.username,
@@ -180,7 +180,7 @@ async function executePushToRegistry({
   try {
     const processCode = await new Promise<number | null>((res, rej) => {
       // For some reason everything is output on stderr, not stdout
-      childProcess.stderr?.on("data", (data: Buffer) => {
+      childProcess.stderr?.on("data", async (data: Buffer) => {
         const text = data.toString();
 
         // Emitted data chunks can contain multiple lines. Remove empty lines.
@@ -191,6 +191,13 @@ async function executePushToRegistry({
           imageTag,
           deployment,
         });
+
+        // Now we can write strings directly
+        if (writer) {
+          for (const line of lines) {
+            await writer.write(`event: log\ndata: ${JSON.stringify({ message: line })}\n\n`);
+          }
+        }
       });
 
       childProcess.on("error", (e) => rej(e));