Move Differential diff out of state to a listener.

Author: stevensJourney

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

@@ -35,7 +35,6 @@ export class PowerSyncYjsProvider extends ObservableV2<PowerSyncYjsEvents> {
     public readonly documentId: string
   ) {
     super();
-
     /**
      * Watch for changes to the `document_updates` table for this document.
      * This will be used to apply updates from other editors.
@@ -71,8 +70,8 @@ export class PowerSyncYjsProvider extends ObservableV2<PowerSyncYjsEvents> {
     let synced = false;
 
     updateQuery.registerListener({
-      onStateChange: async () => {
-        for (const added of updateQuery.state.diff.added) {
+      onDiff: async (diff) => {
+        for (const added of diff.added) {
           Y.applyUpdateV2(doc, b64ToUint8Array(added.update_b64));
         }
         if (!synced) {

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

@@ -82,8 +82,11 @@ export const DEFAULT_WATCH_QUERY_OPTIONS: WatchedQueryOptions = {
   reportFetching: true
 };
 
-export interface WatchedQuery<Data = unknown, Settings extends WatchedQueryOptions = WatchedQueryOptions>
-  extends MetaBaseObserverInterface<WatchedQueryListener<Data>> {
+export interface WatchedQuery<
+  Data = unknown,
+  Settings extends WatchedQueryOptions = WatchedQueryOptions,
+  Listener extends WatchedQueryListener<Data> = WatchedQueryListener<Data>
+> extends MetaBaseObserverInterface<Listener> {
   /**
    * Current state of the watched query.
    */
@@ -95,7 +98,7 @@ export interface WatchedQuery<Data = unknown, Settings extends WatchedQueryOptio
    * Subscribe to watched query events.
    * @returns A function to unsubscribe from the events.
    */
-  registerListener(listener: WatchedQueryListener<Data>): () => void;
+  registerListener(listener: Listener): () => void;
 
   /**
    * Updates the underlying query options.

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

@@ -1,4 +1,10 @@
-import { WatchCompatibleQuery, WatchedQuery, WatchedQueryOptions, WatchedQueryState } from '../WatchedQuery.js';
+import {
+  WatchCompatibleQuery,
+  WatchedQuery,
+  WatchedQueryListener,
+  WatchedQueryOptions,
+  WatchedQueryState
+} from '../WatchedQuery.js';
 import {
   AbstractQueryProcessor,
   AbstractQueryProcessorOptions,
@@ -95,11 +101,17 @@ type MutableDifferentialWatchedQueryState<RowType> = MutableWatchedQueryState<Ro
   diff: WatchedQueryDifferential<RowType>;
 };
 
-export interface DifferentialWatchedQuery<RowType>
-  extends WatchedQuery<ReadonlyArray<Readonly<RowType>>, DifferentialWatchedQuerySettings<RowType>> {
-  readonly state: DifferentialWatchedQueryState<RowType>;
+export interface DifferentialWatchedQueryListener<RowType>
+  extends WatchedQueryListener<ReadonlyArray<Readonly<RowType>>> {
+  onDiff?: (diff: WatchedQueryDifferential<RowType>) => void | Promise<void>;
 }
 
+export type DifferentialWatchedQuery<RowType> = WatchedQuery<
+  ReadonlyArray<Readonly<RowType>>,
+  DifferentialWatchedQuerySettings<RowType>,
+  DifferentialWatchedQueryListener<RowType>
+>;
+
 /**
  * @internal
  */
@@ -286,9 +298,9 @@ export class DifferentialQueryProcessor<RowType>
             currentMap = map;
 
             if (hasChanged) {
+              await this.iterateAsyncListenersWithError((l) => l.onDiff?.(diff));
               Object.assign(partialStateUpdate, {
-                data: diff.all,
-                diff
+                data: diff.all
               });
             }
 

packages/web/tests/watch.test.ts

@@ -1,4 +1,10 @@
-import { AbstractPowerSyncDatabase, ArrayComparator, GetAllQuery, WatchedQueryState } from '@powersync/common';
+import {
+  AbstractPowerSyncDatabase,
+  ArrayComparator,
+  GetAllQuery,
+  WatchedQueryDifferential,
+  WatchedQueryState
+} from '@powersync/common';
 import { PowerSyncDatabase } from '@powersync/web';
 import { v4 as uuid } from 'uuid';
 import { afterEach, beforeEach, describe, expect, it, onTestFinished, vi } from 'vitest';
@@ -495,6 +501,14 @@ describe('Watch Tests', { sequential: true }, () => {
       })
       .differentialWatch();
 
+    const diffs: WatchedQueryDifferential<{ id: string; make: string }>[] = [];
+
+    watch.registerListener({
+      onDiff: (diff) => {
+        diffs.push(diff);
+      }
+    });
+
     // Create sample data
     await powersync.execute(
       /* sql */ `
@@ -508,7 +522,7 @@ describe('Watch Tests', { sequential: true }, () => {
 
     await vi.waitFor(
       () => {
-        expect(watch.state.diff.added[0]?.make).equals('test1');
+        expect(diffs[0].added[0]?.make).equals('test1');
       },
       { timeout: 1000 }
     );
@@ -525,12 +539,13 @@ describe('Watch Tests', { sequential: true }, () => {
 
     await vi.waitFor(
       () => {
+        expect(diffs).toHaveLength(2);
         // This should now reflect that we had one change since the last event
-        expect(watch.state.diff.added).toHaveLength(1);
-        expect(watch.state.diff.added[0]?.make).equals('test2');
+        expect(diffs[1].added).toHaveLength(1);
+        expect(diffs[1].added[0]?.make).equals('test2');
 
-        expect(watch.state.diff.removed).toHaveLength(0);
-        expect(watch.state.diff.all).toHaveLength(2);
+        expect(diffs[1].removed).toHaveLength(0);
+        expect(diffs[1].all).toHaveLength(2);
       },
       { timeout: 1000 }
     );
@@ -546,13 +561,14 @@ describe('Watch Tests', { sequential: true }, () => {
 
     await vi.waitFor(
       () => {
-        expect(watch.state.diff.added).toHaveLength(0);
-        expect(watch.state.diff.all).toHaveLength(1);
-        expect(watch.state.diff.unchanged).toHaveLength(1);
-        expect(watch.state.diff.unchanged[0]?.make).equals('test1');
-
-        expect(watch.state.diff.removed).toHaveLength(1);
-        expect(watch.state.diff.removed[0]?.make).equals('test2');
+        expect(diffs).toHaveLength(3);
+        expect(diffs[2].added).toHaveLength(0);
+        expect(diffs[2].all).toHaveLength(1);
+        expect(diffs[2].unchanged).toHaveLength(1);
+        expect(diffs[2].unchanged[0]?.make).equals('test1');
+
+        expect(diffs[2].removed).toHaveLength(1);
+        expect(diffs[2].removed[0]?.make).equals('test2');
       },
       { timeout: 1000 }
     );
@@ -592,9 +608,18 @@ describe('Watch Tests', { sequential: true }, () => {
       ['test1', uuid()]
     );
 
+    const diffs: WatchedQueryDifferential<{ id: string; make: string }>[] = [];
+
+    watch.registerListener({
+      onDiff: (diff) => {
+        diffs.push(diff);
+      }
+    });
+
     await vi.waitFor(
       () => {
-        expect(watch.state.diff.added[0]?.make).equals('test1');
+        expect(diffs).toHaveLength(1);
+        expect(diffs[0].added[0]?.make).equals('test1');
       },
       { timeout: 1000 }
     );
@@ -611,12 +636,13 @@ describe('Watch Tests', { sequential: true }, () => {
 
     await vi.waitFor(
       () => {
+        expect(diffs).toHaveLength(2);
         // This should now reflect that we had one change since the last event
-        expect(watch.state.diff.added).toHaveLength(1);
-        expect(watch.state.diff.added[0]?.make).equals('test2');
+        expect(diffs[1].added).toHaveLength(1);
+        expect(diffs[1].added[0]?.make).equals('test2');
 
-        expect(watch.state.diff.removed).toHaveLength(0);
-        expect(watch.state.diff.all).toHaveLength(2);
+        expect(diffs[1].removed).toHaveLength(0);
+        expect(diffs[1].all).toHaveLength(2);
       },
       { timeout: 1000 }
     );
@@ -741,6 +767,14 @@ describe('Watch Tests', { sequential: true }, () => {
     // It should have the initial value
     expect(watch.state.data).toHaveLength(1);
 
+    const diffs: WatchedQueryDifferential<{ id: string; make: string }>[] = [];
+
+    watch.registerListener({
+      onDiff: (diff) => {
+        diffs.push(diff);
+      }
+    });
+
     await powersync.execute(
       /* sql */ `
         INSERT INTO
@@ -753,12 +787,13 @@ describe('Watch Tests', { sequential: true }, () => {
 
     await vi.waitFor(
       () => {
+        expect(diffs).toHaveLength(1);
         // This should now reflect that we had one change since the last event
-        expect(watch.state.diff.added).toHaveLength(1);
-        expect(watch.state.diff.added[0]?.make).equals('test2');
+        expect(diffs[0].added).toHaveLength(1);
+        expect(diffs[0].added[0]?.make).equals('test2');
 
-        expect(watch.state.diff.removed).toHaveLength(0);
-        expect(watch.state.diff.all).toHaveLength(2);
+        expect(diffs[0].removed).toHaveLength(0);
+        expect(diffs[0].all).toHaveLength(2);
       },
       { timeout: 1000 }
     );
@@ -774,13 +809,14 @@ describe('Watch Tests', { sequential: true }, () => {
 
     await vi.waitFor(
       () => {
-        expect(watch.state.diff.added).toHaveLength(0);
-        expect(watch.state.diff.all).toHaveLength(1);
-        expect(watch.state.diff.unchanged).toHaveLength(1);
-        expect(watch.state.diff.unchanged[0]?.make).equals('test1');
-
-        expect(watch.state.diff.removed).toHaveLength(1);
-        expect(watch.state.diff.removed[0]?.make).equals('test2');
+        expect(diffs).toHaveLength(2);
+        expect(diffs[1].added).toHaveLength(0);
+        expect(diffs[1].all).toHaveLength(1);
+        expect(diffs[1].unchanged).toHaveLength(1);
+        expect(diffs[1].unchanged[0]?.make).equals('test1');
+
+        expect(diffs[1].removed).toHaveLength(1);
+        expect(diffs[1].removed[0]?.make).equals('test2');
       },
       { timeout: 1000 }
     );
@@ -823,6 +859,14 @@ describe('Watch Tests', { sequential: true }, () => {
       { timeout: 1000, interval: 100 }
     );
 
+    const diffs: WatchedQueryDifferential<{ id: string; make: string }>[] = [];
+
+    watch.registerListener({
+      onDiff: (diff) => {
+        diffs.push(diff);
+      }
+    });
+
     await powersync.execute(
       /* sql */ `
         UPDATE assets
@@ -836,16 +880,17 @@ describe('Watch Tests', { sequential: true }, () => {
 
     await vi.waitFor(
       () => {
-        expect(watch.state.diff.added).toHaveLength(0);
-        const updated = watch.state.diff.updated[0];
+        expect(diffs).toHaveLength(1);
+        expect(diffs[0].added).toHaveLength(0);
+        const updated = diffs[0].updated[0];
 
         // The update should contain previous and current values of changed rows
         expect(updated).toBeDefined();
         expect(updated.previous.make).equals('test1');
         expect(updated.current.make).equals('test2');
 
-        expect(watch.state.diff.removed).toHaveLength(0);
-        expect(watch.state.diff.all).toHaveLength(1);
+        expect(diffs[0].removed).toHaveLength(0);
+        expect(diffs[0].all).toHaveLength(1);
       },
       { timeout: 1000 }
     );