Relevant source files

The following files were used as context for generating this wiki page: - [packages/react-hooks/src/hooks/useRealtime.ts](https://github.com/triggerdotdev/trigger.dev/blob/main/packages/react-hooks/src/hooks/useRealtime.ts) - [rules/4.1.0/realtime.md](https://github.com/triggerdotdev/trigger.dev/blob/main/rules/4.1.0/realtime.md) - [rules/4.0.0/realtime.md](https://github.com/triggerdotdev/trigger.dev/blob/main/rules/4.0.0/realtime.md) - [references/realtime-streams/TESTING.md](https://github.com/triggerdotdev/trigger.dev/blob/main/references/realtime-streams/TESTING.md) - [references/realtime-hooks-test/README.md](https://github.com/triggerdotdev/trigger.dev/blob/main/references/realtime-hooks-test/README.md) - [packages/react-hooks/CHANGELOG.md](https://github.com/triggerdotdev/trigger.dev/blob/main/packages/react-hooks/CHANGELOG.md)

# Realtime Subscriptions & Streaming Trigger.dev Realtime provides a mechanism for monitoring and receiving live updates for task runs, metadata changes, and data streams. It allows developers to build responsive user interfaces that reflect the progress and output of long-running background tasks without persistent polling. This system is primarily exposed through the `@trigger.dev/sdk` for backend subscriptions and `@trigger.dev/react-hooks` for frontend integration. The system supports multiple subscription modes, including single run monitoring, batch updates, and tag-based subscriptions. Furthermore, it facilitates high-performance data streaming, often used for AI responses (e.g., LLM tokens), using a "v2" streaming architecture that supports reconnection and resume functionality. Sources: [rules/4.1.0/realtime.md:1-10](), [README.md:30-35]() ## Architecture & Data Flow The Realtime architecture bridges the gap between background task execution and the frontend. It utilizes a subscription-based model where clients (backend or frontend) establish connections to the Trigger.dev platform to receive pushed updates. ### Core Architecture Components - **Trigger.dev Platform**: The central hub that manages task states and distributes updates to subscribers. In self-hosted environments, this includes services like ElectricSQL for real-time data sync. - **Backend SDK**: Provides `runs.subscribeToRun` and `streams.read` for server-to-server real-time communication. - **Frontend Hooks**: The `@trigger.dev/react-hooks` package provides high-level React hooks like `useRealtimeRun` and `useRealtimeStream` that manage connection lifecycles, abort controllers, and state updates. - **Access Tokens**: Security is handled via Public Access Tokens which scope read permissions to specific runs, tasks, or tags. Sources: [rules/4.1.0/realtime.md:15-30](), [hosting/k8s/helm/README.md:120-130](), [packages/react-hooks/src/hooks/useRealtime.ts:1-10]() ### Connection Flow Diagram The following diagram illustrates how a frontend application establishes a realtime subscription to a run. ```mermaid sequenceDiagram participant UI as "React Frontend" participant API as "Trigger.dev API" participant Task as "Background Task" UI->>API: GET /realtime/v1/streams/ (with Access Token) API-->>UI: Established SSE/Stream Connection Task->>API: Update Metadata/Status API-->>UI: Push Delta/Chunk Note over UI: Hook updates local state & re-renders Task->>API: Run Completed API-->>UI: Final Run Object Note over UI: onComplete() callback fires ``` Sources: [packages/react-hooks/src/hooks/useRealtime.ts:98-150](), [references/realtime-streams/TESTING.md:10-25]() ## Frontend React Hooks The `@trigger.dev/react-hooks` package is the primary interface for frontend realtime integration. These hooks utilize SWR for state management and `AbortController` for clean connection termination. ### Hook Comparison Table | Hook | Purpose | Key Parameters | | :--- | :--- | :--- | | `useRealtimeRun` | Subscribe to a single run's status and metadata. | `runId`, `accessToken`, `onComplete` | | `useRealtimeRunWithStreams` | Subscribe to a run AND multiple associated data streams. | `runId`, `accessToken`, `throttleInMs` | | `useRealtimeRunsWithTag` | Monitor multiple runs sharing a specific tag. | `tag`, `createdAt` (filter duration) | | `useRealtimeBatch` | Subscribe to all runs within a specific batch. | `batchId`, `accessToken` | | `useRealtimeStream` | High-performance subscription to a specific named stream. | `runId`, `streamKey`, `startIndex` | Sources: [packages/react-hooks/src/hooks/useRealtime.ts:60-500](), [references/realtime-hooks-test/README.md:10-20]() ### Hook Implementation Logic The hooks use `processRealtime...` helper functions to iterate over asynchronous iterators returned by the API client. They implement throttling to prevent excessive UI re-renders during high-frequency updates, particularly for streams. ```typescript // Example of the inner loop logic in useRealtimeRun async function processRealtimeRun( runId: string, filters: { skipColumns?: RealtimeRunSkipColumns }, apiClient: ApiClient, mutateRunData: KeyedMutator>, onError: (e: Error) => void, abortControllerRef: React.MutableRefObject ) { const subscription = apiClient.subscribeToRun>(runId, { signal: abortControllerRef.current?.signal, onFetchError: onError, }); for await (const part of subscription) { mutateRunData(part); // Updates SWR state } } ``` Sources: [packages/react-hooks/src/hooks/useRealtime.ts:725-743]() ## Streaming (v2) Realtime Streams v2 (introduced in version 4.1.0) is designed for reliable, resumable data transmission. This is critical for AI applications where tokens are generated sequentially and network interruptions must not result in data loss. ### Resumability Features - **Resume from Index**: Streams can be started from a specific `startIndex`. - **Durable Streams**: Support for "Durable" scenarios where state is maintained across page refreshes. - **Reconnection Logic**: The system uses `X-Resume-From-Chunk` and `X-Last-Chunk-Index` headers to negotiate where a stream should resume after a failure. Sources: [rules/4.1.0/realtime.md:65-80](), [references/realtime-streams/TESTING.md:30-45]() ### Streaming Architecture Diagram ```mermaid flowchart TD Task[Task Execution] --> Pipe[aiStream.pipe] Pipe --> Storage[Trigger.dev Stream Storage] Storage --> SSE[Server-Sent Events / Stream API] SSE --> Client[useRealtimeStream Hook] Client -- Reconnect with index --> SSE SSE -- Resume from index --> Client ``` Sources: [rules/4.1.0/realtime.md:75-90](), [references/realtime-streams/TESTING.md:60-70]() ## Authentication & Security Realtime access is secured via Public Access Tokens. These tokens are created on the backend and passed to the frontend. ### Scopes and Permissions Tokens are scoped to restrict what a frontend client can see: - **read:runs**: Specific run IDs. - **read:tasks**: Specific task identifiers. - **read:tags**: Runs containing specific tags. ```typescript // Backend: Creating a scoped token const publicToken = await auth.createPublicToken({ scopes: { read: { runs: ["run_123"], tags: ["user_456"] }, }, expirationTime: "1h", }); ``` Sources: [rules/4.1.0/realtime.md:18-35](), [packages/react-hooks/CHANGELOG.md:120-130]() ## Best Practices 1. **Throttling**: For high-frequency streams (like AI tokens), use `throttleInMs` (default 16ms) to batch UI updates and maintain 60fps performance. 2. **Cleanup**: Always ensure subscriptions are stopped. React hooks handle this via `useEffect` cleanup calling `abortController.abort()`. 3. **onComplete Handling**: Use the `onComplete` callback for side effects like navigation or success notifications, rather than relying on status changes alone. 4. **Resuming**: In streaming scenarios, store the last received chunk index to allow for seamless reconnection if the user refreshes the page. Sources: [packages/react-hooks/src/hooks/useRealtime.ts:16-30](), [references/realtime-streams/TESTING.md:40-55](), [rules/4.1.0/realtime.md:160-170]() ## Summary The Realtime Subscriptions & Streaming system in Trigger.dev provides a robust, type-safe infrastructure for building live-updating applications. By combining SSE-based streaming with resumable v2 stream logic and specialized React hooks, the platform enables developers to treat background tasks as foreground experiences with minimal boilerplate. Sources: [README.md:30-35](), [rules/4.1.0/realtime.md:5-15]()