Add ability to fetch streams through metadata

Author: ericallam

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

@@ -59,12 +59,14 @@ import {
   SubscribeToRunsQueryParams,
   UpdateEnvironmentVariableParams,
 } from "./types.js";
+import type { AsyncIterableStream } from "./stream.js";
 
 export type {
   CreateEnvironmentVariableParams,
   ImportEnvironmentVariablesParams,
   SubscribeToRunsQueryParams,
   UpdateEnvironmentVariableParams,
+  AsyncIterableStream,
 };
 
 export type ClientTriggerOptions = {

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

@@ -1,4 +1,5 @@
 import { DeserializedJson } from "../../schemas/json.js";
+import { AsyncIterableStream } from "../apiClient/stream.js";
 import { getGlobal, registerGlobal } from "../utils/globals.js";
 import { ApiRequestOptions } from "../zodfetch.js";
 import { NoopRunMetadataManager } from "./noopManager.js";
@@ -84,6 +85,10 @@ export class RunMetadataAPI implements RunMetadataManager {
     return this.#getManager().stream(key, value, signal);
   }
 
+  public fetchStream<T>(key: string, signal?: AbortSignal): Promise<AsyncIterableStream<T>> {
+    return this.#getManager().fetchStream(key, signal);
+  }
+
   flush(requestOptions?: ApiRequestOptions): Promise<void> {
     return this.#getManager().flush(requestOptions);
   }

packages/core/src/v3/runMetadata/manager.ts

@@ -7,6 +7,8 @@ import { MetadataStream } from "./metadataStream.js";
 import { ApiClient } from "../apiClient/index.js";
 import { FlushedRunMetadata, RunMetadataChangeOperation } from "../schemas/common.js";
 import { applyMetadataOperations } from "./operations.js";
+import { SSEStreamSubscriptionFactory } from "../apiClient/runStream.js";
+import { AsyncIterableStream } from "../apiClient/stream.js";
 
 const MAXIMUM_ACTIVE_STREAMS = 5;
 const MAXIMUM_TOTAL_STREAMS = 10;
@@ -197,6 +199,25 @@ export class StandardMetadataManager implements RunMetadataManager {
     return this.doStream(key, value, "self", this, signal);
   }
 
+  public async fetchStream<T>(key: string, signal?: AbortSignal): Promise<AsyncIterableStream<T>> {
+    if (!this.runId) {
+      throw new Error("Run ID is required to fetch metadata streams.");
+    }
+
+    const baseUrl = this.getKey("$$streamsBaseUrl");
+
+    const $baseUrl = typeof baseUrl === "string" ? baseUrl : this.streamsBaseUrl;
+
+    const streamFactory = new SSEStreamSubscriptionFactory($baseUrl, {
+      headers: this.apiClient.getHeaders(),
+      signal,
+    });
+
+    const subscription = streamFactory.createSubscription(this.store ?? {}, this.runId, key);
+
+    return (await subscription.subscribe()) as AsyncIterableStream<T>;
+  }
+
   private async doStream<T>(
     key: string,
     value: AsyncIterable<T> | ReadableStream<T>,

packages/core/src/v3/runMetadata/noopManager.ts

@@ -1,4 +1,5 @@
 import { DeserializedJson } from "../../schemas/json.js";
+import { AsyncIterableStream } from "../apiClient/stream.js";
 import { ApiRequestOptions } from "../zodfetch.js";
 import type { RunMetadataManager, RunMetadataUpdater } from "./types.js";
 
@@ -18,6 +19,9 @@ export class NoopRunMetadataManager implements RunMetadataManager {
   stream<T>(key: string, value: AsyncIterable<T>): Promise<AsyncIterable<T>> {
     throw new Error("Method not implemented.");
   }
+  fetchStream<T>(key: string, signal?: AbortSignal): Promise<AsyncIterableStream<T>> {
+    throw new Error("Method not implemented.");
+  }
   flush(requestOptions?: ApiRequestOptions): Promise<void> {
     throw new Error("Method not implemented.");
   }

packages/core/src/v3/runMetadata/types.ts

@@ -1,4 +1,5 @@
 import { DeserializedJson } from "../../schemas/json.js";
+import { AsyncIterableStream } from "../apiClient/stream.js";
 import { ApiRequestOptions } from "../zodfetch.js";
 
 export interface RunMetadataUpdater {
@@ -23,6 +24,7 @@ export interface RunMetadataManager extends RunMetadataUpdater {
   getKey(key: string): DeserializedJson | undefined;
   flush(requestOptions?: ApiRequestOptions): Promise<void>;
   refresh(requestOptions?: ApiRequestOptions): Promise<void>;
+  fetchStream<T>(key: string, signal?: AbortSignal): Promise<AsyncIterableStream<T>>;
 
   get parent(): RunMetadataUpdater;
   get root(): RunMetadataUpdater;

packages/trigger-sdk/src/v3/metadata.ts

@@ -4,6 +4,7 @@ import {
   mergeRequestOptions,
   runMetadata,
   type RunMetadataUpdater,
+  type AsyncIterableStream,
 } from "@trigger.dev/core/v3";
 import { tracer } from "./tracer.js";
 
@@ -36,6 +37,7 @@ export const metadata = {
   save: saveMetadata,
   replace: replaceMetadata,
   stream: stream,
+  fetchStream: fetchStream,
   parent: parentMetadataUpdater,
   root: rootMetadataUpdater,
   refresh: refreshMetadata,
@@ -233,3 +235,7 @@ async function stream<T>(
 ): Promise<AsyncIterable<T>> {
   return runMetadata.stream(key, value, signal);
 }
+
+async function fetchStream<T>(key: string, signal?: AbortSignal): Promise<AsyncIterableStream<T>> {
+  return runMetadata.fetchStream<T>(key, signal);
+}

references/nextjs-realtime/src/trigger/csv.ts

@@ -62,6 +62,16 @@ export const handleCSVUpload = schemaTask({
     const successfulRows = results.runs.filter((r) => r.ok);
     const failedRows = results.runs.filter((r) => !r.ok);
 
+    const firstSuccessfulRow = successfulRows[0];
+
+    if (firstSuccessfulRow) {
+      const stream = await metadata.fetchStream<string>(firstSuccessfulRow.id);
+
+      for await (const value of stream) {
+        logger.info(`Stream value from ${firstSuccessfulRow.id}`, { value });
+      }
+    }
+
     return {
       file,
       rows,