refactor(webapp): make S2 account and basin hosts env-configurable with cloud defaults

Author: ericallam

apps/webapp/app/env.server.ts

@@ -2121,6 +2121,8 @@ const EnvironmentSchema = z
     REALTIME_STREAMS_S2_BASIN: z.string().optional(),
     REALTIME_STREAMS_S2_ACCESS_TOKEN: z.string().optional(),
     REALTIME_STREAMS_S2_ENDPOINT: z.string().optional(),
+    REALTIME_STREAMS_S2_ACCOUNT_URL: z.string().default("https://a.s2.dev/v1"),
+    REALTIME_STREAMS_S2_BASIN_URL: z.string().default("https://{basin}.b.s2.dev/v1"),
     REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS: z.enum(["true", "false"]).default("false"),
     REALTIME_STREAMS_S2_ACCESS_TOKEN_EXPIRATION_IN_MS: z.coerce
       .number()

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

@@ -44,6 +44,10 @@ export type S2RealtimeStreamsOptions = {
 
   // Custom endpoint for s2-lite (self-hosted)
   endpoint?: string; // e.g., "http://localhost:4566/v1"
+  /** Account-level API base for account/basin ops. Defaults to S2 cloud. */
+  accountUrl?: string;
+  /** Per-basin API base, with a `{basin}` placeholder. Defaults to S2 cloud. */
+  basinUrl?: string;
 
   // Skip access token issuance (s2-lite doesn't support /access-tokens)
   skipAccessTokens?: boolean;
@@ -116,8 +120,10 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
 
   constructor(opts: S2RealtimeStreamsOptions) {
     this.basin = opts.basin;
-    this.baseUrl = opts.endpoint ?? `https://${this.basin}.b.s2.dev/v1`;
-    this.accountUrl = opts.endpoint ?? `https://a.s2.dev/v1`;
+    this.baseUrl =
+      opts.endpoint ??
+      (opts.basinUrl ?? `https://{basin}.b.s2.dev/v1`).replace("{basin}", this.basin);
+    this.accountUrl = opts.endpoint ?? opts.accountUrl ?? `https://a.s2.dev/v1`;
     this.endpoint = opts.endpoint;
     this.token = opts.accessToken;
     this.streamPrefix = opts.streamPrefix ?? "";

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

@@ -185,7 +185,7 @@ type CreateBasinOptions = {
 };
 
 async function s2CreateBasin(name: string, opts: CreateBasinOptions): Promise<void> {
-  const url = `https://a.s2.dev/v1/basins`;
+  const url = `${env.REALTIME_STREAMS_S2_ACCOUNT_URL}/basins`;
   const body = {
     basin: name,
     config: {
@@ -222,7 +222,7 @@ type ReconfigureBasinOptions = {
 };
 
 async function s2ReconfigureBasin(name: string, opts: ReconfigureBasinOptions): Promise<void> {
-  const url = `https://a.s2.dev/v1/basins/${encodeURIComponent(name)}`;
+  const url = `${env.REALTIME_STREAMS_S2_ACCOUNT_URL}/basins/${encodeURIComponent(name)}`;
   const body = {
     default_stream_config: {
       retention_policy: { age: parseDuration(opts.retentionPolicy) },

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

@@ -70,6 +70,8 @@ export function getRealtimeStreamInstance(
       basin: resolvedBasin,
       accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN ?? "",
       endpoint: env.REALTIME_STREAMS_S2_ENDPOINT,
+      accountUrl: env.REALTIME_STREAMS_S2_ACCOUNT_URL,
+      basinUrl: env.REALTIME_STREAMS_S2_BASIN_URL,
       skipAccessTokens: env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true",
       streamPrefix: streamPrefixFor(environment, resolvedBasin),
       logLevel: env.REALTIME_STREAMS_S2_LOG_LEVEL,