Add ALL_PROXY, WS_PROXY, WSS_PROXY support.

Author: rkistner

packages/node/package.json

@@ -53,8 +53,7 @@
     "async-lock": "^1.4.0",
     "bson": "^6.6.0",
     "comlink": "^4.4.2",
-    "proxy-agent": "^6.5.0",
-    "undici": "^7.8.0",
+    "undici": "^7.10.0",
     "ws": "^8.18.1"
   },
   "devDependencies": {

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

@@ -6,12 +6,11 @@ import { Dispatcher } from 'undici';
  */
 export class ErrorRecordingDispatcher extends Dispatcher {
   private targetDispatcher: Dispatcher;
-  private onError: (error: Error) => void;
+  public onError: ((error: Error) => void) | undefined;
 
-  constructor(targetDispatcher: Dispatcher, onError: (error: Error) => void) {
+  constructor(targetDispatcher: Dispatcher) {
     super();
     this.targetDispatcher = targetDispatcher;
-    this.onError = onError;
   }
 
   dispatch(opts: Dispatcher.DispatchOptions, handler: Dispatcher.DispatchHandler): boolean {
@@ -25,7 +24,7 @@ export class ErrorRecordingDispatcher extends Dispatcher {
       onResponseEnd: handler.onResponseEnd?.bind(handler),
 
       onResponseError: (controller: any, error: Error) => {
-        this.onError(error);
+        this.onError?.(error);
         // Pass through to original handler
         return handler.onResponseError?.(controller, error);
       },
@@ -38,7 +37,7 @@ export class ErrorRecordingDispatcher extends Dispatcher {
       onComplete: handler.onComplete?.bind(handler),
 
       onError: (error: Error) => {
-        this.onError(error);
+        this.onError?.(error);
 
         // Pass through to original handler
         return handler.onError?.(error);

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

@@ -11,7 +11,14 @@ import {
   RemoteConnector
 } from '@powersync/common';
 import { BSON } from 'bson';
-import { Dispatcher, EnvHttpProxyAgent, ErrorEvent, WebSocket as UndiciWebSocket } from 'undici';
+import {
+  Dispatcher,
+  EnvHttpProxyAgent,
+  ErrorEvent,
+  getGlobalDispatcher,
+  ProxyAgent,
+  WebSocket as UndiciWebSocket
+} from 'undici';
 import { ErrorRecordingDispatcher } from './ErrorRecordingDispatcher.js';
 
 export const STREAMING_POST_TIMEOUT_MS = 30_000;
@@ -35,51 +42,64 @@ export type NodeCustomConnectionOptions = {
 export type NodeRemoteOptions = AbstractRemoteOptions & NodeCustomConnectionOptions;
 
 export class NodeRemote extends AbstractRemote {
-  private dispatcher: Dispatcher;
+  private wsDispatcher: Dispatcher | undefined;
 
   constructor(
     protected connector: RemoteConnector,
     protected logger: ILogger = DEFAULT_REMOTE_LOGGER,
     options?: Partial<NodeRemoteOptions>
   ) {
-    // EnvHttpProxyAgent automatically uses relevant env vars for HTTP
-    const dispatcher = options?.dispatcher ?? new EnvHttpProxyAgent();
+    const fetchDispatcher = options?.dispatcher ?? defaultFetchDispatcher();
 
     super(connector, logger, {
       fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider(),
       fetchOptions: {
-        dispatcher
+        dispatcher: fetchDispatcher
       },
       ...(options ?? {})
     });
 
-    this.dispatcher = dispatcher;
+    this.wsDispatcher = options?.dispatcher;
   }
 
   protected createSocket(url: string): globalThis.WebSocket {
     // Create dedicated dispatcher for this WebSocket
-    let ws: UndiciWebSocket | undefined;
-    const onError = (error: Error) => {
+    const baseDispatcher = this.getWebsocketDispatcher(url);
+    const errorRecordingDispatcher = new ErrorRecordingDispatcher(baseDispatcher);
+
+    // Create WebSocket with dedicated dispatcher
+    const ws = new UndiciWebSocket(url, {
+      dispatcher: errorRecordingDispatcher,
+      headers: {
+        'User-Agent': this.getUserAgent()
+      }
+    });
+
+    errorRecordingDispatcher.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);
+      ws.dispatchEvent(event);
     };
 
-    const errorRecordingDispatcher = new ErrorRecordingDispatcher(this.dispatcher, onError);
+    return ws as globalThis.WebSocket;
+  }
 
-    // Create WebSocket with dedicated dispatcher
-    ws = new UndiciWebSocket(url, {
-      dispatcher: errorRecordingDispatcher,
-      headers: {
-        'User-Agent': this.getUserAgent()
-      }
-    });
+  protected getWebsocketDispatcher(url: string) {
+    if (this.wsDispatcher != null) {
+      return this.wsDispatcher;
+    }
 
-    return ws as globalThis.WebSocket;
+    const protocol = new URL(url).protocol.replace(':', '');
+    const proxy = getProxyForProtocol(protocol);
+    if (proxy != null) {
+      return new ProxyAgent(proxy);
+    } else {
+      return getGlobalDispatcher();
+    }
   }
 
   getUserAgent(): string {
@@ -95,3 +115,21 @@ export class NodeRemote extends AbstractRemote {
     return BSON;
   }
 }
+
+function defaultFetchDispatcher(): Dispatcher {
+  // EnvHttpProxyAgent automatically uses HTTP_PROXY, HTTPS_PROXY and NO_PROXY env vars by default.
+  // We add ALL_PROXY support.
+  return new EnvHttpProxyAgent({
+    httpProxy: getProxyForProtocol('http'),
+    httpsProxy: getProxyForProtocol('https')
+  });
+}
+
+function getProxyForProtocol(protocol: string): string | undefined {
+  return (
+    process.env[`${protocol.toLowerCase()}_proxy`] ??
+    process.env[`${protocol.toUpperCase()}_PROXY`] ??
+    process.env[`all_proxy`] ??
+    process.env[`ALL_PROXY`]
+  );
+}