cleanup api

Author: stevensJourney

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -35,8 +35,7 @@ import {
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
 import { WatchedQuery } from './watched/WatchedQuery.js';
 import { WatchedQueryImpl } from './watched/WatchedQueryImpl.js';
-import { ComparisonQueryProcessor } from './watched/processors/comparison/ComparisonQueryProcessor.js';
-import { InlineWatchComparator } from './watched/processors/comparison/WatchComparator.js';
+import { OnChangeQueryProcessor } from './watched/processors/OnChangeQueryProcessor.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -877,12 +876,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     reportFetching?: boolean;
   }): WatchedQuery<T> {
     return new WatchedQueryImpl({
-      processor: new ComparisonQueryProcessor({
+      processor: new OnChangeQueryProcessor({
         db: this,
-        comparator: new InlineWatchComparator({
-          hash: (row) => JSON.stringify(row),
-          identify: (row) => (row as any).id ?? JSON.stringify(row) // TODO
-        }),
+        compareBy: (item) => JSON.stringify(item), // TODO make configurable
         watchedQuery: {
           query: options.sql,
           parameters: options.parameters,

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

@@ -1,16 +1,16 @@
 import { DataStream } from '../../utils/DataStream.js';
-import { WatchedQueryResult } from './WatchedQueryResult.js';
 
 export interface WatchedQueryState<T> {
-  loading: boolean;
-  fetching: boolean;
+  isLoading: boolean;
+  isFetching: boolean;
   error: Error | null;
   lastUpdated: Date | null;
-  data: WatchedQueryResult<T>;
+  data: T[];
 }
 
 /**
  * Performs underlaying watching and yields a stream of results.
+ * @internal
  */
 export interface WatchedQueryProcessor<T> {
   readonly state: WatchedQueryState<T>;
@@ -20,6 +20,9 @@ export interface WatchedQueryProcessor<T> {
   updateQuery(query: WatchedQueryOptions<T>): void;
 }
 
+/**
+ * @internal
+ */
 export interface WatchedQueryOptions<T> {
   query: string;
   parameters?: any[];

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

@@ -8,9 +8,14 @@ export interface WatchedQueryImplOptions<T> {
 
 export class WatchedQueryImpl<T> implements WatchedQuery<T> {
   protected lazyStreamPromise: Promise<DataStream<WatchedQueryState<T>>>;
+  protected _stream: DataStream<WatchedQueryState<T>> | null;
 
   constructor(protected options: WatchedQueryImplOptions<T>) {
-    this.lazyStreamPromise = this.options.processor.generateStream();
+    this._stream = null;
+    this.lazyStreamPromise = this.options.processor.generateStream().then((s) => {
+      this._stream = s;
+      return s;
+    });
   }
 
   get state() {
@@ -67,9 +72,13 @@ export class WatchedQueryImpl<T> implements WatchedQuery<T> {
   }
 
   close(): void {
+    if (this._stream) {
+      this._stream.close().catch(() => {});
+      return;
+    }
     this.lazyStreamPromise
-      .then((s) => {
-        s.close();
+      .then(async (s) => {
+        await s.close();
       })
       .catch(() => {
         // In rare cases where the DB might be closed before the stream is created

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

@@ -37,14 +37,11 @@ export abstract class AbstractQueryProcessor<T>
   implements WatchedQueryProcessor<T>
 {
   readonly state: WatchedQueryState<T> = {
-    loading: true,
-    fetching: true,
+    isLoading: true,
+    isFetching: true,
     error: null,
     lastUpdated: null,
-    data: {
-      all: [],
-      delta: () => ({ added: [], removed: [], unchanged: [], updated: [] })
-    }
+    data: []
   };
 
   protected _stream: DataStream<WatchedQueryState<T>> | null;
@@ -78,7 +75,7 @@ export abstract class AbstractQueryProcessor<T>
   protected abstract linkStream(options: LinkQueryStreamOptions<T>): Promise<void>;
 
   protected updateState(update: Partial<WatchedQueryState<T>>) {
-    Object.assign(this.state, update);
+    Object.assign(this.state, { lastUpdated: new Date() } satisfies Partial<WatchedQueryState<T>>, update);
 
     if (this._stream?.closed) {
       // Don't enqueue data in a closed stream.

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

@@ -1,22 +1,52 @@
 import { WatchedQueryState } from '../WatchedQuery.js';
-import { WatchedQueryResult } from '../WatchedQueryResult.js';
-import { AbstractQueryProcessor, LinkQueryStreamOptions } from './AbstractQueryProcessor.js';
+import {
+  AbstractQueryProcessor,
+  AbstractQueryProcessorOptions,
+  LinkQueryStreamOptions
+} from './AbstractQueryProcessor.js';
+
+export interface OnChangeQueryProcessorOptions<T> extends AbstractQueryProcessorOptions<T> {
+  compareBy?: (element: T) => string;
+}
 
 /**
  * Uses the PowerSync onChange event to trigger watched queries.
  * Results are emitted on every change of the relevant tables.
  */
 export class OnChangeQueryProcessor<T> extends AbstractQueryProcessor<T> {
-  /**
-   * Always returns the result set on every onChange event. Deltas are not supported by this processor.
+  constructor(protected options: OnChangeQueryProcessorOptions<T>) {
+    super(options);
+  }
+
+  /*
+   * @returns If the sets are equal
    */
-  protected processResultSet(result: T[]): WatchedQueryResult<T> | null {
-    return {
-      all: result,
-      delta: () => {
-        throw new Error('Delta not implemented for OnChangeQueryProcessor');
+  protected checkEquality(current: T[], previous: T[]): boolean {
+    if (current.length == 0 && previous.length == 0) {
+      return true;
+    }
+
+    if (current.length !== previous.length) {
+      return false;
+    }
+
+    const { compareBy } = this.options;
+    // Assume items are not equal if we can't compare them
+    if (!compareBy) {
+      return false;
+    }
+
+    // At this point the lengths are equal
+    for (let i = 0; i < current.length; i++) {
+      const currentItem = compareBy(current[i]);
+      const previousItem = compareBy(previous[i]);
+
+      if (currentItem !== previousItem) {
+        return false;
       }
-    };
+    }
+
+    return true;
   }
 
   protected async linkStream(options: LinkQueryStreamOptions<T>): Promise<void> {
@@ -31,7 +61,7 @@ export class OnChangeQueryProcessor<T> extends AbstractQueryProcessor<T> {
           // This fires for each change of the relevant tables
           try {
             if (this.reportFetching) {
-              this.updateState({ fetching: true });
+              this.updateState({ isFetching: true });
             }
 
             const partialStateUpdate: Partial<WatchedQueryState<T>> = {};
@@ -42,18 +72,16 @@ export class OnChangeQueryProcessor<T> extends AbstractQueryProcessor<T> {
               : await db.getAll<T>(watchedQuery.query, watchedQuery.parameters);
 
             if (this.reportFetching) {
-              partialStateUpdate.fetching = false;
+              partialStateUpdate.isFetching = false;
             }
 
-            if (this.state.loading) {
-              partialStateUpdate.loading = false;
+            if (this.state.isLoading) {
+              partialStateUpdate.isLoading = false;
             }
 
             // Check if the result has changed
-            const watchedQueryResult = this.processResultSet(result);
-            if (watchedQueryResult) {
-              partialStateUpdate.data = watchedQueryResult;
-              partialStateUpdate.lastUpdated = new Date();
+            if (!this.checkEquality(result, this.state.data)) {
+              partialStateUpdate.data = result;
             }
 
             if (Object.keys(partialStateUpdate).length > 0) {