feat(data-connect): add support for streaming transport

[stephenarosaj]

Streaming Transport (WebSockets)

Status

The feature branch is ready for review and merging.

Overview

✨ This PR introduces full-duplex streaming transport capability to Firebase Data Connect via WebSockets. It introduces a TransportManager to intelligently route requests between standard REST and a persistent WebSocket stream, implements a stateful Stream Transport with request tracking and idle timeouts, and fully integrates these changes into the query layer to support query subscriptions.

We've opened, reviewed, and merged a bunch of smaller PRs from sub-branches (pasta/spaghetti, pasta/lasagna, etc.) into this pasta/main branch. We've also been pulling from main into pasta/main as we go.

See go/fdc-sdk-streaming-design for more information.

Changes

1. Transport Manager & Routing

• DataConnectTransportManager (manager.ts):
  • Created DataConnectTransportManager to act as the primary entry point for network operations, routing between RESTTransport and WebSocketTransport, using the stream when available and falling back to REST if the stream is closed or encounters a fatal error.
  • Implemented lazy initialization of the stream transport, opening the connection only when a subscription is requested.
  • Added routing logic to fall back to REST if the stream is unhealthy, pending closure, or unavailable.
• Refactored existing network source code
  • Moved existing REST files to a flattened structure under network/rest/ to make room for the streaming implementation.
  • Created AbstractDataConnectTransport to centralize token management (Auth and App Check) and emulator configuration, reducing duplication across transports.
  • URL Utilities (url.ts):
  • Added websocketUrlBuilder for constructing the stream endpoint targeting the ConnectorStreamService.

2. Abstract Stream Transport

• AbstractDataConnectStreamTransport (streamTransport.ts):
  • Handles request tracking (mapping RequestIds to execution promises or subscription hooks), de-duplication of queries, and routing incoming server messages.
  • Added tracking maps for active queries, mutations, and subscriptions to correlate outgoing requests with incoming responses via requestId.
  • Implemented prepareMessage logic to attach stateful headers (tokens) and resource names efficiently, ensuring tokens are only sent when changed and resource names are only sent on the first message.
  • Added a 60-second idle timeout to close the connection gracefully when no active subscriptions exist.
  • Handled auth state changes to disconnect the stream on user logout, login, or user change.
• Wire Protocol (wire.ts):
  • Defined TypeScript interfaces for DataConnectStreamRequest and DataConnectStreamResponse.
  • Used mapped types to enforce mutual exclusivity of payload kinds (execute, subscribe, resume, cancel).

3. Concrete (WebSocket) Stream Transport

• WebSocket Transport (websocket.ts):
  • Implemented WebSocketTransport extending the abstract stream transport.
  • Handled browser WebSocket API interactions and binary decoding (using TextDecoder) for production responses.
  • Implemented defensive truncation for close reason strings to comply with protocol limits.

4. Query Layer Integration

• Query Manager (QueryManager.ts):
  • Updated QueryManager to use the new DataConnectTransportInterface.
  • Wired subscribe and unsubscribe to trigger transport-level stream operations when callbacks are added or removed.
  • Added SubscribeObserver interfaces and methods to handle incoming stream data and error notifications, updating the cache and notifying listeners accordingly.

Testing

Extensive unit tests were added to cover the new transport architecture:

• test/unit/streamTransport.test.ts: Verified stateful header logic, tracking maps, and idle timeout behavior.
• test/unit/streaming.test.ts: Verified streaming queries and cache integration.
• test/unit/transportManager.test.ts: Verified routing logic and fallback behavior.
• test/unit/websocketTransport.test.ts: Verified WebSocket lifecycle by mocking the global WebSocket object.

Description of tests added (based on the `describe()` and `it()` trees inside each test file)

streaming.test.ts

• Streaming & Query Layer Integration
  • using stream via user-facing API
    • executeQuery / executeMutation should not initialize stream
    • subscribe should initialize stream
    • query layer should call invokeUnsubscribe on last unsubscribe
    • executeQuery should use stream when stream is active
    • executeMutation should use stream when stream is active
  • incoming notifications
    • should notify all relevant subscribers when data is received
    • should update cache when data is pushed
    • should notify all relevant subscribers of errors when they are pushed
    • should clean up subscriptions in query layer when observer receives a disconnect error
    • should NOT clean up subscriptions in query layer when observer receives a non-disconnect error

websocketTransport.test.ts

• WebSocketTransport
  • openConnection
    • should resolve when onopen fires
    • should reject when onerror fires
    • should de-duplicate calls when already connected
    • should de-duplicate calls when open connection is pending
  • streamIsReady
    • should be false initially
    • should be false while connecting
    • should be true when connected
    • should be false after disconnected
  • closeConnection
    • should close the underlying websocket and reset connection states
    • should be idempotent if connection is already closed
    • should return rejected promise and clean up if ws.close() throws
  • onclose handler (handleWebsocketDisconnect)
    • should reset connection state when websocket is closed externally
    • should call rejectAllActiveRequests and clean up when closed externally
  • sendMessage
    • should not send until connection is open
    • should send stringified json payload
  • handleWebSocketMessage
    • should correctly parse incoming JSON and handle response
    • should map extensions to empty array if not strictly provided
    • should close connection with error if message is not an object
    • should close connection with error if result is missing
    • should close connection with error if result is not an object
    • should close connection with error if requestId is missing

transportManager.test.ts

• DataConnectTransportManager
  • stream transport initialization
    • initStreamTransport should initialize stream transport only once
  • default transport routing
    • stream transport should not be initialized by default
    • invokeQuery should route to REST by default and not initialize stream transport
    • invokeMutation should route to REST by default and not initialize stream transport
    • invokeSubscribe should route to streaming by default and initialize stream transport
    • invokeUnsubscribe should route to streaming by default when stream transport is already initialized and not initialize stream transport
    • useEmulator should route to REST and streaming
    • onAuthTokenChanged should route to REST and streaming
  • dynamic transport routing
    • executeShouldUseStream should return false if streamTransport is not initialized
    • executeShouldUseStream should return true only when all stream conditions are met
    • invokeQuery dynamic routing
      • invokeQuery should route to stream if executeShouldUseStream returns true
      • invokeQuery should throw an error if stream transport throws an error and executeShouldUseStream remains true
      • invokeQuery should fallback to REST if stream transport throws an error and then executeShouldUseStream becomes false
    • invokeMutation dynamic routing
      • invokeMutation should route to stream if executeShouldUseStream() returns true
      • invokeMutation should throw an error if stream transport throws an error and executeShouldUseStream remains true
      • invokeMutation should fallback to REST if stream transport throws an error and then executeShouldUseStream becomes false
      • invokeMutation should fallback to REST for all in-flight mutations if stream fails
    • invokeSubscribe dynamic routing
      • invokeSubscribe should throw an error if isUnableToConnect is true
  • disconnects
    • subscriber onDisconnect should be called when stream fails during active subscription
    • idle timeout
      • should route to REST during idle timeout and disconnect after 60s
      • should route to REST after stream automatically closes
      • should route back to stream after reconnect

streamTransport.test.ts

• AbstractDataConnectStreamTransport
  • prepareMessage
    • should not change data fields
    • should handle headers properly
      • auth token
        • should add auth token to the first message
        • should NOT add the same auth token to subsequent messages
        • should include auth token when it changes
      • app check token
        • should add app check token to the first message
        • should NOT add the same app check token to subsequent messages
        • should NOT include app check token when it changes
      • x-firebase-gmpid
        • should add x-firebase-gmpid to every message if appId is set
      • should add X-Goog-Api-Client to every message based on caller sdk type
    • should handle name properly
      • should add name to the first message
      • should NOT add name to subsequent messages
    • should reset connection state on onConnectionReady()
  • Request Tracking
    • getMapKey should sort keys consistently for map lookups
    • should generate unique request IDs
    • Incoming Requests from Transport Layer
      • invokeQuery
        • should populate tracking maps synchronously and then call sendMessage
        • should asynchronously clean up and reject if sendMessage fails
      • invokeMutation
        • should populate tracking maps synchronously and then call sendMessage
        • should asynchronously clean up and reject if sendMessage fails
      • invokeSubscribe
        • should populate tracking maps synchronously and then call sendMessage
        • should asynchronously call observer with error and clean up if sendMessage fails
      • invokeUnsubscribe
        • should de-populate tracking maps and call sendMessage
        • should asynchronously clean up and log error if sendMessage fails
    • Incoming Responses from Server
      • should throw an error if an unrecognized requestId is received
      • invokeQuery tracking
        • should route data to resolve the correct query promise when response is received
        • should clean map when response is received
        • should reject the correct query promise with DataConnectOperationError if response has errors
        • should clean map correctly when handleResponse rejects
      • invokeMutation tracking
        • should route data to resolve the correct mutation promise when response is received
        • should clean map of the correct tracked request when response is received
        • should reject the correct mutation promise with DataConnectOperationError if response has errors
        • should clean map correctly when handleResponse rejects
      • invokeSubscribe tracking
        • should route data to the correct subscribe observer whenever a response is received
        • should route error response to the correct subscribe observer whenever an error response is received
        • should NOT clean map when handleResponse rejects
  • Disconnects
    • should close connection after 60 seconds of idle (no active subscriptions)
    • should cancel close if a new subscription arrives during timeout
    • should restart close timeout if subscribe fails and leaves no active subscriptions
    • should not close connection if there are active execute requests
    • should close connection when last execute request finishes after idle timeout
    • Auth Disconnects
      • should close stream immediately on illegal auth change (login)
      • should close stream immediately on illegal auth change (logout)
      • should close stream immediately on illegal auth change (user change)
      • should NOT close stream on valid auth token refresh (same user)

[changeset-bot]

🦋 Changeset detected

Latest commit: 9042e51

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 2 packages

Name	Type
firebase	Minor
@firebase/data-connect	Minor

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[gemini-code-assist]

Code Review

This pull request adds a new transport layer to the Data Connect SDK, introducing a DataConnectTransportManager that manages routing between REST and a new WebSocket-based streaming transport. This architecture supports subscriptions and push notifications. Key implementation details include the AbstractDataConnectStreamTransport for request management and WebSocketTransport for connection handling. Review feedback highlights a hardcoded health check that disables the REST fallback, potential race conditions in query tracking maps, and several TypeScript type-safety issues where error arrays are incorrectly typed as empty tuples. Suggestions were also provided to improve variable initialization and error handling consistency.

[hsubox76]

Make sure to add a changeset for data-connect (minor) - and don't forget to manually add "firebase": minor as well.

[hsubox76]

Websockets are stable in Node only as of Node 22. Officially the JS SDK supports Node 20+ (we will likely bump to 22 at the next breaking change release, maybe July). Will there be docs telling users to use a polyfill or make sure they're using Node 22+? I think there should at least be an engines field in the package.json reflecting the minimum required Node version. See https://github.com/firebase/firebase-js-sdk/blob/pasta/main/packages/ai/package.json#L7

[stephenarosaj]

I see Node 21 is the first to add WebSockets support, but it's experimental, while Node 22 enables it by default, like you mentioned, and it looks like Node 20 is going to enter End of Life mode on 4/30/26

There's currently no mention in the documentation of WebSocket support in Node - but we can add it. Just to be clear, we should be asking users to either use Node 20/21 with a polyfill, Node 21 with the --experimental-websockets flag, or Node 22+, and allowing users to call initializeWebSocket() themselves until the next major release which updates the Node dependency to 22+, right?

[stephenarosaj]

@hsubox76 after discussing with our team, we don't anticipate usage of streaming / WebSockets in Node environments. While it's technically possible that users could decide to use our client SDK in a server environment, typically we see this happen in SSR use cases - and we don't expect initial loads of the page / data to come from subscribe / WebSocket calls - we'd like to omit this from the documentation for simplicity unless people start opening requests complaining about it.

If users really want to use streaming / WebSockets in a Node environment, they can add one to the globalThis object - and I did go ahead and add a warning to the initializeWebsocket() check in index.node.ts, which is part of this PR: 'WebSocket is not available in this environment. Use a polyfill or upgrade your Node version to one that supports WebSockets.'

WDYT?

[hsubox76]

That works, warning message makes it clear enough, thanks.

[stephenarosaj]

The final PR associated with this work has been merged into the pasta/main branch.