Merge branch 'main' into sentry-user-attribution

Author: d-cs

.server-changes/chat-snapshot-default-protocol.md

@@ -0,0 +1,8 @@
+---
+area: webapp
+type: fix
+---
+
+Pin chat.agent session snapshots to a single object store so writes and reads
+always round-trip through the same provider when `OBJECT_STORE_DEFAULT_PROTOCOL`
+is set.

.server-changes/env-var-value-environment-id-index.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Speed up env-var lookups on the projects API by indexing `EnvironmentVariableValue.environmentId`.

apps/webapp/app/routes/api.v1.sessions.$sessionId.snapshot-url.ts

@@ -0,0 +1,69 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server";
+import { resolveSessionByIdOrExternalId } from "~/services/realtime/sessions.server";
+import {
+  createActionApiRoute,
+  createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
+import { generatePresignedUrl } from "~/v3/objectStore.server";
+
+const ParamsSchema = z.object({
+  sessionId: z.string(),
+});
+
+// `chatSnapshotStoragePath` is stamped on every new Session at row creation
+// (see api.v1.sessions.ts). The fallback handles sessions created before
+// the column existed — read against the currently-configured default
+// protocol and compute the same path the SDK uploaded under.
+function snapshotKey(session: { friendlyId: string; chatSnapshotStoragePath: string | null }) {
+  return session.chatSnapshotStoragePath ?? chatSnapshotStoragePathForSession(session.friendlyId);
+}
+
+const routeConfig = {
+  params: ParamsSchema,
+  allowJWT: true,
+  corsStrategy: "all" as const,
+  findResource: async (params: z.infer<typeof ParamsSchema>, auth: { environment: { id: string } }) =>
+    resolveSessionByIdOrExternalId($replica, auth.environment.id, params.sessionId),
+};
+
+export const { action } = createActionApiRoute(
+  { ...routeConfig, method: "PUT" },
+  async ({ authentication, resource: session }) => {
+    if (!session) {
+      return json({ error: "Session not found" }, { status: 404 });
+    }
+
+    const signed = await generatePresignedUrl(
+      authentication.environment.project.externalRef,
+      authentication.environment.slug,
+      snapshotKey(session),
+      "PUT"
+    );
+    if (!signed.success) {
+      return json({ error: `Failed to generate presigned URL: ${signed.error}` }, { status: 500 });
+    }
+
+    return json({ presignedUrl: signed.url });
+  }
+);
+
+export const loader = createLoaderApiRoute(routeConfig, async ({ authentication, resource: session }) => {
+  if (!session) {
+    return json({ error: "Session not found" }, { status: 404 });
+  }
+
+  const signed = await generatePresignedUrl(
+    authentication.environment.project.externalRef,
+    authentication.environment.slug,
+    snapshotKey(session),
+    "GET"
+  );
+  if (!signed.success) {
+    return json({ error: `Failed to generate presigned URL: ${signed.error}` }, { status: 500 });
+  }
+
+  return json({ presignedUrl: signed.url });
+});

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

@@ -17,6 +17,7 @@ import {
   ensureRunForSession,
   type SessionTriggerConfig,
 } from "~/services/realtime/sessionRunManager.server";
+import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server";
 import { serializeSession } from "~/services/realtime/sessions.server";
 import { SessionsRepository } from "~/services/sessionsRepository/sessionsRepository.server";
 import {
@@ -181,6 +182,7 @@ const { action } = createActionApiRoute(
             environmentType: authentication.environment.type,
             organizationId: authentication.environment.organizationId,
             streamBasinName: authentication.environment.organization.streamBasinName,
+            chatSnapshotStoragePath: chatSnapshotStoragePathForSession(friendlyId),
           },
           update: { triggerConfig: triggerConfigJson },
         });
@@ -201,6 +203,7 @@ const { action } = createActionApiRoute(
             environmentType: authentication.environment.type,
             organizationId: authentication.environment.organizationId,
             streamBasinName: authentication.environment.organization.streamBasinName,
+            chatSnapshotStoragePath: chatSnapshotStoragePathForSession(friendlyId),
           },
         });
       }

apps/webapp/app/services/apiRateLimit.server.ts

@@ -70,6 +70,7 @@ export const apiRateLimiter = authorizationRateLimitMiddleware({
     // customer-facing surface so customer rate limits shouldn't apply.
     /^\/api\/v1\/packets\//,
     /^\/api\/v2\/packets\//,
+    /^\/api\/v1\/sessions\/[^\/]+\/snapshot-url$/,
   ],
   log: {
     rejections: env.API_RATE_LIMIT_REJECTION_LOGS_ENABLED === "1",

apps/webapp/app/services/realtime/chatSnapshot.server.ts

@@ -0,0 +1,13 @@
+import { env } from "~/env.server";
+
+/**
+ * Canonical storage URI for a session's chat.agent snapshot. Stamped on
+ * `Session.chatSnapshotStoragePath` at row creation so PUT/GET presigns
+ * resolve to the same store even if `OBJECT_STORE_DEFAULT_PROTOCOL`
+ * changes later.
+ */
+export function chatSnapshotStoragePathForSession(friendlyId: string): string {
+  const path = `sessions/${friendlyId}/snapshot.json`;
+  const protocol = env.OBJECT_STORE_DEFAULT_PROTOCOL;
+  return protocol ? `${protocol}://${path}` : path;
+}

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

@@ -2,7 +2,7 @@ import { Prisma, type PrismaClient, type RuntimeEnvironmentType } from "@trigger
 import type { AuthenticatedEnvironment } from "@trigger.dev/core/v3/auth/environment";
 import { z } from "zod";
 import { environmentFullTitle } from "~/components/environments/EnvironmentLabel";
-import { $transaction, prisma } from "~/db.server";
+import { $replica, $transaction, prisma, type PrismaReplicaClient } from "~/db.server";
 import { env } from "~/env.server";
 import { getSecretStore } from "~/services/secrets/secretStore.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
@@ -47,7 +47,10 @@ function parseSecretKey(key: string) {
 const SecretValue = z.object({ secret: z.string() });
 
 export class EnvironmentVariablesRepository implements Repository {
-  constructor(private prismaClient: PrismaClient = prisma) {}
+  constructor(
+    private prismaClient: PrismaClient = prisma,
+    private replicaClient: PrismaReplicaClient = $replica
+  ) {}
 
   async create(projectId: string, options: CreateEnvironmentVariables): Promise<CreateResult> {
     const project = await this.prismaClient.project.findFirst({
@@ -582,7 +585,7 @@ export class EnvironmentVariablesRepository implements Repository {
     const variables = await this.getEnvironment(projectId, environmentId, parentEnvironmentId);
 
     // Get the keys of all secret variables
-    const secretValues = await this.prismaClient.environmentVariableValue.findMany({
+    const secretValues = await this.replicaClient.environmentVariableValue.findMany({
       where: {
         environmentId: parentEnvironmentId
           ? { in: [environmentId, parentEnvironmentId] }

apps/webapp/test/chat-snapshot-integration.test.ts

@@ -26,6 +26,7 @@ import {
 import type { UIMessage } from "ai";
 import { afterEach, describe, expect, vi } from "vitest";
 import { env } from "~/env.server";
+import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server";
 import { generatePresignedUrl } from "~/v3/objectStore.server";
 
 vi.setConfig({ testTimeout: 60_000 });
@@ -54,22 +55,21 @@ function makeSnapshot(opts: { messages?: UIMessage[]; lastOutEventId?: string }
 
 /**
  * Stub `apiClientManager.clientOrThrow()` so the SDK helpers see a fake
- * api client whose `getPayloadUrl` / `createUploadPayloadUrl` return
- * presigned URLs minted by the webapp's real `generatePresignedUrl`
- * (which signs against MinIO).
- *
- * The SDK helpers internally do `fetch(presignedUrl, ...)` to read/write
- * the blob, so MinIO ends up holding the actual bytes.
+ * api client. Mirrors the snapshot-url route: derive the canonical
+ * `sessions/{id}/snapshot.json` key (with optional default-protocol
+ * prefix) and sign it via `generatePresignedUrl` against MinIO.
  */
 function stubApiClient(opts: { projectRef: string; envSlug: string }) {
   vi.spyOn(apiClientManager, "clientOrThrow").mockReturnValue({
-    async getPayloadUrl(filename: string) {
-      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, filename, "GET");
+    async getChatSnapshotUrl(sessionId: string) {
+      const key = chatSnapshotStoragePathForSession(sessionId);
+      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, key, "GET");
       if (!result.success) throw new Error(result.error);
       return { presignedUrl: result.url };
     },
-    async createUploadPayloadUrl(filename: string) {
-      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, filename, "PUT");
+    async createChatSnapshotUploadUrl(sessionId: string) {
+      const key = chatSnapshotStoragePathForSession(sessionId);
+      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, key, "PUT");
       if (!result.success) throw new Error(result.error);
       return { presignedUrl: result.url };
     },

apps/webapp/test/replay-after-crash.test.ts

@@ -33,6 +33,7 @@ import {
 import type { UIMessageChunk } from "ai";
 import { afterEach, describe, expect, vi } from "vitest";
 import { env } from "~/env.server";
+import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server";
 import { generatePresignedUrl } from "~/v3/objectStore.server";
 
 vi.setConfig({ testTimeout: 60_000 });
@@ -77,13 +78,15 @@ function stubApiClient(opts: {
     })
   );
   vi.spyOn(apiClientManager, "clientOrThrow").mockReturnValue({
-    async getPayloadUrl(filename: string) {
-      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, filename, "GET");
+    async getChatSnapshotUrl(sessionId: string) {
+      const key = chatSnapshotStoragePathForSession(sessionId);
+      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, key, "GET");
       if (!result.success) throw new Error(result.error);
       return { presignedUrl: result.url };
     },
-    async createUploadPayloadUrl(filename: string) {
-      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, filename, "PUT");
+    async createChatSnapshotUploadUrl(sessionId: string) {
+      const key = chatSnapshotStoragePathForSession(sessionId);
+      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, key, "PUT");
       if (!result.success) throw new Error(result.error);
       return { presignedUrl: result.url };
     },

internal-packages/database/prisma/migrations/20260520120000_add_environment_variable_value_environment_id_idx/migration.sql

@@ -0,0 +1,3 @@
+-- CreateIndex
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "EnvironmentVariableValue_environmentId_idx"
+  ON "EnvironmentVariableValue"("environmentId");