Upgrade the deployment s2 client to 0.22.5 as well and fix some typecheck issues

Author: ericallam

apps/webapp/app/env.server.ts

@@ -39,6 +39,7 @@ const S2EnvSchema = z.preprocess(
       S2_ENABLED: z.literal("1"),
       S2_ACCESS_TOKEN: z.string(),
       S2_DEPLOYMENT_LOGS_BASIN_NAME: z.string(),
+      S2_DEPLOYMENT_STREAMS_LOCAL: z.string().default("0"),
     }),
     z.object({
       S2_ENABLED: z.literal("0"),

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

@@ -217,7 +217,7 @@ export class DeploymentPresenter {
     let eventStream = undefined;
     if (
       env.S2_ENABLED === "1" &&
-      (buildServerMetadata || gitMetadata?.source === "trigger_github_app")
+      (buildServerMetadata || gitMetadata?.source === "trigger_github_app" || env.S2_DEPLOYMENT_STREAMS_LOCAL === "1")
     ) {
       const [error, accessToken] = await tryCatch(this.getS2AccessToken(project.externalRef));
 
@@ -290,9 +290,9 @@ export class DeploymentPresenter {
       return cachedToken;
     }
 
-    const { access_token: accessToken } = await s2.accessTokens.issue({
+    const { accessToken } = await s2.accessTokens.issue({
       id: `${projectRef}-${new Date().getTime()}`,
-      expires_at: new Date(Date.now() + 60 * 60 * 1000).toISOString(), // 1 hour
+      expiresAt: new Date(Date.now() + 60 * 60 * 1000), // 1 hour
       scope: {
         ops: ["read"],
         basins: {

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.deployments.$deploymentParam/route.tsx

@@ -198,17 +198,14 @@ export default function Page() {
 
         const readSession = await stream.readSession(
           {
-            seq_num: 0,
-            wait: 60,
-            as: "bytes",
+            start: { from: { seqNum: 0 }, clamp: true },
+            stop: { waitSecs: 60 },
           },
           { signal: abortController.signal }
         );
 
-        const decoder = new TextDecoder();
-
         for await (const record of readSession) {
-          const decoded = decoder.decode(record.body);
+          const decoded = record.body;
           const result = DeploymentEventFromString.safeParse(decoded);
 
           if (!result.success) {
@@ -217,8 +214,8 @@ export default function Page() {
               const headers: Record<string, string> = {};
 
               if (record.headers) {
-                for (const [nameBytes, valueBytes] of record.headers) {
-                  headers[decoder.decode(nameBytes)] = decoder.decode(valueBytes);
+                for (const [name, value] of record.headers) {
+                  headers[name] = value;
                 }
               }
               const level = (headers["level"]?.toLowerCase() as LogEntry["level"]) ?? "info";

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts

@@ -101,29 +101,27 @@ const { action, loader } = createActionApiRoute(
             run.realtimeStreamsVersion
           );
 
-          if (realtimeStream.readRecords) {
-            const records = await realtimeStream.readRecords(
-              run.friendlyId,
-              `$trigger.input:${body.streamId}`,
-              body.lastSeqNum
-            );
-
-            if (records.length > 0) {
-              const record = records[0]!;
-
-              // Record data is the raw user payload — no wrapper to unwrap
-              await engine.completeWaitpoint({
-                id: result.waitpoint.id,
-                output: {
-                  value: record.data,
-                  type: "application/json",
-                  isError: false,
-                },
-              });
-
-              // Clean up the Redis cache since we completed it ourselves
-              await deleteInputStreamWaitpoint(run.friendlyId, body.streamId);
-            }
+          const records = await realtimeStream.readRecords(
+            run.friendlyId,
+            `$trigger.input:${body.streamId}`,
+            body.lastSeqNum
+          );
+
+          if (records.length > 0) {
+            const record = records[0]!;
+
+            // Record data is the raw user payload — no wrapper to unwrap
+            await engine.completeWaitpoint({
+              id: result.waitpoint.id,
+              output: {
+                value: record.data,
+                type: "application/json",
+                isError: false,
+              },
+            });
+
+            // Clean up the Redis cache since we completed it ourselves
+            await deleteInputStreamWaitpoint(run.friendlyId, body.streamId);
           }
         } catch {
           // Non-fatal: if the S2 check fails, the waitpoint is still PENDING.

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

@@ -466,4 +466,8 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
       });
     }
   }
+
+  async readRecords(): Promise<never> {
+    throw new Error("readRecords is not implemented for Redis realtime streams");
+  }
 }

apps/webapp/app/services/realtime/types.ts

@@ -23,16 +23,11 @@ export interface StreamIngestor {
 
   getLastChunkIndex(runId: string, streamId: string, clientId: string): Promise<number>;
 
-  /**
-   * Read records from a stream starting after a given sequence number.
-   * Returns immediately with whatever records exist (non-blocking).
-   * Not all backends support this — returns undefined if unsupported.
-   */
-  readRecords?(
+  readRecords(
     runId: string,
     streamId: string,
     afterSeqNum?: number
-  ): Promise<StreamRecord[] | undefined>;
+  ): Promise<StreamRecord[]>;
 }
 
 export type StreamResponseOptions = {

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

@@ -13,7 +13,7 @@ import { env } from "~/env.server";
 import { createRemoteImageBuild } from "../remoteImageBuilder.server";
 import { FINAL_DEPLOYMENT_STATUSES } from "./failDeployment.server";
 import { enqueueBuild, generateRegistryCredentials } from "~/services/platform.v3.server";
-import { AppendRecord, S2 } from "@s2-dev/streamstore";
+import { AppendInput, AppendRecord, S2 } from "@s2-dev/streamstore";
 import { createRedisClient } from "~/redis.server";
 
 const S2_TOKEN_KEY_PREFIX = "s2-token:read:deployment-event-stream:project:";
@@ -368,7 +368,11 @@ export class DeploymentService extends BaseService {
     );
 
     return fromPromise(
-      stream.append(events.map((event) => AppendRecord.string({ body: JSON.stringify(event) }))),
+      stream.append(
+        AppendInput.create(
+          events.map((event) => AppendRecord.string({ body: JSON.stringify(event) }))
+        )
+      ),
       (error) => ({
         type: "failed_to_append_to_event_log" as const,
         cause: error,
@@ -396,9 +400,9 @@ export class DeploymentService extends BaseService {
         type: "failed_to_create_event_stream" as const,
         cause: error,
       })
-    ).map(({ name }) => ({
+    ).map(() => ({
       basin: basin.name,
-      stream: name,
+      stream: `projects/${project.externalRef}/deployments/${deployment.shortCode}`,
     }));
   }
 
@@ -426,7 +430,7 @@ export class DeploymentService extends BaseService {
       fromPromise(
         s2.accessTokens.issue({
           id: `${project.externalRef}-${new Date().getTime()}`,
-          expires_at: new Date(Date.now() + 60 * 60 * 1000).toISOString(), // 1 hour
+          expiresAt: new Date(Date.now() + 60 * 60 * 1000), // 1 hour
           scope: {
             ops: ["read"],
             basins: {
@@ -441,7 +445,7 @@ export class DeploymentService extends BaseService {
           type: "other" as const,
           cause: error,
         })
-      ).map(({ access_token }) => access_token);
+      ).map(({ accessToken }) => accessToken);
 
     const cacheToken = (token: string) =>
       fromPromise(

packages/cli-v3/package.json

@@ -92,7 +92,7 @@
     "@opentelemetry/resources": "2.0.1",
     "@opentelemetry/sdk-trace-node": "2.0.1",
     "@opentelemetry/semantic-conventions": "1.36.0",
-    "@s2-dev/streamstore": "^0.17.6",
+    "@s2-dev/streamstore": "^0.22.5",
     "@trigger.dev/build": "workspace:4.4.1",
     "@trigger.dev/core": "workspace:4.4.1",
     "@trigger.dev/schema-to-json": "workspace:4.4.1",

packages/cli-v3/src/commands/deploy.ts

@@ -1155,9 +1155,8 @@ async function handleNativeBuildServerDeploy({
   const [readSessionError, readSession] = await tryCatch(
     stream.readSession(
       {
-        seq_num: 0,
-        wait: 60 * 20, // 20 minutes
-        as: "bytes",
+        start: { from: { seqNum: 0 }, clamp: true },
+        stop: { waitSecs: 60 * 20 }, // 20 minutes
       },
       { signal: abortController.signal }
     )
@@ -1176,12 +1175,11 @@ async function handleNativeBuildServerDeploy({
     return process.exit(0);
   }
 
-  const decoder = new TextDecoder();
   let finalDeploymentEvent: DeploymentFinalizedEvent["data"] | undefined;
   let queuedSpinnerStopped = false;
 
   for await (const record of readSession) {
-    const decoded = decoder.decode(record.body);
+    const decoded = record.body;
     const result = DeploymentEventFromString.safeParse(decoded);
     if (!result.success) {
       logger.debug("Failed to parse deployment event, skipping", {
@@ -1195,7 +1193,7 @@ async function handleNativeBuildServerDeploy({
 
     switch (event.type) {
       case "log": {
-        if (record.seq_num === 0) {
+        if (record.seqNum === 0) {
           $queuedSpinner.stop("Build started");
           console.log("│");
           queuedSpinnerStopped = true;