fix(core): define RunEvent schema and update ApiClient validation

.changeset/define-runevent-schema.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": minor
+---
+
+Define RunEvent schema and update ApiClient to use it

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

@@ -501,9 +501,8 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
   const version = deployment.version;
 
   const rawDeploymentLink = `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project}/deployments/${deployment.shortCode}`;
-  const rawTestLink = `${authorization.dashboardUrl}/projects/v3/${
-    resolvedConfig.project
-  }/test?environment=${options.env === "prod" ? "prod" : "stg"}`;
+  const rawTestLink = `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project
+    }/test?environment=${options.env === "prod" ? "prod" : "stg"}`;
 
   const deploymentLink = cliLink("View deployment", rawDeploymentLink);
   const testLink = cliLink("Test tasks", rawTestLink);
@@ -720,8 +719,7 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
     }
   } else {
     outro(
-      `Version ${version} deployed with ${taskCount} detected task${taskCount === 1 ? "" : "s"} ${
-        isLinksSupported ? `| ${deploymentLink} | ${testLink}` : ""
+      `Version ${version} deployed with ${taskCount} detected task${taskCount === 1 ? "" : "s"} ${isLinksSupported ? `| ${deploymentLink} | ${testLink}` : ""
       }`
     );
 
@@ -745,18 +743,16 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
       TRIGGER_VERSION: version,
       TRIGGER_DEPLOYMENT_SHORT_CODE: deployment.shortCode,
       TRIGGER_DEPLOYMENT_URL: `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project}/deployments/${deployment.shortCode}`,
-      TRIGGER_TEST_URL: `${authorization.dashboardUrl}/projects/v3/${
-        resolvedConfig.project
-      }/test?environment=${options.env === "prod" ? "prod" : "stg"}`,
+      TRIGGER_TEST_URL: `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project
+        }/test?environment=${options.env === "prod" ? "prod" : "stg"}`,
     },
     outputs: {
       deploymentVersion: version,
       workerVersion: version,
       deploymentShortCode: deployment.shortCode,
       deploymentUrl: `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project}/deployments/${deployment.shortCode}`,
-      testUrl: `${authorization.dashboardUrl}/projects/v3/${
-        resolvedConfig.project
-      }/test?environment=${options.env === "prod" ? "prod" : "stg"}`,
+      testUrl: `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project
+        }/test?environment=${options.env === "prod" ? "prod" : "stg"}`,
       needsPromotion: options.skipPromotion ? "true" : "false",
     },
   });
@@ -799,8 +795,7 @@ async function failDeploy(
       checkLogsForErrors(logs);
 
       outro(
-        `${chalkError(`${prefix}:`)} ${
-          error.message
+        `${chalkError(`${prefix}:`)} ${error.message
         }. Full build logs have been saved to ${logPath}`
       );
 
@@ -1100,9 +1095,8 @@ async function handleNativeBuildServerDeploy({
   const deployment = initializeDeploymentResult.data;
 
   const rawDeploymentLink = `${dashboardUrl}/projects/v3/${config.project}/deployments/${deployment.shortCode}`;
-  const rawTestLink = `${dashboardUrl}/projects/v3/${config.project}/test?environment=${
-    options.env === "prod" ? "prod" : "stg"
-  }`;
+  const rawTestLink = `${dashboardUrl}/projects/v3/${config.project}/test?environment=${options.env === "prod" ? "prod" : "stg"
+    }`;
 
   const exposedDeploymentLink = isLinksSupported
     ? cliLink(chalk.bold(rawDeploymentLink), rawDeploymentLink)
@@ -1155,8 +1149,9 @@ async function handleNativeBuildServerDeploy({
   const [readSessionError, readSession] = await tryCatch(
     stream.readSession(
       {
-        start: { from: { seqNum: 0 }, clamp: true },
-        stop: { waitSecs: 60 * 20 }, // 20 minutes
+        seq_num: 0,
+        clamp: true,
+        wait: 60 * 20, // 20 minutes
       },
       { signal: abortController.signal }
     )
@@ -1167,8 +1162,7 @@ async function handleNativeBuildServerDeploy({
     log.warn(`Failed streaming build logs, open the deployment in the dashboard to view the logs`);
 
     outro(
-      `Version ${deployment.version} is being deployed ${
-        isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+      `Version ${deployment.version} is being deployed ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
       }`
     );
 
@@ -1193,7 +1187,7 @@ async function handleNativeBuildServerDeploy({
 
     switch (event.type) {
       case "log": {
-        if (record.seqNum === 0) {
+        if (record.seq_num === 0) {
           $queuedSpinner.stop("Build started");
           console.log("│");
           queuedSpinnerStopped = true;
@@ -1214,10 +1208,10 @@ async function handleNativeBuildServerDeploy({
           level === "error"
             ? chalk.bold(chalkError(message))
             : level === "warn"
-            ? chalkWarning(message)
-            : level === "debug"
-            ? chalkGrey(message)
-            : message;
+              ? chalkWarning(message)
+              : level === "debug"
+                ? chalkGrey(message)
+                : message;
 
         // We use console.log here instead of clack's logger as the current version does not support changing the line spacing.
         // And the logs look verbose with the default spacing.
@@ -1250,8 +1244,7 @@ async function handleNativeBuildServerDeploy({
     log.error("Failed dequeueing build, please try again shortly");
 
     throw new OutroCommandError(
-      `Version ${deployment.version} ${
-        isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+      `Version ${deployment.version} ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
       }`
     );
   }
@@ -1266,8 +1259,7 @@ async function handleNativeBuildServerDeploy({
     }
 
     throw new OutroCommandError(
-      `Version ${deployment.version} ${
-        isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+      `Version ${deployment.version} ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
       }`
     );
   }
@@ -1293,13 +1285,12 @@ async function handleNativeBuildServerDeploy({
       }
 
       outro(
-        `Version ${deployment.version} was deployed ${
-          isLinksSupported
-            ? `| ${cliLink("Test tasks", rawTestLink)} | ${cliLink(
-                "View deployment",
-                rawDeploymentLink
-              )}`
-            : ""
+        `Version ${deployment.version} was deployed ${isLinksSupported
+          ? `| ${cliLink("Test tasks", rawTestLink)} | ${cliLink(
+            "View deployment",
+            rawDeploymentLink
+          )}`
+          : ""
         }`
       );
       return process.exit(0);
@@ -1313,14 +1304,13 @@ async function handleNativeBuildServerDeploy({
         chalk.bold(
           chalkError(
             "Deployment failed" +
-              (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
+            (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
           )
         )
       );
 
       throw new OutroCommandError(
-        `Version ${deployment.version} deployment failed ${
-          isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+        `Version ${deployment.version} deployment failed ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
         }`
       );
     }
@@ -1333,14 +1323,13 @@ async function handleNativeBuildServerDeploy({
         chalk.bold(
           chalkError(
             "Deployment timed out" +
-              (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
+            (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
           )
         )
       );
 
       throw new OutroCommandError(
-        `Version ${deployment.version} deployment timed out ${
-          isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+        `Version ${deployment.version} deployment timed out ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
         }`
       );
     }
@@ -1353,14 +1342,13 @@ async function handleNativeBuildServerDeploy({
         chalk.bold(
           chalkError(
             "Deployment was canceled" +
-              (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
+            (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
           )
         )
       );
 
       throw new OutroCommandError(
-        `Version ${deployment.version} deployment canceled ${
-          isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+        `Version ${deployment.version} deployment canceled ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
         }`
       );
     }
@@ -1379,13 +1367,12 @@ async function handleNativeBuildServerDeploy({
       }
 
       outro(
-        `Version ${deployment.version} ${
-          isLinksSupported
-            ? `| ${cliLink("Test tasks", rawTestLink)} | ${cliLink(
-                "View deployment",
-                rawDeploymentLink
-              )}`
-            : ""
+        `Version ${deployment.version} ${isLinksSupported
+          ? `| ${cliLink("Test tasks", rawTestLink)} | ${cliLink(
+            "View deployment",
+            rawDeploymentLink
+          )}`
+          : ""
         }`
       );
       return process.exit(0);

packages/core/src/v3/apiClient/index.ts

@@ -28,6 +28,7 @@ import {
   EnvironmentVariableResponseBody,
   EnvironmentVariableWithSecret,
   ListQueueOptions,
+  ListRunEventsResponse,
   ListRunResponseItem,
   ListScheduleOptions,
   QueueItem,
@@ -42,6 +43,7 @@ import {
   RetrieveQueueParam,
   RetrieveRunResponse,
   RetrieveRunTraceResponseBody,
+  RunEvent,
   ScheduleObject,
   SendInputStreamResponseBody,
   StreamBatchItemsResponse,
@@ -700,7 +702,7 @@ export class ApiClient {
 
   listRunEvents(runId: string, requestOptions?: ZodFetchOptions) {
     return zodfetch(
-      z.any(), // TODO: define a proper schema for this
+      ListRunEventsResponse,
       `${this.baseUrl}/api/v1/runs/${runId}/events`,
       {
         method: "GET",

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts

@@ -61,11 +61,11 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       accessToken: options.accessToken,
       ...(options.endpoint
         ? {
-            endpoints: {
-              account: options.endpoint,
-              basin: options.endpoint,
-            },
-          }
+          endpoints: {
+            account: options.endpoint,
+            basin: options.endpoint,
+          },
+        }
         : {}),
     });
     this.flushIntervalMs = options.flushIntervalMs ?? 200;
@@ -135,7 +135,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       const stream = basin.stream(this.options.stream);
 
       const session = await stream.appendSession({
-        maxInflightBytes: this.maxInflightBytes,
+        maxQueuedBytes: this.maxInflightBytes,
       });
 
       this.sessionWritable = session.writable;
@@ -152,7 +152,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
                 return;
               }
               // Convert each chunk to JSON string and wrap in AppendRecord
-              controller.enqueue(AppendRecord.string({ body: JSON.stringify({ data: chunk, id: nanoid(7) }) }));
+              controller.enqueue(AppendRecord.make(JSON.stringify({ data: chunk, id: nanoid(7) })));
             },
           })
         )
@@ -169,9 +169,9 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       const lastAcked = session.lastAckedPosition();
 
       if (lastAcked?.end) {
-        const recordsWritten = lastAcked.end.seqNum;
+        const recordsWritten = lastAcked.end.seq_num;
         this.log(
-          `[S2MetadataStream] Written ${recordsWritten} records, ending at seqNum=${lastAcked.end.seqNum}`
+          `[S2MetadataStream] Written ${recordsWritten} records, ending at seq_num=${lastAcked.end.seq_num}`
         );
       }
     } catch (error) {
@@ -223,5 +223,5 @@ async function* streamToAsyncIterator<T>(stream: ReadableStream<T>): AsyncIterab
 function safeReleaseLock(reader: ReadableStreamDefaultReader<any>) {
   try {
     reader.releaseLock();
-  } catch (error) {}
+  } catch (error) { }
 }