Loads more stuff

Author: ericallam

apps/webapp/app/env.server.ts

@@ -1234,7 +1234,7 @@ const EnvironmentSchema = z
       .default("info"),
     REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS: z.coerce.number().int().default(100),
     REALTIME_STREAMS_S2_MAX_RETRIES: z.coerce.number().int().default(10),
-    REALTIME_STREAMS_S2_WAIT_SECONDS: z.coerce.number().int().default(600),
+    REALTIME_STREAMS_S2_WAIT_SECONDS: z.coerce.number().int().default(60),
   })
   .and(GithubAppEnvSchema)
   .and(S2EnvSchema);

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

@@ -551,6 +551,36 @@ export class SpanPresenter extends BasePresenter {
           },
         };
       }
+      case "realtime-stream": {
+        if (!span.entity.id) {
+          logger.error(`SpanPresenter: No realtime stream id`, {
+            spanId,
+            realtimeStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        const [runId, streamKey] = span.entity.id.split(":");
+
+        if (!runId || !streamKey) {
+          logger.error(`SpanPresenter: Invalid realtime stream id`, {
+            spanId,
+            realtimeStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        return {
+          ...data,
+          entity: {
+            type: "realtime-stream" as const,
+            object: {
+              runId,
+              streamKey,
+            },
+          },
+        };
+      }
       default:
         return { ...data, entity: null };
     }

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

@@ -43,17 +43,29 @@ export const loader = createLoaderApiRoute(
     // Get Last-Event-ID header for resuming from a specific position
     const lastEventId = request.headers.get("Last-Event-ID") || undefined;
 
+    const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined;
+    const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined;
+
+    if (timeoutInSeconds && isNaN(timeoutInSeconds)) {
+      return new Response("Invalid timeout seconds", { status: 400 });
+    }
+
+    if (timeoutInSeconds && timeoutInSeconds < 1) {
+      return new Response("Timeout seconds must be greater than 0", { status: 400 });
+    }
+
+    if (timeoutInSeconds && timeoutInSeconds > 600) {
+      return new Response("Timeout seconds must be less than 600", { status: 400 });
+    }
+
     const realtimeStream = getRealtimeStreamInstance(
       authentication.environment,
       run.realtimeStreamsVersion
     );
 
-    return realtimeStream.streamResponse(
-      request,
-      run.friendlyId,
-      params.streamId,
-      request.signal,
-      lastEventId
-    );
+    return realtimeStream.streamResponse(request, run.friendlyId, params.streamId, request.signal, {
+      lastEventId,
+      timeoutInSeconds,
+    });
   }
 );

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -80,6 +80,7 @@ import { createTimelineSpanEventsFromSpanEvents } from "~/utils/timelineSpanEven
 import { CompleteWaitpointForm } from "../resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route";
 import { requireUserId } from "~/services/session.server";
 import type { SpanOverride } from "~/v3/eventRepository/eventRepository.types";
+import { RealtimeStreamViewer } from "../resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route";
 
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
@@ -1146,6 +1147,24 @@ function SpanEntity({ span }: { span: Span }) {
         </div>
       );
     }
+    case "realtime-stream": {
+      return (
+        <div className="grid h-full grid-rows-[1fr_auto]">
+          <div className="flex flex-col gap-4 overflow-y-auto px-3 pt-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
+            <div>
+              <Header2>Realtime stream</Header2>
+              <Paragraph variant="small">
+                A realtime stream is a stream of data that is sent to the client.
+              </Paragraph>
+            </div>
+          </div>
+          <RealtimeStreamViewer
+            runId={span.entity.object.runId}
+            streamKey={span.entity.object.streamKey}
+          />
+        </div>
+      );
+    }
     default: {
       assertNever(span.entity);
     }