Port sync streams API

Author: simolus3

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -9,14 +9,13 @@ import {
   UpdateNotification,
   isBatchedUpdateNotification
 } from '../db/DBAdapter.js';
-import { FULL_SYNC_PRIORITY } from '../db/crud/SyncProgress.js';
-import { SyncPriorityStatus, SyncStatus } from '../db/crud/SyncStatus.js';
+import { SyncStatus } from '../db/crud/SyncStatus.js';
 import { UploadQueueStats } from '../db/crud/UploadQueueStatus.js';
 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 } from './ConnectionManager.js';
+import { ConnectionManager, CreateSyncImplementationOptions } from './ConnectionManager.js';
 import { CustomQuery } from './CustomQuery.js';
 import { ArrayQueryDefinition, Query } from './Query.js';
 import { SQLOpenFactory, SQLOpenOptions, isDBAdapter, isSQLOpenFactory, isSQLOpenOptions } from './SQLOpenFactory.js';
@@ -40,6 +39,7 @@ import { TriggerManagerImpl } from './triggers/TriggerManagerImpl.js';
 import { DEFAULT_WATCH_THROTTLE_MS, WatchCompatibleQuery } from './watched/WatchedQuery.js';
 import { OnChangeQueryProcessor } from './watched/processors/OnChangeQueryProcessor.js';
 import { WatchedQueryComparator } from './watched/processors/comparators.js';
+import { coreStatusToJs, CoreSyncStatus } from './sync/stream/core-instruction.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -237,9 +237,15 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
 
     // Start async init
     this.connectionManager = new ConnectionManager({
+      firstStatusMatching: (predicate) => this.waitForStatus(predicate),
+      resolveOfflineSyncStatus: () => this.resolveOfflineSyncStatus(),
+      rustSubscriptionsCommand: async (payload) => {
+        await this.writeTransaction((tx) => {
+          return tx.execute('select powersync_control(?,?)', ['subscriptions', JSON.stringify(payload)]);
+        });
+      },
       createSyncImplementation: async (connector, options) => {
         await this.waitForReady();
-
         return this.runExclusive(async () => {
           const sync = this.generateSyncStreamImplementation(connector, this.resolvedConnectionOptions(options));
           const onDispose = sync.registerListener({
@@ -304,7 +310,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
 
   protected abstract generateSyncStreamImplementation(
     connector: PowerSyncBackendConnector,
-    options: RequiredAdditionalConnectionOptions
+    options: CreateSyncImplementationOptions & RequiredAdditionalConnectionOptions
   ): StreamingSyncImplementation;
 
   protected abstract generateBucketStorageAdapter(): BucketStorageAdapter;
@@ -338,13 +344,18 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
         ? (status: SyncStatus) => status.hasSynced
         : (status: SyncStatus) => status.statusForPriority(priority).hasSynced;
 
-    if (statusMatches(this.currentStatus)) {
+    return this.waitForStatus(statusMatches, signal);
+  }
+
+  private async waitForStatus(predicate: (status: SyncStatus) => any, signal?: AbortSignal): Promise<void> {
+    if (predicate(this.currentStatus)) {
       return;
     }
+
     return new Promise((resolve) => {
       const dispose = this.registerListener({
         statusChanged: (status) => {
-          if (statusMatches(status)) {
+          if (predicate(status)) {
             dispose();
             resolve();
           }
@@ -373,7 +384,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     await this.bucketStorageAdapter.init();
     await this._loadVersion();
     await this.updateSchema(this.options.schema);
-    await this.updateHasSynced();
+    await this.resolveOfflineSyncStatus();
     await this.database.execute('PRAGMA RECURSIVE_TRIGGERS=TRUE');
     this.ready = true;
     this.iterateListeners((cb) => cb.initialized?.());
@@ -403,30 +414,13 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     }
   }
 
-  protected async updateHasSynced() {
-    const result = await this.database.getAll<{ priority: number; last_synced_at: string }>(
-      'SELECT priority, last_synced_at FROM ps_sync_state ORDER BY priority DESC'
-    );
-    let lastCompleteSync: Date | undefined;
-    const priorityStatusEntries: SyncPriorityStatus[] = [];
+  protected async resolveOfflineSyncStatus() {
+    const result = await this.database.get<{ r: string }>('SELECT powersync_offline_sync_status() as r');
+    const parsed = JSON.parse(result.r) as CoreSyncStatus;
 
-    for (const { priority, last_synced_at } of result) {
-      const parsedDate = new Date(last_synced_at + 'Z');
-
-      if (priority == FULL_SYNC_PRIORITY) {
-        // This lowest-possible priority represents a complete sync.
-        lastCompleteSync = parsedDate;
-      } else {
-        priorityStatusEntries.push({ priority, hasSynced: true, lastSyncedAt: parsedDate });
-      }
-    }
-
-    const hasSynced = lastCompleteSync != null;
     const updatedStatus = new SyncStatus({
       ...this.currentStatus.toJSON(),
-      hasSynced,
-      priorityStatusEntries,
-      lastSyncedAt: lastCompleteSync
+      ...coreStatusToJs(parsed)
     });
 
     if (!updatedStatus.isEqual(this.currentStatus)) {
@@ -471,7 +465,9 @@ 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 {
+  protected resolvedConnectionOptions(
+    options: CreateSyncImplementationOptions
+  ): CreateSyncImplementationOptions & RequiredAdditionalConnectionOptions {
     return {
       ...options,
       retryDelayMs:
@@ -540,6 +536,8 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     this.iterateListeners((l) => l.statusChanged?.(this.currentStatus));
   }
 
+  syncStream(name: string, params?: Record<string, any>) {}
+
   /**
    * Close the database, releasing resources.
    *

packages/common/src/client/ConnectionManager.ts

@@ -2,9 +2,18 @@ import { ILogger } from 'js-logger';
 import { BaseListener, BaseObserver } from '../utils/BaseObserver.js';
 import { PowerSyncBackendConnector } from './connection/PowerSyncBackendConnector.js';
 import {
+  AdditionalConnectionOptions,
   InternalConnectionOptions,
-  StreamingSyncImplementation
+  StreamingSyncImplementation,
+  SubscribedStream
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
+import {
+  SyncStream,
+  SyncStreamDescription,
+  SyncStreamSubscribeOptions,
+  SyncStreamSubscription
+} from './sync/sync-streams.js';
+import { SyncStatus } from '../db/crud/SyncStatus.js';
 
 /**
  * @internal
@@ -18,14 +27,26 @@ export interface ConnectionManagerSyncImplementationResult {
   onDispose: () => Promise<void> | void;
 }
 
+/**
+ * The subset of {@link AbstractStreamingSyncImplementationOptions} managed by the connection manager.
+ *
+ * @internal
+ */
+export interface CreateSyncImplementationOptions extends AdditionalConnectionOptions {
+  subscriptions: SubscribedStream[];
+}
+
 /**
  * @internal
  */
 export interface ConnectionManagerOptions {
   createSyncImplementation(
     connector: PowerSyncBackendConnector,
-    options: InternalConnectionOptions
+    options: CreateSyncImplementationOptions
   ): Promise<ConnectionManagerSyncImplementationResult>;
+  firstStatusMatching(predicate: (status: SyncStatus) => any): Promise<void>;
+  resolveOfflineSyncStatus(): Promise<void>;
+  rustSubscriptionsCommand(payload: any): Promise<void>;
   logger: ILogger;
 }
 
@@ -76,6 +97,13 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
    */
   protected syncDisposer: (() => Promise<void> | void) | null;
 
+  /**
+   * Subscriptions managed in this connection manager.
+   *
+   * On the web, these local subscriptions are merged across tabs by a shared worker.
+   */
+  private locallyActiveSubscriptions = new Map<string, ActiveSubscription>();
+
   constructor(protected options: ConnectionManagerOptions) {
     super();
     this.connectingPromise = null;
@@ -102,7 +130,7 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
     // Update pending options to the latest values
     this.pendingConnectionOptions = {
       connector,
-      options: options ?? {}
+      options
     };
 
     // Disconnecting here provides aborting in progress connection attempts.
@@ -169,7 +197,11 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
         appliedOptions = options;
 
         this.pendingConnectionOptions = null;
-        const { sync, onDispose } = await this.options.createSyncImplementation(connector, options);
+
+        const { sync, onDispose } = await this.options.createSyncImplementation(connector, {
+          subscriptions: this.activeStreams,
+          ...options
+        });
         this.iterateListeners((l) => l.syncStreamCreated?.(sync));
         this.syncStreamImplementation = sync;
         this.syncDisposer = onDispose;
@@ -236,4 +268,109 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
     await sync?.dispose();
     await disposer?.();
   }
+
+  stream(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);
+    };
+
+    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({
+          subscribe: {
+            stream: {
+              name,
+              params: parameters
+            },
+            ttl: options?.ttl,
+            priority: options?.priority
+          }
+        });
+
+        if (!this.pendingConnectionOptions) {
+          // 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();
+        }
+
+        const key = `${name}|${JSON.stringify(parameters)}`;
+        let subscription = this.locallyActiveSubscriptions.get(key);
+        if (subscription == null) {
+          const clearSubscription = () => {
+            this.locallyActiveSubscriptions.delete(key);
+            this.subscriptionsMayHaveChanged();
+          };
+
+          subscription = new ActiveSubscription(name, parameters, waitForFirstSync, clearSubscription);
+          this.locallyActiveSubscriptions.set(key, subscription);
+          this.subscriptionsMayHaveChanged();
+        }
+
+        return new SyncStreamSubscriptionHandle(subscription);
+      },
+      unsubscribeAll: async () => {
+        await this.options.rustSubscriptionsCommand({ unsubscribe: { name, params: parameters } });
+        this.subscriptionsMayHaveChanged();
+      }
+    };
+  }
+
+  private get activeStreams() {
+    return [...this.locallyActiveSubscriptions.values().map((a) => ({ name: a.name, params: a.parameters }))];
+  }
+
+  private subscriptionsMayHaveChanged() {
+    if (this.syncStreamImplementation) {
+      this.syncStreamImplementation.updateSubscriptions(this.activeStreams);
+    }
+  }
+}
+
+class ActiveSubscription {
+  refcount: number = 0;
+
+  constructor(
+    readonly name: string,
+    readonly parameters: Record<string, any> | null,
+    readonly waitForFirstSync: () => Promise<void>,
+    private clearSubscription: () => void
+  ) {}
+
+  decrementRefCount() {
+    this.refcount--;
+    if (this.refcount == 0) {
+      this.clearSubscription();
+    }
+  }
 }
+
+class SyncStreamSubscriptionHandle implements SyncStreamSubscription {
+  constructor(readonly subscription: ActiveSubscription) {
+    _finalizer?.register(this, subscription);
+  }
+
+  get name() {
+    return this.subscription.name;
+  }
+
+  get parameters() {
+    return this.subscription.parameters;
+  }
+
+  waitForFirstSync(): Promise<void> {
+    return this.subscription.waitForFirstSync();
+  }
+
+  unsubscribe(): void {
+    _finalizer?.unregister(this);
+    this.subscription.decrementRefCount();
+  }
+}
+
+const _finalizer =
+  FinalizationRegistry != null ? new FinalizationRegistry<ActiveSubscription>((sub) => sub.decrementRefCount()) : null;

packages/common/src/client/sync/bucket/BucketStorageAdapter.ts

@@ -65,7 +65,8 @@ export enum PowerSyncControlCommand {
   STOP = 'stop',
   START = 'start',
   NOTIFY_TOKEN_REFRESHED = 'refreshed_token',
-  NOTIFY_CRUD_UPLOAD_COMPLETED = 'completed_upload'
+  NOTIFY_CRUD_UPLOAD_COMPLETED = 'completed_upload',
+  UPDATE_SUBSCRIPTIONS = 'update_subscriptions'
 }
 
 export interface BucketStorageListener extends BaseListener {