Merge branch 'main' into feature/logging-cleanup-demos

Author: Chriztiaan

.changeset/hip-lamps-draw.md

@@ -0,0 +1,5 @@
+---
+'@powersync/node': minor
+---
+
+Introduced support for specifying proxy environment variables for the connection methods. For HTTP it supports `HTTP_PROXY` or `HTTPS_PROXY`, and for WebSockets it supports `WS_PROXY` and `WSS_PROXY`.

.changeset/thick-lies-invent.md

@@ -0,0 +1,6 @@
+---
+'@powersync/common': minor
+---
+
+Added `fetchOptions` to AbstractRemoteOptions. Allows consumers to include fields such as `dispatcher` (e.g. for proxy support) to the fetch invocations.
+Also ensuring all options provided to `connect()` are passed onwards, allows packages to have their own option definitions for `connect()` and the abstract `generateSyncStreamImplementation()`.

demos/example-node/src/main.ts

@@ -1,9 +1,9 @@
-import repl_factory from 'node:repl';
 import { once } from 'node:events';
+import repl_factory from 'node:repl';
 
 import { createBaseLogger, createLogger, PowerSyncDatabase, SyncStreamConnectionMethod } from '@powersync/node';
-import { AppSchema, DemoConnector } from './powersync.js';
 import { exit } from 'node:process';
+import { AppSchema, DemoConnector } from './powersync.js';
 
 const main = async () => {
   const baseLogger = createBaseLogger();

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -421,6 +421,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   // Use the options passed in during connect, or fallback to the options set during database creation or fallback to the default options
   resolvedConnectionOptions(options?: PowerSyncConnectionOptions): RequiredAdditionalConnectionOptions {
     return {
+      ...options,
       retryDelayMs:
         options?.retryDelayMs ?? this.options.retryDelayMs ?? this.options.retryDelay ?? DEFAULT_RETRY_DELAY_MS,
       crudUploadThrottleMs:
@@ -440,12 +441,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       throw new Error('Cannot connect using a closed client');
     }
 
-    const { retryDelayMs, crudUploadThrottleMs } = this.resolvedConnectionOptions(options);
+    const resolvedConnectOptions = this.resolvedConnectionOptions(options);
 
-    this.syncStreamImplementation = this.generateSyncStreamImplementation(connector, {
-      retryDelayMs,
-      crudUploadThrottleMs
-    });
+    this.syncStreamImplementation = this.generateSyncStreamImplementation(connector, resolvedConnectOptions);
     this.syncStatusListenerDisposer = this.syncStreamImplementation.registerListener({
       statusChanged: (status) => {
         this.currentStatus = new SyncStatus({

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

@@ -85,14 +85,23 @@ export type AbstractRemoteOptions = {
    * Binding should be done before passing here.
    */
   fetchImplementation: FetchImplementation | FetchImplementationProvider;
+
+  /**
+   * Optional options to pass directly to all `fetch` calls.
+   *
+   * This can include fields such as `dispatcher` (e.g. for proxy support),
+   * `cache`, or any other fetch-compatible options.
+   */
+  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 +240,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 +262,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 +435,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/README.md

@@ -56,6 +56,22 @@ contains everything you need to know to get started implementing PowerSync in yo
 
 A simple example using `@powersync/node` is available in the [`demos/example-node/`](../demos/example-node) directory.
 
+# Proxy Support
+
+This SDK supports HTTP, HTTPS, and WebSocket proxies via environment variables.
+
+## HTTP Connection Method
+
+Internally we probe the http environment variables and apply it to fetch requests ([undici](https://www.npmjs.com/package/undici/v/5.6.0))
+
+- Set the `HTTPS_PROXY` or `HTTP_PROXY` environment variable to automatically route HTTP requests through a proxy.
+
+## WEB Socket Connection Method
+
+Internally the [proxy-agent](https://www.npmjs.com/package/proxy-agent) dependency for WebSocket proxies, which has its own internal code for automatically picking up the appropriate environment variables:
+
+- Set the `WS_PROXY` or `WSS_PROXY` environment variable to route the webocket connections through a proxy.
+
 # Found a bug or need help?
 
 - Join our [Discord server](https://discord.gg/powersync) where you can browse topics from our community, ask questions, share feedback, or just say hello :)

packages/node/package.json

@@ -51,12 +51,15 @@
     "@powersync/common": "workspace:*",
     "async-lock": "^1.4.0",
     "bson": "^6.6.0",
-    "comlink": "^4.4.2"
+    "comlink": "^4.4.2",
+    "proxy-agent": "^6.5.0",
+    "undici": "^7.8.0",
+    "ws": "^8.18.1"
   },
   "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/db/PowerSyncDatabase.ts

@@ -1,11 +1,14 @@
 import {
   AbstractPowerSyncDatabase,
   AbstractStreamingSyncImplementation,
+  AdditionalConnectionOptions,
   BucketStorageAdapter,
   DBAdapter,
   PowerSyncBackendConnector,
+  PowerSyncConnectionOptions,
   PowerSyncDatabaseOptions,
   PowerSyncDatabaseOptionsWithSettings,
+  RequiredAdditionalConnectionOptions,
   SqliteBucketStorage,
   SQLOpenFactory
 } from '@powersync/common';
@@ -15,11 +18,22 @@ import { NodeStreamingSyncImplementation } from '../sync/stream/NodeStreamingSyn
 
 import { BetterSQLite3DBAdapter } from './BetterSQLite3DBAdapter.js';
 import { NodeSQLOpenOptions } from './options.js';
+import { Dispatcher } from 'undici';
 
 export type NodePowerSyncDatabaseOptions = PowerSyncDatabaseOptions & {
   database: DBAdapter | SQLOpenFactory | NodeSQLOpenOptions;
 };
 
+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 NodePowerSyncConnectionOptions = PowerSyncConnectionOptions & NodeAdditionalConnectionOptions;
+
 /**
  * A PowerSync database which provides SQLite functionality
  * which is automatically synced.
@@ -54,10 +68,18 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
     return new SqliteBucketStorage(this.database, AbstractPowerSyncDatabase.transactionMutex);
   }
 
+  connect(
+    connector: PowerSyncBackendConnector,
+    options?: PowerSyncConnectionOptions & { dispatcher?: Dispatcher }
+  ): Promise<void> {
+    return super.connect(connector, options);
+  }
+
   protected generateSyncStreamImplementation(
-    connector: PowerSyncBackendConnector
+    connector: PowerSyncBackendConnector,
+    options: NodeAdditionalConnectionOptions
   ): AbstractStreamingSyncImplementation {
-    const remote = new NodeRemote(connector);
+    const remote = new NodeRemote(connector, this.options.logger, { dispatcher: options.dispatcher });
 
     return new NodeStreamingSyncImplementation({
       adapter: this.bucketStorageAdapter,

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

@@ -11,6 +11,9 @@ import {
   RemoteConnector
 } from '@powersync/common';
 import { BSON } from 'bson';
+import Agent from 'proxy-agent';
+import { EnvHttpProxyAgent, Dispatcher } from 'undici';
+import { WebSocket } from 'ws';
 
 export const STREAMING_POST_TIMEOUT_MS = 30_000;
 
@@ -20,18 +23,38 @@ class NodeFetchProvider extends FetchImplementationProvider {
   }
 }
 
+export type NodeRemoteOptions = AbstractRemoteOptions & {
+  dispatcher?: Dispatcher;
+};
+
 export class NodeRemote extends AbstractRemote {
   constructor(
     protected connector: RemoteConnector,
     protected logger: ILogger = DEFAULT_REMOTE_LOGGER,
-    options?: Partial<AbstractRemoteOptions>
+    options?: Partial<NodeRemoteOptions>
   ) {
+    // EnvHttpProxyAgent automatically uses relevant env vars for HTTP
+    const dispatcher = options?.dispatcher ?? new EnvHttpProxyAgent();
+
     super(connector, logger, {
       ...(options ?? {}),
-      fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider()
+      fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider(),
+      fetchOptions: {
+        dispatcher
+      }
     });
   }
 
+  protected createSocket(url: string): globalThis.WebSocket {
+    return new WebSocket(url, {
+      // Automatically uses relevant env vars for web sockets
+      agent: new Agent.ProxyAgent(),
+      headers: {
+        'User-Agent': this.getUserAgent()
+      }
+    }) as any as globalThis.WebSocket; // This is compatible in Node environments
+  }
+
   getUserAgent(): string {
     return [
       super.getUserAgent(),