Handle errors thrown by requests in Realtime react hooks (#1599)

Author: ericallam

.changeset/many-panthers-relax.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/react-hooks": patch
+---
+
+Make sure useRealtimeRun onComplete hook fires at the correct time

.changeset/silent-dragons-chew.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/react-hooks": patch
+"@trigger.dev/core": patch
+---
+
+Handle errors thrown by requests in Realtime react hooks

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

@@ -640,20 +640,25 @@ export class ApiClient {
 
   subscribeToRun<TRunTypes extends AnyRunTypes>(
     runId: string,
-    options?: { signal?: AbortSignal; closeOnComplete?: boolean }
+    options?: {
+      signal?: AbortSignal;
+      closeOnComplete?: boolean;
+      onFetchError?: (error: Error) => void;
+    }
   ) {
     return runShapeStream<TRunTypes>(`${this.baseUrl}/realtime/v1/runs/${runId}`, {
       closeOnComplete:
         typeof options?.closeOnComplete === "boolean" ? options.closeOnComplete : true,
       headers: this.#getRealtimeHeaders(),
       client: this,
       signal: options?.signal,
+      onFetchError: options?.onFetchError,
     });
   }
 
   subscribeToRunsWithTag<TRunTypes extends AnyRunTypes>(
     tag: string | string[],
-    options?: { signal?: AbortSignal }
+    options?: { signal?: AbortSignal; onFetchError?: (error: Error) => void }
   ) {
     const searchParams = createSearchQueryForSubscribeToRuns({
       tags: tag,
@@ -666,19 +671,21 @@ export class ApiClient {
         headers: this.#getRealtimeHeaders(),
         client: this,
         signal: options?.signal,
+        onFetchError: options?.onFetchError,
       }
     );
   }
 
   subscribeToBatch<TRunTypes extends AnyRunTypes>(
     batchId: string,
-    options?: { signal?: AbortSignal }
+    options?: { signal?: AbortSignal; onFetchError?: (error: Error) => void }
   ) {
     return runShapeStream<TRunTypes>(`${this.baseUrl}/realtime/v1/batches/${batchId}`, {
       closeOnComplete: false,
       headers: this.#getRealtimeHeaders(),
       client: this,
       signal: options?.signal,
+      onFetchError: options?.onFetchError,
     });
   }
 

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

@@ -66,6 +66,7 @@ export type RunShapeStreamOptions = {
   closeOnComplete?: boolean;
   signal?: AbortSignal;
   client?: ApiClient;
+  onFetchError?: (e: Error) => void;
 };
 
 export type StreamPartResult<TRun, TStreams extends Record<string, any>> = {
@@ -111,6 +112,9 @@ export function runShapeStream<TRunTypes extends AnyRunTypes>(
   const runStreamInstance = zodShapeStream(SubscribeRunRawShape, url, {
     ...options,
     signal: abortController.signal,
+    onError: (e) => {
+      options?.onFetchError?.(e);
+    },
   });
 
   const $options: RunSubscriptionOptions = {

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

@@ -14,6 +14,7 @@ export type ZodShapeStreamOptions = {
   headers?: Record<string, string>;
   fetchClient?: typeof fetch;
   signal?: AbortSignal;
+  onError?: (e: Error) => void;
 };
 
 export type ZodShapeStreamInstance<TShapeSchema extends z.ZodTypeAny> = {
@@ -44,6 +45,9 @@ export function zodShapeStream<TShapeSchema extends z.ZodTypeAny>(
     },
     fetchClient: options?.fetchClient,
     signal: abortController.signal,
+    onError: (e) => {
+      options?.onError?.(e);
+    },
   });
 
   const readableShape = new ReadableShapeStream(shapeStream);

packages/react-hooks/src/hooks/useRealtime.ts

@@ -103,6 +103,7 @@ export function useRealtimeRun<TTask extends AnyTask>(
         runId,
         apiClient,
         mutateRun,
+        setError,
         abortControllerRef,
         typeof options?.stopOnCompletion === "boolean" ? options.stopOnCompletion : true
       );
@@ -150,6 +151,12 @@ export function useRealtimeRun<TTask extends AnyTask>(
     };
   }, [runId, stop, options?.enabled]);
 
+  useEffect(() => {
+    if (run?.finishedAt) {
+      setIsComplete(true);
+    }
+  }, [run]);
+
   return { run, error, stop };
 }
 
@@ -258,6 +265,7 @@ export function useRealtimeRunWithStreams<
         mutateRun,
         mutateStreams,
         streamsRef,
+        setError,
         abortControllerRef,
         typeof options?.stopOnCompletion === "boolean" ? options.stopOnCompletion : true,
         options?.experimental_throttleInMs
@@ -306,6 +314,12 @@ export function useRealtimeRunWithStreams<
     };
   }, [runId, stop, options?.enabled]);
 
+  useEffect(() => {
+    if (run?.finishedAt) {
+      setIsComplete(true);
+    }
+  }, [run]);
+
   return { run, streams: streams ?? initialStreamsFallback, error, stop };
 }
 
@@ -380,7 +394,14 @@ export function useRealtimeRunsWithTag<TTask extends AnyTask>(
       const abortController = new AbortController();
       abortControllerRef.current = abortController;
 
-      await processRealtimeRunsWithTag(tag, apiClient, mutateRuns, runsRef, abortControllerRef);
+      await processRealtimeRunsWithTag(
+        tag,
+        apiClient,
+        mutateRuns,
+        runsRef,
+        setError,
+        abortControllerRef
+      );
     } catch (err) {
       // Ignore abort errors as they are expected.
       if ((err as any).name === "AbortError") {
@@ -470,7 +491,14 @@ export function useRealtimeBatch<TTask extends AnyTask>(
       const abortController = new AbortController();
       abortControllerRef.current = abortController;
 
-      await processRealtimeBatch(batchId, apiClient, mutateRuns, runsRef, abortControllerRef);
+      await processRealtimeBatch(
+        batchId,
+        apiClient,
+        mutateRuns,
+        runsRef,
+        setError,
+        abortControllerRef
+      );
     } catch (err) {
       // Ignore abort errors as they are expected.
       if ((err as any).name === "AbortError") {
@@ -506,10 +534,12 @@ async function processRealtimeBatch<TTask extends AnyTask = AnyTask>(
   apiClient: ApiClient,
   mutateRunsData: KeyedMutator<RealtimeRun<TTask>[]>,
   existingRunsRef: React.MutableRefObject<RealtimeRun<TTask>[]>,
+  onError: (e: Error) => void,
   abortControllerRef: React.MutableRefObject<AbortController | null>
 ) {
   const subscription = apiClient.subscribeToBatch<InferRunTypes<TTask>>(batchId, {
     signal: abortControllerRef.current?.signal,
+    onFetchError: onError,
   });
 
   for await (const part of subscription) {
@@ -541,10 +571,12 @@ async function processRealtimeRunsWithTag<TTask extends AnyTask = AnyTask>(
   apiClient: ApiClient,
   mutateRunsData: KeyedMutator<RealtimeRun<TTask>[]>,
   existingRunsRef: React.MutableRefObject<RealtimeRun<TTask>[]>,
+  onError: (e: Error) => void,
   abortControllerRef: React.MutableRefObject<AbortController | null>
 ) {
   const subscription = apiClient.subscribeToRunsWithTag<InferRunTypes<TTask>>(tag, {
     signal: abortControllerRef.current?.signal,
+    onFetchError: onError,
   });
 
   for await (const part of subscription) {
@@ -582,13 +614,15 @@ async function processRealtimeRunWithStreams<
   mutateRunData: KeyedMutator<RealtimeRun<TTask>>,
   mutateStreamData: KeyedMutator<StreamResults<TStreams>>,
   existingDataRef: React.MutableRefObject<StreamResults<TStreams>>,
+  onError: (e: Error) => void,
   abortControllerRef: React.MutableRefObject<AbortController | null>,
   stopOnCompletion: boolean = true,
   throttleInMs?: number
 ) {
   const subscription = apiClient.subscribeToRun<InferRunTypes<TTask>>(runId, {
     signal: abortControllerRef.current?.signal,
     closeOnComplete: stopOnCompletion,
+    onFetchError: onError,
   });
 
   type StreamUpdate = {
@@ -637,12 +671,14 @@ async function processRealtimeRun<TTask extends AnyTask = AnyTask>(
   runId: string,
   apiClient: ApiClient,
   mutateRunData: KeyedMutator<RealtimeRun<TTask>>,
+  onError: (e: Error) => void,
   abortControllerRef: React.MutableRefObject<AbortController | null>,
   stopOnCompletion: boolean = true
 ) {
   const subscription = apiClient.subscribeToRun<InferRunTypes<TTask>>(runId, {
     signal: abortControllerRef.current?.signal,
     closeOnComplete: stopOnCompletion,
+    onFetchError: onError,
   });
 
   for await (const part of subscription) {

references/nextjs-realtime/src/app/actions.ts

@@ -1,7 +1,7 @@
 "use server";
 
 import type { exampleTask } from "@/trigger/example";
-import { tasks } from "@trigger.dev/sdk/v3";
+import { auth, tasks } from "@trigger.dev/sdk/v3";
 import { cookies } from "next/headers";
 import { redirect } from "next/navigation";
 import { randomUUID } from "node:crypto";
@@ -11,10 +11,19 @@ export async function triggerExampleTask() {
     id: randomUUID(),
   });
 
-  console.log("Setting the run JWT in a cookie", handle.publicAccessToken);
+  const publicToken = await auth.createPublicToken({
+    scopes: {
+      read: {
+        runs: [handle.id],
+      },
+    },
+    expirationTime: "2s",
+  });
+
+  console.log("Setting the run JWT in a cookie", publicToken);
 
   // Set JWT in a secure, HTTP-only cookie
-  cookies().set("run_token", handle.publicAccessToken);
+  cookies().set("run_token", publicToken);
 
   // Redirect to the details page
   redirect(`/runs/${handle.id}`);

references/nextjs-realtime/src/app/runs/[id]/ClientRunDetails.tsx

@@ -32,7 +32,7 @@ function RunDetailsWrapper({
     },
   });
 
-  if (error) {
+  if (error && !run) {
     return (
       <div className="w-full min-h-screen bg-gray-900 p-4">
         <Card className="w-full bg-gray-800 shadow-md">