Use Undici WebSocket with Dispatcher support.

Author: rkistner

packages/node/src/db/PowerSyncDatabase.ts

@@ -13,10 +13,9 @@ import {
   SQLOpenFactory
 } from '@powersync/common';
 
-import { NodeRemote } from '../sync/stream/NodeRemote.js';
+import { NodeCustomConnectionOptions, NodeRemote } from '../sync/stream/NodeRemote.js';
 import { NodeStreamingSyncImplementation } from '../sync/stream/NodeStreamingSyncImplementation.js';
 
-import { Dispatcher } from 'undici';
 import { BetterSQLite3DBAdapter } from './BetterSQLite3DBAdapter.js';
 import { NodeSQLOpenOptions } from './options.js';
 
@@ -30,13 +29,7 @@ export type NodePowerSyncDatabaseOptions = PowerSyncDatabaseOptions & {
   remoteOptions?: Partial<AbstractRemoteOptions>;
 };
 
-export type NodeAdditionalConnectionOptions = AdditionalConnectionOptions & {
-  /**
-   * Optional custom dispatcher for HTTP connections (e.g. using undici).
-   * Only used when the connection method is SyncStreamConnectionMethod.HTTP
-   */
-  dispatcher?: Dispatcher;
-};
+export type NodeAdditionalConnectionOptions = AdditionalConnectionOptions & NodeCustomConnectionOptions;
 
 export type NodePowerSyncConnectionOptions = PowerSyncConnectionOptions & NodeAdditionalConnectionOptions;
 
@@ -76,7 +69,7 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
 
   connect(
     connector: PowerSyncBackendConnector,
-    options?: PowerSyncConnectionOptions & { dispatcher?: Dispatcher }
+    options?: PowerSyncConnectionOptions & NodeCustomConnectionOptions
   ): Promise<void> {
     return super.connect(connector, options);
   }

packages/node/src/sync/stream/ErrorRecordingDispatcher.ts

@@ -0,0 +1,59 @@
+import { Dispatcher } from 'undici';
+
+/**
+ * A simple dispatcher wrapper that only records the last error.
+ * Everything else passes straight through to the original handler.
+ */
+export class ErrorRecordingDispatcher extends Dispatcher {
+  private targetDispatcher: Dispatcher;
+  private onError: (error: Error) => void;
+
+  constructor(targetDispatcher: Dispatcher, onError: (error: Error) => void) {
+    super();
+    this.targetDispatcher = targetDispatcher;
+    this.onError = onError;
+  }
+
+  dispatch(opts: Dispatcher.DispatchOptions, handler: Dispatcher.DispatchHandler): boolean {
+    // Create a simple wrapper that only intercepts errors
+    const errorRecordingHandler: Dispatcher.DispatchHandler = {
+      // New API methods (preferred)
+      onRequestStart: handler.onRequestStart?.bind(handler),
+      onRequestUpgrade: handler.onRequestUpgrade?.bind(handler),
+      onResponseStart: handler.onResponseStart?.bind(handler),
+      onResponseData: handler.onResponseData?.bind(handler),
+      onResponseEnd: handler.onResponseEnd?.bind(handler),
+
+      onResponseError: (controller: any, error: Error) => {
+        this.onError(error);
+        // Pass through to original handler
+        return handler.onResponseError?.(controller, error);
+      },
+
+      // Legacy API methods (for backward compatibility)
+      onConnect: handler.onConnect?.bind(handler),
+      onUpgrade: handler.onUpgrade?.bind(handler),
+      onHeaders: handler.onHeaders?.bind(handler),
+      onData: handler.onData?.bind(handler),
+      onComplete: handler.onComplete?.bind(handler),
+
+      onError: (error: Error) => {
+        this.onError(error);
+
+        // Pass through to original handler
+        return handler.onError?.(error);
+      }
+    };
+
+    // Delegate to the target dispatcher with our simple error-recording handler
+    return this.targetDispatcher.dispatch(opts, errorRecordingHandler);
+  }
+
+  async close(): Promise<void> {
+    return this.targetDispatcher.close();
+  }
+
+  async destroy(): Promise<void> {
+    return this.targetDispatcher.destroy();
+  }
+}

packages/node/src/sync/stream/NodeRemote.ts

@@ -11,9 +11,8 @@ import {
   RemoteConnector
 } from '@powersync/common';
 import { BSON } from 'bson';
-import Agent from 'proxy-agent';
-import { EnvHttpProxyAgent, Dispatcher } from 'undici';
-import { WebSocket } from 'ws';
+import { Dispatcher, EnvHttpProxyAgent, ErrorEvent, WebSocket as UndiciWebSocket } from 'undici';
+import { ErrorRecordingDispatcher } from './ErrorRecordingDispatcher.js';
 
 export const STREAMING_POST_TIMEOUT_MS = 30_000;
 
@@ -23,11 +22,21 @@ class NodeFetchProvider extends FetchImplementationProvider {
   }
 }
 
-export type NodeRemoteOptions = AbstractRemoteOptions & {
+export type NodeCustomConnectionOptions = {
+  /**
+   * Optional custom dispatcher for HTTP or WEB_SOCKET connections.
+   *
+   * This can be used to customize proxy usage (using undici ProxyAgent),
+   * or other connection options.
+   */
   dispatcher?: Dispatcher;
 };
 
+export type NodeRemoteOptions = AbstractRemoteOptions & NodeCustomConnectionOptions;
+
 export class NodeRemote extends AbstractRemote {
+  private dispatcher: Dispatcher;
+
   constructor(
     protected connector: RemoteConnector,
     protected logger: ILogger = DEFAULT_REMOTE_LOGGER,
@@ -43,16 +52,34 @@ export class NodeRemote extends AbstractRemote {
       },
       ...(options ?? {})
     });
+
+    this.dispatcher = dispatcher;
   }
 
   protected createSocket(url: string): globalThis.WebSocket {
-    return new WebSocket(url, {
-      // Automatically uses relevant env vars for web sockets
-      agent: new Agent.ProxyAgent(),
+    // Create dedicated dispatcher for this WebSocket
+    let ws: UndiciWebSocket | undefined;
+    const onError = (error: Error) => {
+      // When we receive an error from the Dispatcher, emit the event on the websocket.
+      // This will take precedence over the WebSocket's own error event, giving more details on what went wrong.
+      const event = new ErrorEvent('error', {
+        error,
+        message: error.message
+      });
+      ws?.dispatchEvent(event);
+    };
+
+    const errorRecordingDispatcher = new ErrorRecordingDispatcher(this.dispatcher, onError);
+
+    // Create WebSocket with dedicated dispatcher
+    ws = new UndiciWebSocket(url, {
+      dispatcher: errorRecordingDispatcher,
       headers: {
         'User-Agent': this.getUserAgent()
       }
-    }) as any as globalThis.WebSocket; // This is compatible in Node environments
+    });
+
+    return ws as globalThis.WebSocket;
   }
 
   getUserAgent(): string {