From f88d4018cc1f714a687c56ed3e48eb68f7792512 Mon Sep 17 00:00:00 2001
From: Eric Allam <eric@trigger.dev>
Date: Mon, 18 May 2026 10:19:20 +0100
Subject: [PATCH] fix(webapp): dedupe realtimeStreams array push on stream
 create (#3653)

## Summary

The PUT handler at `/realtime/v1/streams/:runId/:target/:streamId` ran
`taskRun.update({ realtimeStreams: { push: streamId } })` on every call,
even when the `streamId` was already present. SDK call patterns that
re-initialize the same stream key on every chunk produce a per-write row
UPDATE, duplicate entries pile up in the array, and the row-lock + TOAST
rewrite cost grows unbounded on long-running stream sessions.

## Fix

Mirror the sibling append handler: read the array first and only push
when the `streamId` isn't already present. Identical behavior for
first-time stream creation; repeat creates short-circuit to a single
indexed read. The dashboard's per-run stream listing keeps working
because the first create still records the entry.

## Test plan

- [ ] A fresh PUT for a new `(run, streamId)` adds the entry to the
array
- [ ] A repeat PUT for the same pair leaves the array unchanged
- [ ] 404 is returned when the run doesn't exist; 400 when the run is
completed
---
 .server-changes/realtimestreams-dedupe.md     |  6 ++++
 ...ime.v1.streams.$runId.$target.$streamId.ts | 28 +++++++++++++------
 2 files changed, 25 insertions(+), 9 deletions(-)
 create mode 100644 .server-changes/realtimestreams-dedupe.md

diff --git a/.server-changes/realtimestreams-dedupe.md b/.server-changes/realtimestreams-dedupe.md
new file mode 100644
index 000000000..69987f7b4
--- /dev/null
+++ b/.server-changes/realtimestreams-dedupe.md
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Dedupe the `realtimeStreams` array push on `PUT /realtime/v1/streams/:runId/:target/:streamId` so repeat stream-init calls for the same `(run, streamId)` skip the row UPDATE, mirroring the existing append handler.
diff --git a/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts b/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts
index 9ca8e36f4..dd3d3bf31 100644
--- a/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts
+++ b/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts
@@ -62,31 +62,41 @@ const { action } = createActionApiRoute(
 
     if (request.method === "PUT") {
       // This is the "create" endpoint
-      const updatedRun = await prisma.taskRun.update({
+      const target = await prisma.taskRun.findFirst({
         where: {
           friendlyId: targetId,
           runtimeEnvironmentId: authentication.environment.id,
         },
-        data: {
-          realtimeStreams: {
-            push: params.streamId,
-          },
-        },
         select: {
+          id: true,
+          realtimeStreams: true,
           realtimeStreamsVersion: true,
           completedAt: true,
         },
       });
 
-      if (updatedRun.completedAt) {
+      if (!target) {
+        return new Response("Run not found", { status: 404 });
+      }
+
+      if (target.completedAt) {
         return new Response("Cannot initialize a realtime stream on a completed run", {
           status: 400,
         });
       }
 
+      if (!target.realtimeStreams.includes(params.streamId)) {
+        await prisma.taskRun.update({
+          where: { id: target.id },
+          data: {
+            realtimeStreams: { push: params.streamId },
+          },
+        });
+      }
+
       const realtimeStream = getRealtimeStreamInstance(
         authentication.environment,
-        updatedRun.realtimeStreamsVersion,
+        target.realtimeStreamsVersion,
         basinContext
       );
 
@@ -94,7 +104,7 @@ const { action } = createActionApiRoute(
 
       return json(
         {
-          version: updatedRun.realtimeStreamsVersion,
+          version: target.realtimeStreamsVersion,
         },
         { status: 202, headers: responseHeaders }
       );