wip: proxy agents

Author: stevensJourney

demos/example-node/src/main.ts

@@ -1,10 +1,10 @@
-import repl_factory from 'node:repl';
 import { once } from 'node:events';
+import repl_factory from 'node:repl';
 
 import { PowerSyncDatabase, SyncStreamConnectionMethod } from '@powersync/node';
 import { default as Logger } from 'js-logger';
-import { AppSchema, DemoConnector } from './powersync.js';
 import { exit } from 'node:process';
+import { AppSchema, DemoConnector } from './powersync.js';
 
 const main = async () => {
   const logger = Logger.get('PowerSyncDemo');

packages/common/src/client/sync/stream/AbstractRemote.ts

@@ -85,14 +85,18 @@ export type AbstractRemoteOptions = {
    * Binding should be done before passing here.
    */
   fetchImplementation: FetchImplementation | FetchImplementationProvider;
+
+  // TODO
+  fetchOptions?: {};
 };
 
 export const DEFAULT_REMOTE_OPTIONS: AbstractRemoteOptions = {
   socketUrlTransformer: (url) =>
     url.replace(/^https?:\/\//, function (match) {
       return match === 'https://' ? 'wss://' : 'ws://';
     }),
-  fetchImplementation: new FetchImplementationProvider()
+  fetchImplementation: new FetchImplementationProvider(),
+  fetchOptions: {}
 };
 
 export abstract class AbstractRemote {
@@ -231,6 +235,10 @@ export abstract class AbstractRemote {
    */
   abstract getBSON(): Promise<BSONImplementation>;
 
+  protected createSocket(url: string): WebSocket {
+    return new WebSocket(url);
+  }
+
   /**
    * Connects to the sync/stream websocket endpoint
    */
@@ -249,7 +257,8 @@ export abstract class AbstractRemote {
 
     const connector = new RSocketConnector({
       transport: new WebsocketClientTransport({
-        url: this.options.socketUrlTransformer(request.url)
+        url: this.options.socketUrlTransformer(request.url),
+        wsCreator: (url) => this.createSocket(url)
       }),
       setup: {
         keepAlive: KEEP_ALIVE_MS,
@@ -421,6 +430,7 @@ export abstract class AbstractRemote {
       body: JSON.stringify(data),
       signal: controller.signal,
       cache: 'no-store',
+      ...(this.options.fetchOptions ?? {}),
       ...options.fetchOptions
     }).catch((ex) => {
       if (ex.name == 'AbortError') {

packages/node/package.json

@@ -51,12 +51,13 @@
     "@powersync/common": "workspace:*",
     "async-lock": "^1.4.0",
     "bson": "^6.6.0",
-    "comlink": "^4.4.2"
+    "comlink": "^4.4.2",
+    "undici": "^7.8.0"
   },
   "devDependencies": {
+    "@powersync/drizzle-driver": "workspace:*",
     "@types/async-lock": "^1.4.0",
     "drizzle-orm": "^0.35.2",
-    "@powersync/drizzle-driver": "workspace:*",
     "rollup": "4.14.3",
     "typescript": "^5.5.3",
     "vitest": "^3.0.5"

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

@@ -12,6 +12,7 @@ import {
   RemoteConnector
 } from '@powersync/common';
 import { BSON } from 'bson';
+import { ProxyAgent } from 'undici';
 
 export const STREAMING_POST_TIMEOUT_MS = 30_000;
 
@@ -27,12 +28,27 @@ export class NodeRemote extends AbstractRemote {
     protected logger: ILogger = DEFAULT_REMOTE_LOGGER,
     options?: Partial<AbstractRemoteOptions>
   ) {
+    // Automatic env vars are not supported by undici
+    // proxy-agent does not work directly with dispatcher
+    const proxy = process.env.HTTPS_PROXY ?? process.env.HTTP_PROXY;
     super(connector, logger, {
       ...(options ?? {}),
-      fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider()
+      fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider(),
+      fetchOptions: {
+        dispatcher: proxy ? new ProxyAgent(proxy) : undefined
+      }
     });
   }
 
+  // protected createSocket(url: string): globalThis.WebSocket {
+  //   return new WebSocket(url, {
+  //     // agent: new ProxyAgent(),
+  //     headers: {
+  //       'User-Agent': this.getUserAgent()
+  //     }
+  //   }) as any as globalThis.WebSocket;
+  // }
+
   getUserAgent(): string {
     return [
       super.getUserAgent(),