wip: react suspense

Author: stevensJourney

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

@@ -1,3 +1,5 @@
+import { BaseListener, BaseObserverInterface } from '../../utils/BaseObserver.js';
+
 export interface WatchedQueryState<Data> {
   /**
    * Indicates the initial loading state (hard loading). Loading becomes false once the first set of results from the watched query is available or an error occurs.
@@ -47,18 +49,37 @@ export interface WatchedQueryOptions<DataType> {
   reportFetching?: boolean;
 }
 
+export enum WatchedQuerySubscriptionEvent {
+  ON_DATA = 'onData',
+  ON_ERROR = 'onError',
+  ON_STATE_CHANGE = 'onStateChange'
+}
+
 export interface WatchedQuerySubscription<Data> {
-  onData?: (data: Data) => void | Promise<void>;
-  onError?: (error: Error) => void | Promise<void>;
-  onStateChange?: (state: WatchedQueryState<Data>) => void | Promise<void>;
+  [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>;
+}
+
+export type SubscriptionCounts = Record<WatchedQuerySubscriptionEvent, number> & {
+  total: number;
+};
+
+export interface WatchedQueryListener extends BaseListener {
+  closed: () => void;
+  subscriptionsChanged: (counts: SubscriptionCounts) => void;
 }
 
-export interface WatchedQuery<Data> {
+export interface WatchedQuery<Data> extends BaseObserverInterface<WatchedQueryListener> {
   /**
    * 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.

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

@@ -1,6 +1,14 @@
 import { AbstractPowerSyncDatabase } from '../../../client/AbstractPowerSyncDatabase.js';
-import { BaseListener, BaseObserver } from '../../../utils/BaseObserver.js';
-import { WatchedQuery, WatchedQueryOptions, WatchedQueryState, WatchedQuerySubscription } from '../WatchedQuery.js';
+import { BaseObserver } from '../../../utils/BaseObserver.js';
+import {
+  SubscriptionCounts,
+  WatchedQuery,
+  WatchedQueryListener,
+  WatchedQueryOptions,
+  WatchedQueryState,
+  WatchedQuerySubscription,
+  WatchedQuerySubscriptionEvent
+} from '../WatchedQuery.js';
 
 /**
  * @internal
@@ -18,7 +26,7 @@ export interface LinkQueryOptions<Data> {
   query: WatchedQueryOptions<Data>;
 }
 
-type WatchedQueryProcessorListener<Data> = WatchedQuerySubscription<Data> & BaseListener;
+type WatchedQueryProcessorListener<Data> = WatchedQuerySubscription<Data> & WatchedQueryListener;
 
 /**
  * Performs underlaying watching and yields a stream of results.
@@ -32,10 +40,25 @@ export abstract class AbstractQueryProcessor<Data = unknown[]>
 
   protected abortController: AbortController;
   protected initialized: Promise<void>;
+  protected _closed: boolean;
+
+  get closed() {
+    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>) {
     super();
     this.abortController = new AbortController();
+    this._closed = false;
     this.state = {
       isLoading: true,
       isFetching: this.reportFetching, // Only set to true if we will report updates in future
@@ -108,12 +131,23 @@ export abstract class AbstractQueryProcessor<Data = unknown[]>
   }
 
   subscribe(subscription: WatchedQuerySubscription<Data>): () => void {
-    return this.registerListener({ ...subscription });
+    // hook in to subscription events in order to report changes
+    const baseDispose = this.registerListener({ ...subscription });
+
+    const counts = this.subscriptionCounts;
+    this.iterateListeners((l) => l.subscriptionsChanged?.(counts));
+
+    return () => {
+      baseDispose();
+      this.iterateListeners((l) => l.subscriptionsChanged?.(counts));
+    };
   }
 
   async close() {
     await this.initialized;
     this.abortController.abort();
+    this._closed = true;
+    this.iterateListeners((l) => l.closed?.());
   }
 
   /**

packages/react/src/QueryStore.ts

@@ -1,13 +1,13 @@
-import { AbstractPowerSyncDatabase } from '@powersync/common';
-import { Query, WatchedQuery } from './WatchedQuery';
+import { AbstractPowerSyncDatabase, WatchedQuery } from '@powersync/common';
+import { Query } from './WatchedQuery';
 import { AdditionalOptions } from './hooks/useQuery';
 
 export function generateQueryKey(sqlStatement: string, parameters: any[], options: AdditionalOptions): string {
   return `${sqlStatement} -- ${JSON.stringify(parameters)} -- ${JSON.stringify(options)}`;
 }
 
 export class QueryStore {
-  cache = new Map<string, WatchedQuery>();
+  cache = new Map<string, WatchedQuery<unknown[]>>();
 
   constructor(private db: AbstractPowerSyncDatabase) {}
 
@@ -16,17 +16,40 @@ export class QueryStore {
       return this.cache.get(key);
     }
 
-    const q = new WatchedQuery(this.db, query, options);
-    const disposer = q.registerListener({
-      disposed: () => {
+    const customExecutor = typeof query.rawQuery !== 'string' ? query.rawQuery : null;
+
+    const watchedQuery = this.db.incrementalWatch({
+      sql: query.sqlStatement,
+      parameters: query.queryParameters,
+      customExecutor: customExecutor
+        ? {
+            initialData: [],
+            execute: () => customExecutor.execute()
+          }
+        : undefined,
+      throttleMs: options.throttleMs
+    });
+
+    const disposer = watchedQuery.registerListener({
+      closed: () => {
         this.cache.delete(key);
         disposer?.();
       }
     });
 
-    this.cache.set(key, q);
+    watchedQuery.registerListener({
+      subscriptionsChanged: (counts) => {
+        // Dispose this query if there are no subscribers present
+        if (counts.total == 0) {
+          watchedQuery.close();
+          this.cache.delete(key);
+        }
+      }
+    });
+
+    this.cache.set(key, watchedQuery);
 
-    return q;
+    return watchedQuery;
   }
 }