feat(streams): make v2 streams the default when using 4.1.0+ if they are supported (#2677)

Author: ericallam

apps/webapp/app/env.server.ts

@@ -1236,6 +1236,7 @@ const EnvironmentSchema = z
     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(60),
+    REALTIME_STREAMS_DEFAULT_VERSION: z.enum(["v1", "v2"]).default("v1"),
     WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),
   })
   .and(GithubAppEnvSchema)

apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts

@@ -9,6 +9,7 @@ import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { BatchTriggerV3Service } from "~/v3/services/batchTriggerV3.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 
 const ParamsSchema = z.object({
   taskId: z.string(),
@@ -40,6 +41,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
     "trigger-version": triggerVersion,
     "x-trigger-span-parent-as-link": spanParentAsLink,
     "x-trigger-worker": isFromWorker,
+    "x-trigger-realtime-streams-version": realtimeStreamsVersion,
     traceparent,
     tracestate,
   } = headers.data;
@@ -100,6 +102,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
       triggerVersion: triggerVersion ?? undefined,
       traceContext,
       spanParentAsLink: spanParentAsLink === 1,
+      realtimeStreamsVersion: determineRealtimeStreamsVersion(realtimeStreamsVersion ?? undefined),
     });
 
     if (!result) {

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -11,6 +11,7 @@ import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { ApiAuthenticationResultSuccess, getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import {
@@ -126,7 +127,9 @@ const { action, loader } = createActionApiRoute(
           traceContext,
           spanParentAsLink: spanParentAsLink === 1,
           oneTimeUseToken,
-          realtimeStreamsVersion: realtimeStreamsVersion ?? undefined,
+          realtimeStreamsVersion: determineRealtimeStreamsVersion(
+            realtimeStreamsVersion ?? undefined
+          ),
         },
         engineVersion ?? undefined
       );

apps/webapp/app/routes/api.v1.tasks.batch.ts

@@ -16,6 +16,7 @@ import {
 } from "~/v3/services/batchTriggerV3.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 
 const { action, loader } = createActionApiRoute(
   {
@@ -69,6 +70,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-client": triggerClient,
       "x-trigger-engine-version": engineVersion,
       "batch-processing-strategy": batchProcessingStrategy,
+      "x-trigger-realtime-streams-version": realtimeStreamsVersion,
       traceparent,
       tracestate,
     } = headers;
@@ -107,6 +109,9 @@ const { action, loader } = createActionApiRoute(
         traceContext,
         spanParentAsLink: spanParentAsLink === 1,
         oneTimeUseToken,
+        realtimeStreamsVersion: determineRealtimeStreamsVersion(
+          realtimeStreamsVersion ?? undefined
+        ),
       });
 
       const $responseHeaders = await responseHeaders(

apps/webapp/app/routes/api.v2.tasks.batch.ts

@@ -18,6 +18,7 @@ import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { BatchProcessingStrategy } from "~/v3/services/batchTriggerV3.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 
 const { action, loader } = createActionApiRoute(
   {
@@ -59,6 +60,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-engine-version": engineVersion,
       "batch-processing-strategy": batchProcessingStrategy,
       "x-trigger-request-idempotency-key": requestIdempotencyKey,
+      "x-trigger-realtime-streams-version": realtimeStreamsVersion,
       traceparent,
       tracestate,
     } = headers;
@@ -119,6 +121,9 @@ const { action, loader } = createActionApiRoute(
         traceContext,
         spanParentAsLink: spanParentAsLink === 1,
         oneTimeUseToken,
+        realtimeStreamsVersion: determineRealtimeStreamsVersion(
+          realtimeStreamsVersion ?? undefined
+        ),
       });
 
       const $responseHeaders = await responseHeaders(

apps/webapp/app/runEngine/services/batchTrigger.server.ts

@@ -47,6 +47,7 @@ export type BatchTriggerTaskServiceOptions = {
   traceContext?: Record<string, string | undefined | Record<string, string | undefined>>;
   spanParentAsLink?: boolean;
   oneTimeUseToken?: string;
+  realtimeStreamsVersion?: "v1" | "v2";
 };
 
 /**
@@ -708,6 +709,7 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
         batchIndex: currentIndex,
         skipChecks: true, // Skip entitlement and queue checks since we already validated at batch/chunk level
         planType, // Pass planType from batch-level entitlement check
+        realtimeStreamsVersion: options?.realtimeStreamsVersion,
       },
       "V2"
     );

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

@@ -60,6 +60,22 @@ export function getRealtimeStreamInstance(
   }
 }
 
+export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" {
+  if (!streamVersion) {
+    return env.REALTIME_STREAMS_DEFAULT_VERSION;
+  }
+
+  if (
+    streamVersion === "v2" &&
+    env.REALTIME_STREAMS_S2_BASIN &&
+    env.REALTIME_STREAMS_S2_ACCESS_TOKEN
+  ) {
+    return "v2";
+  }
+
+  return "v1";
+}
+
 const s2RealtimeStreamsCache = singleton(
   "s2RealtimeStreamsCache",
   initializeS2RealtimeStreamsCache

apps/webapp/app/v3/services/batchTriggerV3.server.ts

@@ -56,6 +56,7 @@ export type BatchTriggerTaskServiceOptions = {
   traceContext?: Record<string, string | undefined>;
   spanParentAsLink?: boolean;
   oneTimeUseToken?: string;
+  realtimeStreamsVersion?: "v1" | "v2";
 };
 
 type RunItemData = {
@@ -851,6 +852,7 @@ export class BatchTriggerV3Service extends BaseService {
         batchId: batch.friendlyId,
         skipChecks: true,
         runFriendlyId: task.runId,
+        realtimeStreamsVersion: options?.realtimeStreamsVersion,
       }
     );
 

apps/webapp/app/v3/services/replayTaskRun.server.ts

@@ -11,6 +11,7 @@ import { BaseService } from "./baseService.server";
 import { OutOfEntitlementError, TriggerTaskService } from "./triggerTask.server";
 import { type RunOptionsData } from "../testTask";
 import { replaceSuperJsonPayload } from "@trigger.dev/core/v3/utils/ioSerialization";
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 
 type OverrideOptions = {
   environmentId?: string;
@@ -118,7 +119,9 @@ export class ReplayTaskRunService extends BaseService {
           traceContext: {
             traceparent: `00-${existingTaskRun.traceId}-${existingTaskRun.spanId}-01`,
           },
-          realtimeStreamsVersion: existingTaskRun.realtimeStreamsVersion,
+          realtimeStreamsVersion: determineRealtimeStreamsVersion(
+            existingTaskRun.realtimeStreamsVersion
+          ),
         }
       );
 

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -33,7 +33,7 @@ export type TriggerTaskServiceOptions = {
   overrideCreatedAt?: Date;
   replayedFromTaskRunFriendlyId?: string;
   planType?: string;
-  realtimeStreamsVersion?: string;
+  realtimeStreamsVersion?: "v1" | "v2";
 };
 
 export class OutOfEntitlementError extends Error {