refactor(webapp): route tag and realtime-stream appends through RunStore

Author: d-cs

apps/webapp/app/routes/api.v1.runs.$runId.tags.ts

@@ -9,6 +9,7 @@ import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { logger } from "~/services/logger.server";
 import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server";
 import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
+import { runStore } from "~/v3/runStore.server";
 
 // Pull the existing tags out of a buffer entry's serialised payload so
 // the buffer-path response can dedup against them, matching the
@@ -84,14 +85,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
         if (newTags.length === 0) {
           return json({ message: "No new tags to add" }, { status: 200 });
         }
-        const updated = await prisma.taskRun.update({
-          where: {
-            id: taskRun.id,
-            runtimeEnvironmentId: env.id,
-          },
-          data: { runTags: { push: newTags } },
-          select: { updatedAt: true },
-        });
+        const updated = await runStore.pushTags(taskRun.id, newTags, { runtimeEnvironmentId: env.id }, prisma);
         // Publish a run-changed record with the NEW tag set so tag feeds reindex
         // (no-op unless enabled). updatedAt is the read-your-writes watermark.
         publishChangeRecord({

apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.append.ts

@@ -6,6 +6,7 @@ import { $replica, prisma } from "~/db.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { ServiceValidationError } from "~/v3/services/common.server";
+import { runStore } from "~/v3/runStore.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -87,16 +88,7 @@ const { action } = createActionApiRoute(
     }
 
     if (!targetRun.realtimeStreams.includes(params.streamId)) {
-      await prisma.taskRun.update({
-        where: {
-          id: targetRun.id,
-        },
-        data: {
-          realtimeStreams: {
-            push: params.streamId,
-          },
-        },
-      });
+      await runStore.pushRealtimeStream(targetRun.id, params.streamId, prisma);
     }
 
     const part = await request.text();

apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts

@@ -6,6 +6,7 @@ import {
   createActionApiRoute,
   createLoaderApiRoute,
 } from "~/services/routeBuilders/apiBuilder.server";
+import { runStore } from "~/v3/runStore.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -86,12 +87,7 @@ const { action } = createActionApiRoute(
       }
 
       if (!target.realtimeStreams.includes(params.streamId)) {
-        await prisma.taskRun.update({
-          where: { id: target.id },
-          data: {
-            realtimeStreams: { push: params.streamId },
-          },
-        });
+        await runStore.pushRealtimeStream(target.id, params.streamId, prisma);
       }
 
       const realtimeStream = getRealtimeStreamInstance(