feat(webapp): let the deployment S2 client endpoints be overridden

apps/webapp/app/env.server.ts

@@ -4,6 +4,7 @@ import { BoolEnv } from "./utils/boolEnv";
 import { isValidDatabaseUrl } from "./utils/db";
 import { parseRunOpsShards, validateShardListAgainstNewUrl } from "~/v3/runOpsShards.server";
 import { isValidRegex } from "./utils/regex";
+import { isValidS2Endpoint } from "./utils/s2Endpoint";
 import { isValidDuration } from "./services/realtime/duration.server";
 
 // `z.string()` constrained to a `parseDuration`-parseable string (e.g.
@@ -79,6 +80,19 @@ const S2EnvSchema = z.preprocess(
       S2_ACCESS_TOKEN: z.string(),
       S2_DEPLOYMENT_LOGS_BASIN_NAME: z.string(),
       S2_DEPLOYMENT_STREAMS_LOCAL: z.string().default("0"),
+      // Points deployment event logs at an S2 service other than the hosted one, e.g. the
+      // local s2-lite in docker compose. One value covers both the account and basin
+      // endpoints: splitting them lets a half-set config send the access token to the
+      // hosted service while the operator believes they are entirely local.
+      S2_DEPLOYMENT_ENDPOINT: z
+        .string()
+        .trim()
+        .transform((value) => (value === "" ? undefined : value))
+        .refine(
+          (value) => value === undefined || isValidS2Endpoint(value),
+          "must be an http(s) URL; http is only allowed for a loopback host, because the S2 access token is sent to it as a bearer token"
+        )
+        .optional(),
     }),
     z.object({
       S2_ENABLED: z.literal("0"),

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

@@ -13,11 +13,12 @@ import { type User } from "~/models/user.server";
 import { getUsername } from "~/utils/username";
 import { processGitMetadata } from "./BranchesPresenter.server";
 import { VercelProjectIntegrationDataSchema } from "~/v3/vercel/vercelProjectIntegrationSchema";
-import { S2 } from "@s2-dev/streamstore";
+import { createDeploymentS2Client } from "~/v3/s2Client.server";
 import { env } from "~/env.server";
 import { createRedisClient } from "~/redis.server";
 import { tryCatch } from "@trigger.dev/core";
 import { logger } from "~/services/logger.server";
+import { s2CacheScope } from "~/v3/s2CacheScope";
 
 const S2_TOKEN_KEY_PREFIX = "s2-token:project:";
 
@@ -30,7 +31,7 @@ const s2TokenRedis = createRedisClient("s2-token-cache", {
   clusterMode: env.CACHE_REDIS_CLUSTER_MODE_ENABLED === "1",
 });
 
-const s2 = env.S2_ENABLED === "1" ? new S2({ accessToken: env.S2_ACCESS_TOKEN }) : undefined;
+const s2 = createDeploymentS2Client();
 
 export type ErrorData = {
   name: string;
@@ -286,7 +287,7 @@ export class DeploymentPresenter {
       throw new Error("Failed getting S2 access token: S2 is not enabled");
     }
 
-    const redisKey = `${S2_TOKEN_KEY_PREFIX}${projectRef}`;
+    const redisKey = `${S2_TOKEN_KEY_PREFIX}${s2CacheScope(env.S2_DEPLOYMENT_ENDPOINT)}${projectRef}`;
     const cachedToken = await s2TokenRedis.get(redisKey);
 
     if (cachedToken) {

apps/webapp/app/utils/s2Endpoint.ts

@@ -0,0 +1,34 @@
+const LOOPBACK_HOSTS = new Set(["localhost", "127.0.0.1", "::1", "[::1]"]);
+
+function isPrivateIpv4(hostname: string): boolean {
+  const octets = hostname.split(".");
+  if (octets.length !== 4 || octets.some((o) => !/^\d{1,3}$/.test(o))) {
+    return false;
+  }
+
+  const [a, b] = octets.map(Number) as [number, number, number, number];
+  return a === 10 || (a === 172 && b >= 16 && b <= 31) || (a === 192 && b === 168);
+}
+
+// A single-label hostname (no dot) is a container or service name on a private network, which is
+// how the self-hosted stack reaches S2, e.g. `http://s2/v1`.
+function isPrivateHost(hostname: string): boolean {
+  return LOOPBACK_HOSTS.has(hostname) || !hostname.includes(".") || isPrivateIpv4(hostname);
+}
+
+// The S2 access token is sent as a bearer header to whatever endpoint is configured, so cleartext
+// is only acceptable to a host that is not reachable from the public internet.
+export function isValidS2Endpoint(value: string): boolean {
+  let url: URL;
+  try {
+    url = new URL(value);
+  } catch {
+    return false;
+  }
+
+  if (url.protocol === "https:") {
+    return true;
+  }
+
+  return url.protocol === "http:" && url.hostname !== "" && isPrivateHost(url.hostname);
+}

apps/webapp/app/v3/s2CacheScope.ts

@@ -0,0 +1,6 @@
+// Cached S2 read tokens are issued by whichever S2 service the endpoint names, and Redis outlives
+// a restart, so a key scoped only by project would serve a token from the previous service after
+// the endpoint changes. Hosted keeps its existing unscoped keys so nothing is invalidated.
+export function s2CacheScope(endpoint: string | undefined): string {
+  return endpoint === undefined ? "" : `endpoint:${endpoint}:`;
+}

apps/webapp/app/v3/s2Client.server.ts

@@ -0,0 +1,14 @@
+import type { S2 } from "@s2-dev/streamstore";
+import { env } from "~/env.server";
+import { buildDeploymentS2Client } from "~/v3/s2ClientConfig";
+
+export function createDeploymentS2Client(): S2 | undefined {
+  if (env.S2_ENABLED !== "1") {
+    return undefined;
+  }
+
+  return buildDeploymentS2Client({
+    accessToken: env.S2_ACCESS_TOKEN,
+    endpoint: env.S2_DEPLOYMENT_ENDPOINT,
+  });
+}

apps/webapp/app/v3/s2ClientConfig.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, it } from "vitest";
+import { buildDeploymentS2Client, deploymentS2ClientOptions } from "./s2ClientConfig";
+
+const BASIN = "trigger-local";
+
+describe("buildDeploymentS2Client", () => {
+  // The SDK resolves an endpoints key with undefined members to the same hosted URLs, so nothing
+  // on the built client distinguishes the two calls. Assert on the options instead.
+  it("hands the SDK no endpoints key at all when no endpoint is configured", () => {
+    expect(deploymentS2ClientOptions({ accessToken: "token" })).toEqual({ accessToken: "token" });
+    expect(deploymentS2ClientOptions({ accessToken: "token" })).not.toHaveProperty("endpoints");
+  });
+
+  it("hands the SDK one endpoint for both hosts when configured", () => {
+    expect(
+      deploymentS2ClientOptions({ accessToken: "token", endpoint: "http://localhost:4566" })
+    ).toEqual({
+      accessToken: "token",
+      endpoints: { account: "http://localhost:4566", basin: "http://localhost:4566" },
+    });
+  });
+
+  it("still resolves the SDK's hosted defaults when no endpoint is configured", () => {
+    const client = buildDeploymentS2Client({ accessToken: "token" });
+
+    expect(client.endpoints.accountBaseUrl()).toBe("https://a.s2.dev/v1");
+    expect(client.endpoints.basinBaseUrl(BASIN)).toBe(`https://${BASIN}.b.s2.dev/v1`);
+    expect(client.endpoints.includeBasinHeader).toBe(false);
+  });
+
+  it("points both the account and basin hosts at a configured endpoint", () => {
+    const client = buildDeploymentS2Client({
+      accessToken: "token",
+      endpoint: "http://localhost:4566",
+    });
+
+    expect(client.endpoints.accountBaseUrl()).toBe("http://localhost:4566/v1");
+    expect(client.endpoints.basinBaseUrl(BASIN)).toBe("http://localhost:4566/v1");
+    expect(client.endpoints.includeBasinHeader).toBe(true);
+  });
+
+  // A split configuration would send the access token to the hosted service while the operator
+  // believed the client was entirely local, so one value has to drive both hosts.
+  it("never leaves one host hosted while the other is overridden", () => {
+    const client = buildDeploymentS2Client({
+      accessToken: "token",
+      endpoint: "http://localhost:4566",
+    });
+
+    expect(client.endpoints.accountBaseUrl()).not.toContain("s2.dev");
+    expect(client.endpoints.basinBaseUrl(BASIN)).not.toContain("s2.dev");
+  });
+});

apps/webapp/app/v3/s2ClientConfig.ts

@@ -0,0 +1,30 @@
+import { S2 } from "@s2-dev/streamstore";
+
+export type DeploymentS2Config = {
+  accessToken: string;
+  endpoint?: string;
+};
+
+type DeploymentS2ClientOptions = {
+  accessToken: string;
+  endpoints?: { account: string; basin: string };
+};
+
+// Exported so a test can pin the shape handed to the SDK: with no endpoint the options must carry
+// no `endpoints` key, matching the call production already makes.
+export function deploymentS2ClientOptions({
+  accessToken,
+  endpoint,
+}: DeploymentS2Config): DeploymentS2ClientOptions {
+  if (endpoint === undefined) {
+    return { accessToken };
+  }
+
+  // One value drives both hosts. Overriding just one would send the access token to the hosted
+  // service while the other half went elsewhere.
+  return { accessToken, endpoints: { account: endpoint, basin: endpoint } };
+}
+
+export function buildDeploymentS2Client(config: DeploymentS2Config): S2 {
+  return new S2(deploymentS2ClientOptions(config));
+}

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

@@ -23,8 +23,10 @@ import {
 import { FEATURE_FLAG, type FeatureFlagKey } from "../featureFlags";
 import { flags } from "../featureFlags.server";
 import { globalFlagsRegistry } from "../globalFlagsRegistry.server";
-import { AppendInput, AppendRecord, S2 } from "@s2-dev/streamstore";
+import { AppendInput, AppendRecord } from "@s2-dev/streamstore";
+import { createDeploymentS2Client } from "~/v3/s2Client.server";
 import { createRedisClient } from "~/redis.server";
+import { s2CacheScope } from "~/v3/s2CacheScope";
 
 const S2_TOKEN_KEY_PREFIX = "s2-token:read:deployment-event-stream:project:";
 const s2TokenRedis = createRedisClient("s2-token-cache", {
@@ -35,7 +37,7 @@ const s2TokenRedis = createRedisClient("s2-token-cache", {
   tlsDisabled: env.CACHE_REDIS_TLS_DISABLED === "true",
   clusterMode: env.CACHE_REDIS_CLUSTER_MODE_ENABLED === "1",
 });
-const s2 = env.S2_ENABLED === "1" ? new S2({ accessToken: env.S2_ACCESS_TOKEN }) : undefined;
+const s2 = createDeploymentS2Client();
 
 const DEPLOY_BUILD_PATH_ENV_FLAG: Partial<Record<RuntimeEnvironmentType, FeatureFlagKey>> = {
   PREVIEW: FEATURE_FLAG.deployBuildPathPreview,
@@ -522,7 +524,7 @@ export class DeploymentService extends BaseService {
       return errAsync({ type: "s2_is_disabled" as const });
     }
     const basinName = env.S2_DEPLOYMENT_LOGS_BASIN_NAME;
-    const redisKey = `${S2_TOKEN_KEY_PREFIX}${project.externalRef}`;
+    const redisKey = `${S2_TOKEN_KEY_PREFIX}${s2CacheScope(env.S2_DEPLOYMENT_ENDPOINT)}${project.externalRef}`;
 
     const getTokenFromCache = () =>
       fromPromise(s2TokenRedis.get(redisKey), (error) => ({

apps/webapp/test/s2CacheScope.test.ts

@@ -0,0 +1,20 @@
+import { describe, expect, it } from "vitest";
+import { s2CacheScope } from "~/v3/s2CacheScope";
+
+describe("s2CacheScope", () => {
+  // Hosted must keep the keys it already has in Redis, or every project takes a needless miss.
+  it("adds nothing when no endpoint is configured", () => {
+    expect(s2CacheScope(undefined)).toBe("");
+  });
+
+  // Redis outlives a restart, so a token issued by the previous S2 service must not be served
+  // once the endpoint changes.
+  it("gives each endpoint its own namespace", () => {
+    const local = s2CacheScope("http://localhost:4566");
+    const other = s2CacheScope("http://s2/v1");
+
+    expect(local).not.toBe("");
+    expect(local).not.toBe(other);
+    expect(local).toBe(s2CacheScope("http://localhost:4566"));
+  });
+});

apps/webapp/test/s2Endpoint.test.ts

@@ -0,0 +1,41 @@
+import { describe, expect, it } from "vitest";
+import { isValidS2Endpoint } from "~/utils/s2Endpoint";
+
+describe("isValidS2Endpoint", () => {
+  it("accepts https anywhere", () => {
+    expect(isValidS2Endpoint("https://a.s2.dev")).toBe(true);
+    expect(isValidS2Endpoint("https://s2.internal:4566/v1")).toBe(true);
+  });
+
+  it("accepts http to a loopback host", () => {
+    expect(isValidS2Endpoint("http://localhost:4566")).toBe(true);
+    expect(isValidS2Endpoint("http://127.0.0.1:4566")).toBe(true);
+    expect(isValidS2Endpoint("http://[::1]:4566")).toBe(true);
+  });
+
+  // This is how the self-hosted stack reaches S2, so rejecting it would force TLS on a private
+  // container network.
+  it("accepts http to a container or service name and a private address", () => {
+    expect(isValidS2Endpoint("http://s2/v1")).toBe(true);
+    expect(isValidS2Endpoint("http://s2:80/v1")).toBe(true);
+    expect(isValidS2Endpoint("http://10.0.0.5:4566")).toBe(true);
+    expect(isValidS2Endpoint("http://172.20.0.3")).toBe(true);
+    expect(isValidS2Endpoint("http://192.168.1.9")).toBe(true);
+  });
+
+  // The access token is sent to whatever is configured, so cleartext to a routable host leaks it.
+  it("rejects http to a public host", () => {
+    expect(isValidS2Endpoint("http://s2.example.com")).toBe(false);
+    expect(isValidS2Endpoint("http://a.s2.dev")).toBe(false);
+    expect(isValidS2Endpoint("http://8.8.8.8")).toBe(false);
+    expect(isValidS2Endpoint("http://172.32.0.1")).toBe(false);
+  });
+
+  // All four pass zod's `.url()`, which is why the schema refines on this instead.
+  it("rejects malformed and non-http schemes", () => {
+    expect(isValidS2Endpoint("htp:/localhost:4566")).toBe(false);
+    expect(isValidS2Endpoint("ftp://localhost")).toBe(false);
+    expect(isValidS2Endpoint("not a url")).toBe(false);
+    expect(isValidS2Endpoint("")).toBe(false);
+  });
+});

apps/webapp/vitest.config.ts

@@ -13,6 +13,7 @@ export default defineConfig({
       "test/**/*.test.ts",
       "app/v3/runOpsMigration/**/*.test.ts",
       "app/v3/runStore.server.test.ts",
+      "app/v3/s2ClientConfig.test.ts",
       "app/v3/utils/**/*.test.ts",
       "app/v3/services/bulk/**/*.test.ts",
       "app/runEngine/concerns/**/*.test.ts",