Override local env vars with env vars in the server (#914)

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

@@ -99,7 +99,9 @@ export class SpanPresenter {
       event: {
         ...event,
         events: transformEvents(events, event.metadata as Attributes),
-        output: isEmptyJson(event.output) ? null : JSON.stringify(event.output, null, 2),
+        output: isEmptyJson(event.output)
+          ? null
+          : JSON.stringify(unflattenAttributes(event.output as Attributes), null, 2),
         payload: payload ? JSON.stringify(payload, null, 2) : undefined,
         properties: sanitizedAttributesStringified(event.properties),
         style,

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

@@ -0,0 +1,57 @@
+import { ActionFunctionArgs, LoaderFunctionArgs, json } from "@remix-run/server-runtime";
+import { CreateBackgroundWorkerRequestBody } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { EnvironmentVariablesRepository } from "~/v3/environmentVariables/environmentVariablesRepository.server";
+import { CreateBackgroundWorkerService } from "~/v3/services/createBackgroundWorker.server";
+
+const ParamsSchema = z.object({
+  projectRef: z.string(),
+});
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  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) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const authenticatedEnv = authenticationResult.environment;
+
+  const { projectRef } = parsedParams.data;
+
+  const project = await prisma.project.findUnique({
+    where: {
+      externalRef: projectRef,
+      environments: {
+        some: {
+          id: authenticatedEnv.id,
+        },
+      },
+    },
+  });
+
+  if (!project) {
+    return json({ error: "Project not found" }, { status: 404 });
+  }
+
+  const repository = new EnvironmentVariablesRepository();
+
+  const variables = await repository.getEnvironmentVariables(project.id, authenticatedEnv.id);
+
+  return json({
+    variables: variables.reduce((acc: Record<string, string>, variable) => {
+      acc[variable.key] = variable.value;
+      return acc;
+    }, {}),
+  });
+}

apps/webapp/app/v3/authenticatedSocketConnection.server.ts

@@ -8,15 +8,15 @@ import { Evt } from "evt";
 import { randomUUID } from "node:crypto";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { EnvironmentQueueConsumer } from "./marqs/environmentQueueConsumer.server";
+import { DevQueueConsumer } from "./marqs/devQueueConsumer.server";
 import type { WebSocket, MessageEvent, CloseEvent, ErrorEvent } from "ws";
 
 export class AuthenticatedSocketConnection {
   public id: string;
   public onClose: Evt<CloseEvent> = new Evt();
 
   private _sender: ZodMessageSender<typeof serverWebsocketMessages>;
-  private _environmentConsumer: EnvironmentQueueConsumer;
+  private _environmentConsumer: DevQueueConsumer;
   private _messageHandler: ZodMessageHandler<typeof clientWebsocketMessages>;
 
   constructor(public ws: WebSocket, public authenticatedEnv: AuthenticatedEnvironment) {
@@ -38,7 +38,7 @@ export class AuthenticatedSocketConnection {
       },
     });
 
-    this._environmentConsumer = new EnvironmentQueueConsumer(authenticatedEnv, this._sender);
+    this._environmentConsumer = new DevQueueConsumer(authenticatedEnv, this._sender);
 
     ws.addEventListener("message", this.#handleMessage.bind(this));
     ws.addEventListener("close", this.#handleClose.bind(this));

apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts

@@ -317,10 +317,7 @@ export class EnvironmentVariablesRepository implements Repository {
       prismaClient: this.prismaClient,
     });
 
-    const secrets = await secretStore.getSecrets(
-      SecretValue,
-      secretKeyProjectPrefix(projectId)
-    );
+    const secrets = await secretStore.getSecrets(SecretValue, secretKeyProjectPrefix(projectId));
 
     const values = secrets.map((secret) => {
       const { projectId, environmentId, key } = parseSecretKey(secret.key);
@@ -402,6 +399,13 @@ export class EnvironmentVariablesRepository implements Repository {
       return [];
     }
 
+    return this.getEnvironmentVariables(projectId, environmentId);
+  }
+
+  async getEnvironmentVariables(
+    projectId: string,
+    environmentId: string
+  ): Promise<EnvironmentVariable[]> {
     const secretStore = getSecretStore("DATABASE", {
       prismaClient: this.prismaClient,
     });

apps/webapp/app/v3/environmentVariables/repository.ts

@@ -68,5 +68,6 @@ export interface Repository {
     userId: string,
     environmentId: string
   ): Promise<EnvironmentVariable[]>;
+  getEnvironmentVariables(projectId: string, environmentId: string): Promise<EnvironmentVariable[]>;
   delete(projectId: string, userId: string, options: DeleteEnvironmentVariable): Promise<Result>;
 }

apps/webapp/app/v3/marqs/environmentQueueConsumer.server.ts → apps/webapp/app/v3/marqs/devQueueConsumer.server.ts

@@ -3,6 +3,7 @@ import {
   RetryOptions,
   TaskRunContext,
   TaskRunExecution,
+  TaskRunExecutionPayload,
   TaskRunExecutionResult,
   ZodMessageSender,
   defaultRetryOptions,
@@ -18,6 +19,7 @@ import { generateFriendlyId } from "../friendlyIdentifiers";
 import { marqs } from "../marqs.server";
 import { attributesFromAuthenticatedEnv } from "../tracer.server";
 import { eventRepository } from "../eventRepository.server";
+import { EnvironmentVariablesRepository } from "../environmentVariables/environmentVariablesRepository.server";
 
 const tracer = trace.getTracer("environmentQueueConsumer");
 
@@ -35,7 +37,7 @@ export type EnvironmentQueueConsumerOptions = {
   traceTimeoutSeconds?: number;
 };
 
-export class EnvironmentQueueConsumer {
+export class DevQueueConsumer {
   private _backgroundWorkers: Map<string, BackgroundWorkerWithTasks> = new Map();
   private _enabled = false;
   private _options: Required<EnvironmentQueueConsumerOptions>;
@@ -430,9 +432,19 @@ export class EnvironmentQueueConsumer {
       },
     };
 
-    const payload = {
+    const environmentRepository = new EnvironmentVariablesRepository();
+    const variables = await environmentRepository.getEnvironmentVariables(
+      this.env.project.id,
+      this.env.id
+    );
+
+    const payload: TaskRunExecutionPayload = {
       execution,
       traceContext: lockedTaskRun.traceContext as Record<string, unknown>,
+      environment: variables.reduce((acc: Record<string, string>, curr) => {
+        acc[curr.key] = curr.value;
+        return acc;
+      }, {}),
     };
 
     try {

packages/cli-v3/src/apiClient.ts

@@ -6,6 +6,7 @@ import {
   CreateBackgroundWorkerRequestBody,
   WhoAmIResponseSchema,
   CreateBackgroundWorkerResponse,
+  GetEnvironmentVariablesResponseBody,
 } from "@trigger.dev/core/v3";
 
 export class ApiClient {
@@ -77,6 +78,23 @@ export class ApiClient {
       },
     });
   }
+
+  async getEnvironmentVariables(projectRef: string) {
+    if (!this.accessToken) {
+      throw new Error("getEnvironmentVariables: No access token");
+    }
+
+    return zodfetch(
+      GetEnvironmentVariablesResponseBody,
+      `${this.apiURL}/api/v1/projects/${projectRef}/envvars`,
+      {
+        headers: {
+          Authorization: `Bearer ${this.accessToken}`,
+          "Content-Type": "application/json",
+        },
+      }
+    );
+  }
 }
 
 type ApiResult<TSuccessResult> =

packages/cli-v3/src/commands/dev.tsx

@@ -391,76 +391,86 @@ function useDev({ config, apiUrl, apiKey, environmentClient, projectName, debugg
                   await fs.promises.writeFile(sourceMapPath, sourceMapFile.text);
                 }
 
+                const environmentVariablesResponse =
+                  await environmentClient.getEnvironmentVariables(config.project);
+
                 const backgroundWorker = new BackgroundWorker(fullPath, {
                   projectDir: config.projectDir,
                   env: {
                     TRIGGER_API_URL: apiUrl,
                     TRIGGER_API_KEY: apiKey,
+                    ...(environmentVariablesResponse.success
+                      ? environmentVariablesResponse.data.variables
+                      : {}),
                   },
                   debuggerOn,
                 });
 
-                await backgroundWorker.initialize();
-
-                latestWorkerContentHash = contentHash;
+                try {
+                  await backgroundWorker.initialize();
 
-                let packageVersion: string | undefined;
+                  latestWorkerContentHash = contentHash;
 
-                const taskResources: Array<TaskResource> = [];
+                  let packageVersion: string | undefined;
 
-                if (!backgroundWorker.tasks) {
-                  throw new Error(`Background Worker started without tasks`);
-                }
+                  const taskResources: Array<TaskResource> = [];
 
-                for (const task of backgroundWorker.tasks) {
-                  taskResources.push(task);
+                  if (!backgroundWorker.tasks) {
+                    throw new Error(`Background Worker started without tasks`);
+                  }
 
-                  packageVersion = task.packageVersion;
-                }
+                  for (const task of backgroundWorker.tasks) {
+                    taskResources.push(task);
 
-                if (!packageVersion) {
-                  throw new Error(`Background Worker started without package version`);
-                }
+                    packageVersion = task.packageVersion;
+                  }
 
-                const backgroundWorkerBody: CreateBackgroundWorkerRequestBody = {
-                  localOnly: true,
-                  metadata: {
-                    packageVersion,
-                    cliPackageVersion: packageJson.version,
-                    tasks: taskResources,
-                    contentHash: contentHash,
-                  },
-                };
+                  if (!packageVersion) {
+                    throw new Error(`Background Worker started without package version`);
+                  }
 
-                const backgroundWorkerRecord = await environmentClient.createBackgroundWorker(
-                  config.project,
-                  backgroundWorkerBody
-                );
+                  const backgroundWorkerBody: CreateBackgroundWorkerRequestBody = {
+                    localOnly: true,
+                    metadata: {
+                      packageVersion,
+                      cliPackageVersion: packageJson.version,
+                      tasks: taskResources,
+                      contentHash: contentHash,
+                    },
+                  };
 
-                if (!backgroundWorkerRecord.success) {
-                  throw new Error(backgroundWorkerRecord.error);
-                }
-
-                backgroundWorker.metadata = backgroundWorkerRecord.data;
-
-                if (firstBuild) {
-                  logger.log(
-                    chalk.green(
-                      `Background worker started (${backgroundWorkerRecord.data.version})`
-                    )
+                  const backgroundWorkerRecord = await environmentClient.createBackgroundWorker(
+                    config.project,
+                    backgroundWorkerBody
                   );
-                } else {
-                  logger.log(
-                    chalk.dim(`Background worker rebuilt (${backgroundWorkerRecord.data.version})`)
-                  );
-                }
 
-                firstBuild = false;
-
-                await backgroundWorkerCoordinator.registerWorker(
-                  backgroundWorkerRecord.data,
-                  backgroundWorker
-                );
+                  if (!backgroundWorkerRecord.success) {
+                    throw new Error(backgroundWorkerRecord.error);
+                  }
+
+                  backgroundWorker.metadata = backgroundWorkerRecord.data;
+
+                  if (firstBuild) {
+                    logger.log(
+                      chalk.green(
+                        `Background worker started (${backgroundWorkerRecord.data.version})`
+                      )
+                    );
+                  } else {
+                    logger.log(
+                      chalk.dim(
+                        `Background worker rebuilt (${backgroundWorkerRecord.data.version})`
+                      )
+                    );
+                  }
+
+                  firstBuild = false;
+
+                  await backgroundWorkerCoordinator.registerWorker(
+                    backgroundWorkerRecord.data,
+                    backgroundWorker
+                  );
+                } catch (e) {}
               });
             },
           },

packages/cli-v3/src/dev/backgroundWorker.ts

@@ -228,8 +228,8 @@ export class BackgroundWorker {
       const child = fork(this.path, {
         stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"],
         env: {
-          ...this.params.env,
           ...this.#readEnvVars(),
+          ...this.params.env,
         },
       });
 
@@ -295,8 +295,9 @@ export class BackgroundWorker {
       const taskRunProcess = new TaskRunProcess(
         this.path,
         {
-          ...this.params.env,
           ...this.#readEnvVars(),
+          ...this.params.env,
+          ...(payload.environment ?? {}),
         },
         this.metadata,
         this.params

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

@@ -79,3 +79,11 @@ export const GetBatchResponseBody = z.object({
 });
 
 export type GetBatchResponseBody = z.infer<typeof GetBatchResponseBody>;
+
+export const GetEnvironmentVariablesResponseBody = z.object({
+  variables: z.record(z.string()),
+});
+
+export type GetEnvironmentVariablesResponseBody = z.infer<
+  typeof GetEnvironmentVariablesResponseBody
+>;

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

@@ -1,9 +1,10 @@
 import { z } from "zod";
-import { TaskRunExecutionResult, TaskRunExecution, TaskRunFailedExecutionResult } from "./common";
+import { TaskRunExecution, TaskRunExecutionResult } from "./common";
 
 export const TaskRunExecutionPayload = z.object({
   execution: TaskRunExecution,
   traceContext: z.record(z.unknown()),
+  environment: z.record(z.string()).optional(),
 });
 
 export type TaskRunExecutionPayload = z.infer<typeof TaskRunExecutionPayload>;