unify watched query listeners and subscriptions to consistent API

Author: stevensJourney

demos/react-supabase-todolist/src/components/providers/SystemProvider.tsx

@@ -72,7 +72,7 @@ export const SystemProvider = ({ children }: { children: React.ReactNode }) => {
     });
 
     // This updates a cache in order to display results instantly on page load.
-    listsQuery.subscribe({
+    listsQuery.registerListener({
       onData: (data) => {
         // Store the data in localStorage for instant caching
         localStorage.setItem('listscache', JSON.stringify(data));

demos/yjs-react-supabase-text-collab/powersync.yaml

@@ -23,11 +23,6 @@ replication:
   # Multiple connection support is on the roadmap
   connections:
     - type: postgresql
-      # The PowerSync server container can access the Postgres DB via the DB's service name.
-      # In this case the hostname is pg-db
-
-      # The connection URI or individual parameters can be specified.
-      # Individual params take precedence over URI params
       uri: postgresql://postgres:postgres@supabase_db_yjs-react-supabase-text-collab:5432/postgres
 
       # SSL settings

demos/yjs-react-supabase-text-collab/src/library/powersync/PowerSyncYjsProvider.ts

@@ -72,7 +72,7 @@ export class PowerSyncYjsProvider extends ObservableV2<PowerSyncYjsEvents> {
 
     let synced = false;
 
-    updateQuery.subscribe({
+    updateQuery.registerListener({
       onData: async (diff) => {
         for (const added of diff.added) {
           Y.applyUpdateV2(doc, b64ToUint8Array(added.update_b64));

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -921,7 +921,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     const builderFactory = WatchedQueryBuilderMap[mode];
     if (!builderFactory) {
       throw new Error(
-        `Unsupported watch mode: ${mode}. Please specify on of [${Object.values(IncrementalWatchMode).join(', ')}]`
+        `Unsupported watch mode: ${mode}. Please specify one of [${Object.values(IncrementalWatchMode).join(', ')}]`
       );
     }
     return builderFactory(this) as WatchedQueryBuilderMap[Mode];
@@ -964,7 +964,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       }
     });
 
-    const dispose = watchedQuery.subscribe({
+    const dispose = watchedQuery.registerListener({
       onData: (data) => {
         if (!data) {
           // This should not happen. We only use null for the initial data.

packages/common/src/client/watched/GetAllQuery.ts

@@ -9,16 +9,16 @@ export type GetAllQueryOptions<RowType = unknown> = {
   sql: string;
   parameters?: ReadonlyArray<unknown>;
   /**
-   * Optional transformer function to convert raw rows into the desired RowType.
+   * Optional mapper function to convert raw rows into the desired RowType.
    * @example
-   * ```typescript
-   * (rawRow: Record<string, unknown>) => ({
-   *   id: rawRow.id as string,
+   * ```javascript
+   * (rawRow) => ({
+   *   id: rawRow.id,
    *   created_at: new Date(rawRow.created_at),
    * })
    * ```
    */
-  transformer?: (rawRow: Record<string, unknown>) => RowType;
+  mapper?: (rawRow: Record<string, unknown>) => RowType;
 };
 
 /**
@@ -38,8 +38,8 @@ export class GetAllQuery<RowType = unknown> implements WatchCompatibleQuery<RowT
     const { db } = options;
     const { sql, parameters = [] } = this.compile();
     const rawResult = await db.getAll<unknown>(sql, [...parameters]);
-    if (this.options.transformer) {
-      return rawResult.map(this.options.transformer);
+    if (this.options.mapper) {
+      return rawResult.map(this.options.mapper);
     }
     return rawResult as RowType[];
   }

packages/common/src/client/watched/WatchedQuery.ts

@@ -1,7 +1,11 @@
 import { CompiledQuery } from '../../types/types.js';
-import { BaseListener, BaseObserverInterface } from '../../utils/BaseObserver.js';
+import { BaseListener } from '../../utils/BaseObserver.js';
+import { MetaBaseObserverInterface } from '../../utils/MetaBaseObserver.js';
 import { AbstractPowerSyncDatabase } from '../AbstractPowerSyncDatabase.js';
 
+/**
+ * State for {@link WatchedQuery} instances.
+ */
 export interface WatchedQueryState<Data> {
   /**
    * Indicates the initial loading state (hard loading).
@@ -57,43 +61,34 @@ export interface WatchedQueryOptions {
   reportFetching?: boolean;
 }
 
-export enum WatchedQuerySubscriptionEvent {
+export enum WatchedQueryListenerEvent {
   ON_DATA = 'onData',
   ON_ERROR = 'onError',
-  ON_STATE_CHANGE = 'onStateChange'
-}
-
-export interface WatchedQuerySubscription<Data> {
-  [WatchedQuerySubscriptionEvent.ON_DATA]?: (data: Data) => void | Promise<void>;
-  [WatchedQuerySubscriptionEvent.ON_ERROR]?: (error: Error) => void | Promise<void>;
-  [WatchedQuerySubscriptionEvent.ON_STATE_CHANGE]?: (state: WatchedQueryState<Data>) => void | Promise<void>;
+  ON_STATE_CHANGE = 'onStateChange',
+  CLOSED = 'closed'
 }
 
-export type SubscriptionCounts = Record<WatchedQuerySubscriptionEvent, number> & {
-  total: number;
-};
-
-export interface WatchedQueryListener extends BaseListener {
-  closed: () => void;
-  subscriptionsChanged: (counts: SubscriptionCounts) => void;
+export interface WatchedQueryListener<Data> extends BaseListener {
+  [WatchedQueryListenerEvent.ON_DATA]?: (data: Data) => void | Promise<void>;
+  [WatchedQueryListenerEvent.ON_ERROR]?: (error: Error) => void | Promise<void>;
+  [WatchedQueryListenerEvent.ON_STATE_CHANGE]?: (state: WatchedQueryState<Data>) => void | Promise<void>;
+  [WatchedQueryListenerEvent.CLOSED]?: () => void | Promise<void>;
 }
 
 export interface WatchedQuery<Data = unknown, Settings extends WatchedQueryOptions = WatchedQueryOptions>
-  extends BaseObserverInterface<WatchedQueryListener> {
+  extends MetaBaseObserverInterface<WatchedQueryListener<Data>> {
   /**
    * Current state of the watched query.
    */
   readonly state: WatchedQueryState<Data>;
 
   readonly closed: boolean;
 
-  readonly subscriptionCounts: SubscriptionCounts;
-
   /**
    * Subscribe to watched query events.
    * @returns A function to unsubscribe from the events.
    */
-  subscribe(subscription: WatchedQuerySubscription<Data>): () => void;
+  registerListener(listener: WatchedQueryListener<Data>): () => void;
 
   /**
    * Updates the underlying query options.

packages/common/src/client/watched/processors/AbstractQueryProcessor.ts

@@ -1,14 +1,6 @@
 import { AbstractPowerSyncDatabase } from '../../../client/AbstractPowerSyncDatabase.js';
-import { BaseObserver } from '../../../utils/BaseObserver.js';
-import {
-  SubscriptionCounts,
-  WatchedQuery,
-  WatchedQueryListener,
-  WatchedQueryOptions,
-  WatchedQueryState,
-  WatchedQuerySubscription,
-  WatchedQuerySubscriptionEvent
-} from '../WatchedQuery.js';
+import { MetaBaseObserver } from '../../../utils/MetaBaseObserver.js';
+import { WatchedQuery, WatchedQueryListener, WatchedQueryOptions, WatchedQueryState } from '../WatchedQuery.js';
 
 /**
  * @internal
@@ -27,7 +19,7 @@ export interface LinkQueryOptions<Data, Settings extends WatchedQueryOptions = W
   settings: Settings;
 }
 
-type WatchedQueryProcessorListener<Data> = WatchedQuerySubscription<Data> & WatchedQueryListener;
+type WatchedQueryProcessorListener<Data> = WatchedQueryListener<Data>;
 
 /**
  * Performs underlying watching and yields a stream of results.
@@ -37,7 +29,7 @@ export abstract class AbstractQueryProcessor<
     Data = unknown[],
     Settings extends WatchedQueryOptions = WatchedQueryOptions
   >
-  extends BaseObserver<WatchedQueryProcessorListener<Data>>
+  extends MetaBaseObserver<WatchedQueryProcessorListener<Data>>
   implements WatchedQuery<Data, Settings>
 {
   readonly state: WatchedQueryState<Data>;
@@ -51,15 +43,6 @@ export abstract class AbstractQueryProcessor<
     return this._closed;
   }
 
-  get subscriptionCounts() {
-    const listenersArray = Array.from(this.listeners);
-    return Object.values(WatchedQuerySubscriptionEvent).reduce((totals: Partial<SubscriptionCounts>, key) => {
-      totals[key] = listenersArray.filter((l) => !!l[key]).length;
-      totals.total = (totals.total ?? 0) + totals[key];
-      return totals;
-    }, {}) as SubscriptionCounts;
-  }
-
   constructor(protected options: AbstractQueryProcessorOptions<Data, Settings>) {
     super();
     this.abortController = new AbortController();
@@ -156,18 +139,6 @@ export abstract class AbstractQueryProcessor<
     });
   }
 
-  subscribe(subscription: WatchedQuerySubscription<Data>): () => void {
-    // hook in to subscription events in order to report changes
-    const baseDispose = this.registerListener({ ...subscription });
-
-    this.iterateListeners((l) => l.subscriptionsChanged?.(this.subscriptionCounts));
-
-    return () => {
-      baseDispose();
-      this.iterateListeners((l) => l.subscriptionsChanged?.(this.subscriptionCounts));
-    };
-  }
-
   async close() {
     await this.initialized;
     this.abortController.abort();

packages/common/src/client/watched/processors/ComparisonWatchedQueryBuilder.ts

@@ -4,11 +4,28 @@ import { WatchedQueryBuilder } from '../WatchedQueryBuilder.js';
 import { WatchedQueryComparator } from './comparators.js';
 import { ComparisonWatchedQuerySettings, OnChangeQueryProcessor } from './OnChangeQueryProcessor.js';
 
+/**
+ * Options for building incrementally watched queries that compare the result set.
+ * It uses a comparator to determine if the result set has changed since the last update.
+ * If the result set has changed, it emits the new result set.
+ */
 export interface ComparisonWatchProcessorOptions<DataType> {
   comparator?: WatchedQueryComparator<DataType>;
   watch: ComparisonWatchedQuerySettings<DataType>;
 }
 
+/**
+ * Default implementation of the {@link WatchedQueryComparator} for watched queries.
+ * It uses JSON stringification to compare the entire result set.
+ * Array based results should use {@link ArrayComparator} for more efficient item comparison.
+ */
+export const DEFAULT_WATCHED_QUERY_COMPARATOR: WatchedQueryComparator<any> = {
+  checkEquality: (a, b) => JSON.stringify(a) === JSON.stringify(b)
+};
+
+/**
+ * Builds an incrementally watched query that emits results after comparing the result set for changes.
+ */
 export class ComparisonWatchedQueryBuilder implements WatchedQueryBuilder {
   constructor(protected db: AbstractPowerSyncDatabase) {}
 
@@ -41,9 +58,7 @@ export class ComparisonWatchedQueryBuilder implements WatchedQueryBuilder {
   ): WatchedQuery<DataType, ComparisonWatchedQuerySettings<DataType>> {
     return new OnChangeQueryProcessor({
       db: this.db,
-      comparator: options.comparator ?? {
-        checkEquality: (a, b) => JSON.stringify(a) == JSON.stringify(b)
-      },
+      comparator: options.comparator ?? DEFAULT_WATCHED_QUERY_COMPARATOR,
       watchOptions: options.watch,
       placeholderData: options.watch.placeholderData
     });

packages/common/src/client/watched/processors/DifferentialQueryProcessor.ts

@@ -1,24 +1,44 @@
 import { WatchCompatibleQuery, WatchedQuery, WatchedQueryOptions, WatchedQueryState } from '../WatchedQuery.js';
 import { AbstractQueryProcessor, AbstractQueryProcessorOptions, LinkQueryOptions } from './AbstractQueryProcessor.js';
 
-export interface Differential<RowType> {
+/**
+ * Represents an updated row in a differential watched query.
+ * It contains both the current and previous state of the row.
+ */
+export interface WatchedQueryRowDifferential<RowType> {
   current: RowType;
   previous: RowType;
 }
 
+/**
+ * Represents the result of a watched query that has been differentiated.
+ * {@link WatchedQueryState.data} is of the {@link WatchedQueryDifferential} form when using the {@link IncrementalWatchMode.DIFFERENTIAL} mode.
+ */
 export interface WatchedQueryDifferential<RowType> {
   added: RowType[];
   all: RowType[];
   removed: RowType[];
-  updated: Differential<RowType>[];
+  updated: WatchedQueryRowDifferential<RowType>[];
   unchanged: RowType[];
 }
 
-export interface Differentiator<RowType> {
+/**
+ * Differentiator for incremental watched queries which allows to identify and compare items in the result set.
+ */
+export interface WatchedQueryDifferentiator<RowType> {
+  /**
+   * Unique identifier for the item.
+   */
   identify: (item: RowType) => string;
+  /**
+   * Generates a key for comparing items with matching identifiers.
+   */
   compareBy: (item: RowType) => string;
 }
 
+/**
+ * Settings for incremental watched queries using the {@link IncrementalWatchMode.DIFFERENTIAL} mode.
+ */
 export interface DifferentialWatchedQuerySettings<RowType> extends WatchedQueryOptions {
   /**
    * The query here must return an array of items that can be differentiated.
@@ -37,11 +57,15 @@ export interface DifferentialWatchedQuerySettings<RowType> extends WatchedQueryO
  */
 export interface DifferentialQueryProcessorOptions<RowType>
   extends AbstractQueryProcessorOptions<WatchedQueryDifferential<RowType>, DifferentialWatchedQuerySettings<RowType>> {
-  differentiator: Differentiator<RowType>;
+  differentiator: WatchedQueryDifferentiator<RowType>;
 }
 
 type DataHashMap<RowType> = Map<string, { hash: string; item: RowType }>;
 
+/**
+ * An empty differential result set.
+ * This is used as the initial state for differential incrementally watched queries.
+ */
 export const EMPTY_DIFFERENTIAL = {
   added: [],
   all: [],

packages/common/src/client/watched/processors/DifferentialWatchedQueryBuilder.ts

@@ -4,16 +4,38 @@ import { WatchedQueryBuilder } from '../WatchedQueryBuilder.js';
 import {
   DifferentialQueryProcessor,
   DifferentialWatchedQuerySettings,
-  Differentiator,
   EMPTY_DIFFERENTIAL,
-  WatchedQueryDifferential
+  WatchedQueryDifferential,
+  WatchedQueryDifferentiator
 } from './DifferentialQueryProcessor.js';
 
+/**
+ * Options for creating an incrementally watched query that emits differential results.
+ *
+ */
 export type DifferentialWatchedQueryBuilderOptions<RowType> = {
-  differentiator?: Differentiator<RowType>;
+  differentiator?: WatchedQueryDifferentiator<RowType>;
   watch: DifferentialWatchedQuerySettings<RowType>;
 };
 
+/**
+ * Default implementation of the {@link Differentiator} for watched queries.
+ * It identifies items by their `id` property if available, otherwise it uses JSON stringification
+ * of the entire item for identification and comparison.
+ */
+export const DEFAULT_WATCHED_QUERY_DIFFERENTIATOR: WatchedQueryDifferentiator<any> = {
+  identify: (item) => {
+    if (item && typeof item == 'object' && typeof item['id'] == 'string') {
+      return item['id'];
+    }
+    return JSON.stringify(item);
+  },
+  compareBy: (item) => JSON.stringify(item)
+};
+
+/**
+ * Builds a watched query which emits differential results based on the provided differentiator.
+ */
 export class DifferentialWatchedQueryBuilder implements WatchedQueryBuilder {
   constructor(protected db: AbstractPowerSyncDatabase) {}
 
@@ -39,7 +61,7 @@ export class DifferentialWatchedQueryBuilder implements WatchedQueryBuilder {
    *        FROM
    *          assets
    *      ',
-   *      transformer: (raw) => {
+   *      mapper: (raw) => {
    *        return {
    *          id: raw.id as string,
    *          make: raw.make as string
@@ -55,15 +77,7 @@ export class DifferentialWatchedQueryBuilder implements WatchedQueryBuilder {
   ): WatchedQuery<WatchedQueryDifferential<RowType>, DifferentialWatchedQuerySettings<RowType>> {
     return new DifferentialQueryProcessor({
       db: this.db,
-      differentiator: options.differentiator ?? {
-        identify: (item: RowType) => {
-          if (item && typeof item == 'object' && typeof item['id'] == 'string') {
-            return item['id'];
-          }
-          return JSON.stringify(item);
-        },
-        compareBy: (item: RowType) => JSON.stringify(item)
-      },
+      differentiator: options.differentiator ?? DEFAULT_WATCHED_QUERY_DIFFERENTIATOR,
       watchOptions: options.watch,
       placeholderData: options.watch.placeholderData ?? EMPTY_DIFFERENTIAL
     });

packages/common/src/client/watched/processors/comparators.ts

@@ -13,7 +13,7 @@ export type ArrayComparatorOptions<ItemType> = {
 };
 
 /**
- * Compares array results of watched queries.
+ * Compares array results of watched queries for incrementally watched queries created in the {@link IncrementalWatchMode.COMPARISON} mode.
  */
 export class ArrayComparator<ItemType> implements WatchedQueryComparator<ItemType[]> {
   constructor(protected options: ArrayComparatorOptions<ItemType>) {}