Fix more tests

Author: simolus3

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -15,7 +15,11 @@ import { Schema } from '../db/schema/Schema.js';
 import { BaseObserver } from '../utils/BaseObserver.js';
 import { ControlledExecutor } from '../utils/ControlledExecutor.js';
 import { symbolAsyncIterator, throttleTrailing } from '../utils/async.js';
-import { ConnectionManager, CreateSyncImplementationOptions } from './ConnectionManager.js';
+import {
+  ConnectionManager,
+  CreateSyncImplementationOptions,
+  InternalSubscriptionAdapter
+} from './ConnectionManager.js';
 import { CustomQuery } from './CustomQuery.js';
 import { ArrayQueryDefinition, Query } from './Query.js';
 import { SQLOpenFactory, SQLOpenOptions, isDBAdapter, isSQLOpenFactory, isSQLOpenOptions } from './SQLOpenFactory.js';
@@ -183,6 +187,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   protected bucketStorageAdapter: BucketStorageAdapter;
   protected _isReadyPromise: Promise<void>;
   protected connectionManager: ConnectionManager;
+  private subscriptions: InternalSubscriptionAdapter;
 
   get syncStreamImplementation() {
     return this.connectionManager.syncStreamImplementation;
@@ -237,14 +242,16 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     this.runExclusiveMutex = new Mutex();
 
     // Start async init
-    this.connectionManager = new ConnectionManager({
-      firstStatusMatching: (predicate) => this.waitForStatus(predicate),
+    this.subscriptions = {
+      firstStatusMatching: (predicate, abort) => this.waitForStatus(predicate, abort),
       resolveOfflineSyncStatus: () => this.resolveOfflineSyncStatus(),
       rustSubscriptionsCommand: async (payload) => {
         await this.writeTransaction((tx) => {
           return tx.execute('select powersync_control(?,?)', ['subscriptions', JSON.stringify(payload)]);
         });
-      },
+      }
+    };
+    this.connectionManager = new ConnectionManager({
       createSyncImplementation: async (connector, options) => {
         await this.waitForReady();
         return this.runExclusive(async () => {
@@ -538,7 +545,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   }
 
   syncStream(name: string, params?: Record<string, any>): SyncStream {
-    return this.connectionManager.stream(name, params ?? null);
+    return this.connectionManager.stream(this.subscriptions, name, params ?? null);
   }
 
   /**

packages/common/src/client/ConnectionManager.ts

@@ -36,6 +36,12 @@ export interface CreateSyncImplementationOptions extends AdditionalConnectionOpt
   subscriptions: SubscribedStream[];
 }
 
+export interface InternalSubscriptionAdapter {
+  firstStatusMatching(predicate: (status: SyncStatus) => any, abort?: AbortSignal): Promise<void>;
+  resolveOfflineSyncStatus(): Promise<void>;
+  rustSubscriptionsCommand(payload: any): Promise<void>;
+}
+
 /**
  * @internal
  */
@@ -44,9 +50,7 @@ export interface ConnectionManagerOptions {
     connector: PowerSyncBackendConnector,
     options: CreateSyncImplementationOptions
   ): Promise<ConnectionManagerSyncImplementationResult>;
-  firstStatusMatching(predicate: (status: SyncStatus) => any): Promise<void>;
-  resolveOfflineSyncStatus(): Promise<void>;
-  rustSubscriptionsCommand(payload: any): Promise<void>;
+
   logger: ILogger;
 }
 
@@ -269,19 +273,19 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
     await disposer?.();
   }
 
-  stream(name: string, parameters: Record<string, any> | null): SyncStream {
+  stream(adapter: InternalSubscriptionAdapter, name: string, parameters: Record<string, any> | null): SyncStream {
     const desc = { name, parameters } satisfies SyncStreamDescription;
 
-    const waitForFirstSync = () => {
-      return this.options.firstStatusMatching((s) => s.statusFor(desc)?.subscription.hasSynced ?? false);
+    const waitForFirstSync = (abort?: AbortSignal) => {
+      return adapter.firstStatusMatching((s) => s.statusFor(desc)?.subscription.hasSynced, abort);
     };
 
     return {
       ...desc,
       subscribe: async (options?: SyncStreamSubscribeOptions) => {
         // NOTE: We also run this command if a subscription already exists, because this increases the expiry date
         // (relevant if the app is closed before connecting again, where the last subscribe call determines the ttl).
-        await this.options.rustSubscriptionsCommand({
+        await adapter.rustSubscriptionsCommand({
           subscribe: {
             stream: {
               name,
@@ -295,7 +299,7 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
         if (!this.syncStreamImplementation) {
           // We're not connected. So, update the offline sync status to reflect the new subscription.
           // (With an active iteration, the sync client would include it in its state).
-          await this.options.resolveOfflineSyncStatus();
+          await adapter.resolveOfflineSyncStatus();
         }
 
         const key = `${name}|${JSON.stringify(parameters)}`;
@@ -314,7 +318,7 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
         return new SyncStreamSubscriptionHandle(subscription);
       },
       unsubscribeAll: async () => {
-        await this.options.rustSubscriptionsCommand({ unsubscribe: { name, params: parameters } });
+        await adapter.rustSubscriptionsCommand({ unsubscribe: { name, params: parameters } });
         this.subscriptionsMayHaveChanged();
       }
     };
@@ -337,7 +341,7 @@ class ActiveSubscription {
   constructor(
     readonly name: string,
     readonly parameters: Record<string, any> | null,
-    readonly waitForFirstSync: () => Promise<void>,
+    readonly waitForFirstSync: (abort?: AbortSignal) => Promise<void>,
     private clearSubscription: () => void
   ) {}
 
@@ -362,8 +366,8 @@ class SyncStreamSubscriptionHandle implements SyncStreamSubscription {
     return this.subscription.parameters;
   }
 
-  waitForFirstSync(): Promise<void> {
-    return this.subscription.waitForFirstSync();
+  waitForFirstSync(abort?: AbortSignal): Promise<void> {
+    return this.subscription.waitForFirstSync(abort);
   }
 
   unsubscribe(): void {

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

@@ -95,7 +95,7 @@ export interface LockOptions<T> {
   signal?: AbortSignal;
 }
 
-export interface AbstractStreamingSyncImplementationOptions extends AdditionalConnectionOptions {
+export interface AbstractStreamingSyncImplementationOptions extends RequiredAdditionalConnectionOptions {
   adapter: BucketStorageAdapter;
   subscriptions: SubscribedStream[];
   uploadCrud: () => Promise<void>;
@@ -185,7 +185,9 @@ export interface AdditionalConnectionOptions {
 }
 
 /** @internal */
-export type RequiredAdditionalConnectionOptions = Required<AdditionalConnectionOptions>;
+export interface RequiredAdditionalConnectionOptions extends Required<AdditionalConnectionOptions> {
+  subscriptions: SubscribedStream[];
+}
 
 export interface StreamingSyncImplementation
   extends BaseObserverInterface<StreamingSyncImplementationListener>,
@@ -265,7 +267,7 @@ export abstract class AbstractStreamingSyncImplementation
 
   constructor(options: AbstractStreamingSyncImplementationOptions) {
     super();
-    this.options = { ...DEFAULT_STREAMING_SYNC_OPTIONS, ...options };
+    this.options = options;
     this.activeStreams = options.subscriptions;
     this.logger = options.logger ?? Logger.get('PowerSyncStream');
 

packages/common/src/client/sync/sync-streams.ts

@@ -98,7 +98,7 @@ export interface SyncStreamSubscription extends SyncStreamDescription {
   /**
    * A promise that resolves once data from in this sync stream has been synced and applied.
    */
-  waitForFirstSync(): Promise<void>;
+  waitForFirstSync(abort?: AbortSignal): Promise<void>;
 
   /**
    * Removes this stream subscription.

packages/node/src/db/PowerSyncDatabase.ts

@@ -4,12 +4,12 @@ import {
   AbstractStreamingSyncImplementation,
   AdditionalConnectionOptions,
   BucketStorageAdapter,
-  CreateSyncImplementationOptions,
   DBAdapter,
   PowerSyncBackendConnector,
   PowerSyncConnectionOptions,
   PowerSyncDatabaseOptions,
   PowerSyncDatabaseOptionsWithSettings,
+  RequiredAdditionalConnectionOptions,
   SqliteBucketStorage,
   SQLOpenFactory
 } from '@powersync/common';
@@ -77,7 +77,7 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
 
   protected generateSyncStreamImplementation(
     connector: PowerSyncBackendConnector,
-    options: CreateSyncImplementationOptions & NodeAdditionalConnectionOptions
+    options: RequiredAdditionalConnectionOptions & NodeAdditionalConnectionOptions
   ): AbstractStreamingSyncImplementation {
     const logger = this.logger;
     const remote = new NodeRemote(connector, logger, {
@@ -86,15 +86,13 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
     });
 
     return new NodeStreamingSyncImplementation({
-      subscriptions: options.subscriptions,
       adapter: this.bucketStorageAdapter,
       remote,
       uploadCrud: async () => {
         await this.waitForReady();
         await connector.uploadData(this);
       },
-      retryDelayMs: this.options.retryDelayMs,
-      crudUploadThrottleMs: this.options.crudUploadThrottleMs,
+      ...options,
       identifier: this.database.name,
       logger
     });

packages/web/src/db/PowerSyncDatabase.ts

@@ -189,8 +189,7 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
     const remote = new WebRemote(connector, this.logger);
     const syncOptions: WebStreamingSyncImplementationOptions = {
       ...(this.options as {}),
-      retryDelayMs: options.retryDelayMs,
-      crudUploadThrottleMs: options.crudUploadThrottleMs,
+      ...options,
       flags: this.resolvedFlags,
       adapter: this.bucketStorageAdapter,
       remote,

packages/web/src/db/sync/SSRWebStreamingSyncImplementation.ts

@@ -5,6 +5,7 @@ import {
   LockType,
   PowerSyncConnectionOptions,
   StreamingSyncImplementation,
+  SubscribedStream,
   SyncStatus,
   SyncStatusOptions
 } from '@powersync/common';
@@ -80,4 +81,9 @@ export class SSRStreamingSyncImplementation extends BaseObserver implements Stre
    * This is a no-op in SSR mode.
    */
   triggerCrudUpload() {}
+
+  /**
+   * No-op in SSR mode.
+   */
+  updateSubscriptions(): void {}
 }

packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -1,4 +1,10 @@
-import { PowerSyncConnectionOptions, PowerSyncCredentials, SyncStatus, SyncStatusOptions } from '@powersync/common';
+import {
+  PowerSyncConnectionOptions,
+  PowerSyncCredentials,
+  SubscribedStream,
+  SyncStatus,
+  SyncStatusOptions
+} from '@powersync/common';
 import * as Comlink from 'comlink';
 import { AbstractSharedSyncClientProvider } from '../../worker/sync/AbstractSharedSyncClientProvider';
 import {
@@ -12,6 +18,7 @@ import {
   WebStreamingSyncImplementation,
   WebStreamingSyncImplementationOptions
 } from './WebStreamingSyncImplementation';
+import { WorkerClient } from '../../worker/sync/WorkerClient';
 
 /**
  * The shared worker will trigger methods on this side of the message port
@@ -94,8 +101,11 @@ export interface SharedWebStreamingSyncImplementationOptions extends WebStreamin
   db: WebDBAdapter;
 }
 
+/**
+ * The local part of the sync implementation on the web, which talks to a sync implementation hosted in a shared worker.
+ */
 export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplementation {
-  protected syncManager: Comlink.Remote<SharedSyncImplementation>;
+  protected syncManager: Comlink.Remote<WorkerClient>;
   protected clientProvider: SharedSyncClientProvider;
   protected messagePort: MessagePort;
 
@@ -138,7 +148,7 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
       ).port;
     }
 
-    this.syncManager = Comlink.wrap<SharedSyncImplementation>(this.messagePort);
+    this.syncManager = Comlink.wrap<WorkerClient>(this.messagePort);
     this.syncManager.setLogLevel(this.logger.getLevel());
 
     this.triggerCrudUpload = this.syncManager.triggerCrudUpload;
@@ -152,15 +162,18 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
     const { crudUploadThrottleMs, identifier, retryDelayMs } = this.options;
     const flags = { ...this.webOptions.flags, workers: undefined };
 
-    this.isInitialized = this.syncManager.setParams({
-      dbParams: this.dbAdapter.getConfiguration(),
-      streamOptions: {
-        crudUploadThrottleMs,
-        identifier,
-        retryDelayMs,
-        flags: flags
-      }
-    });
+    this.isInitialized = this.syncManager.setParams(
+      {
+        dbParams: this.dbAdapter.getConfiguration(),
+        streamOptions: {
+          crudUploadThrottleMs,
+          identifier,
+          retryDelayMs,
+          flags: flags
+        }
+      },
+      options.subscriptions
+    );
 
     /**
      * Pass along any sync status updates to this listener
@@ -235,6 +248,10 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
     return this.isInitialized;
   }
 
+  updateSubscriptions(subscriptions: SubscribedStream[]): void {
+    this.syncManager.updateSubscriptions(subscriptions);
+  }
+
   /**
    * Used in tests to force a connection states
    */