Turn on the relay realtime stream service

Author: ericallam

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

@@ -1,6 +1,7 @@
 import { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
+import { relayRealtimeStreams } from "~/services/realtime/relayRealtimeStreams.server";
 import { v1RealtimeStreams } from "~/services/realtime/v1StreamsGlobal.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
@@ -16,7 +17,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return new Response("No body provided", { status: 400 });
   }
 
-  return v1RealtimeStreams.ingestData(request.body, $params.runId, $params.streamId);
+  return relayRealtimeStreams.ingestData(request.body, $params.runId, $params.streamId);
 }
 
 export const loader = createLoaderApiRoute(
@@ -51,7 +52,7 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ params, request, resource: run, authentication }) => {
-    return v1RealtimeStreams.streamResponse(
+    return relayRealtimeStreams.streamResponse(
       request,
       run.friendlyId,
       params.streamId,

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

@@ -27,7 +27,7 @@ export class RelayRealtimeStreams implements StreamIngestor, StreamResponder {
   private waitForBufferInterval: number;
 
   constructor(private options: RelayRealtimeStreamsOptions) {
-    this.waitForBufferTimeout = options.waitForBufferTimeout ?? 5000;
+    this.waitForBufferTimeout = options.waitForBufferTimeout ?? 1200;
     this.waitForBufferInterval = options.waitForBufferInterval ?? 50;
 
     // Periodic cleanup
@@ -106,6 +106,7 @@ export class RelayRealtimeStreams implements StreamIngestor, StreamResponder {
         "Content-Type": "text/event-stream",
         "Cache-Control": "no-cache",
         Connection: "keep-alive",
+        "x-relay-realtime-streams": "true",
       },
     });
   }