Emit the data for the BatchedUpdateNotification event that's available in the updateHook rather than emitting a partial object.

Author: douglascayers

packages/node/src/db/AsyncDatabase.ts

@@ -1,4 +1,4 @@
-import { QueryResult } from '@powersync/common';
+import { BatchedUpdateNotification, QueryResult } from '@powersync/common';
 
 export type ProxiedQueryResult = Omit<QueryResult, 'rows'> & {
   rows?: {
@@ -20,5 +20,5 @@ export interface AsyncDatabase {
   // This happens on the worker because we otherwise get race conditions when wrapping
   // callbacks to invoke on the main thread (we need a guarantee that collectCommittedUpdates
   // contains entries immediately after calling COMMIT).
-  collectCommittedUpdates: () => Promise<string[]>;
+  collectCommittedUpdates: () => Promise<BatchedUpdateNotification>;
 }

packages/node/src/db/BetterSQLite3DBAdapter.ts

@@ -190,13 +190,8 @@ export class BetterSQLite3DBAdapter extends BaseObserver<DBAdapterListener> impl
         } finally {
           const updates = await this.writeConnection.database.collectCommittedUpdates();
 
-          if (updates.length > 0) {
-            const event: BatchedUpdateNotification = {
-              tables: updates,
-              groupedUpdates: {},
-              rawUpdates: []
-            };
-            this.iterateListeners((cb) => cb.tablesUpdated?.(event));
+          if (updates.tables.length > 0) {
+            this.iterateListeners((cb) => cb.tablesUpdated?.(updates));
           }
         }
       } finally {

packages/node/src/db/SqliteWorker.ts

@@ -1,5 +1,6 @@
 import * as path from 'node:path';
 import BetterSQLite3Database, { Database } from '@powersync/better-sqlite3';
+import { BatchedUpdateNotification, RowUpdateType, TableUpdateOperation, UpdateNotification } from '@powersync/common';
 import * as Comlink from 'comlink';
 import { parentPort, threadId } from 'node:worker_threads';
 import OS from 'node:os';
@@ -9,36 +10,69 @@ import { AsyncDatabase, AsyncDatabaseOpener } from './AsyncDatabase.js';
 class BlockingAsyncDatabase implements AsyncDatabase {
   private readonly db: Database;
 
-  private readonly uncommittedUpdatedTables = new Set<string>();
-  private readonly committedUpdatedTables = new Set<string>();
+  private readonly uncommittedUpdates = new Array<UpdateNotification>();
+  private readonly committedUpdates = new Array<UpdateNotification>();
 
   constructor(db: Database) {
     this.db = db;
 
     db.function('node_thread_id', () => threadId);
   }
 
-  collectCommittedUpdates() {
-    const resolved = Promise.resolve([...this.committedUpdatedTables]);
-    this.committedUpdatedTables.clear();
-    return resolved;
+  async collectCommittedUpdates() {
+    const rawUpdates: UpdateNotification[] = [];
+    const groupedUpdates: Record<string, TableUpdateOperation[]> = {};
+
+    for (const rawUpdate of this.committedUpdates) {
+      rawUpdates.push(rawUpdate);
+      groupedUpdates[rawUpdate.table] ??= [];
+      groupedUpdates[rawUpdate.table].push(rawUpdate);
+    }
+
+    const result: BatchedUpdateNotification = {
+      tables: Object.keys(groupedUpdates),
+      rawUpdates,
+      groupedUpdates
+    };
+
+    this.committedUpdates.length = 0;
+
+    return result;
   }
 
   installUpdateHooks() {
-    this.db.updateHook((_op: string, _dbName: string, tableName: string, _rowid: bigint) => {
-      this.uncommittedUpdatedTables.add(tableName);
-    });
+    this.db.updateHook(
+      (
+        operation: 'SQLITE_INSERT' | 'SQLITE_UPDATE' | 'SQLITE_DELETE',
+        _dbName: string,
+        table: string,
+        rowId: number
+      ) => {
+        let opType: RowUpdateType;
+        switch (operation) {
+          case 'SQLITE_INSERT':
+            opType = RowUpdateType.SQLITE_INSERT;
+            break;
+          case 'SQLITE_UPDATE':
+            opType = RowUpdateType.SQLITE_UPDATE;
+            break;
+          case 'SQLITE_DELETE':
+            opType = RowUpdateType.SQLITE_DELETE;
+            break;
+        }
 
-    this.db.commitHook(() => {
-      for (const tableName of this.uncommittedUpdatedTables) {
-        this.committedUpdatedTables.add(tableName);
+        this.uncommittedUpdates.push({ table, opType, rowId });
       }
-      this.uncommittedUpdatedTables.clear();
+    );
+
+    this.db.commitHook(() => {
+      this.committedUpdates.push(...this.uncommittedUpdates);
+      this.uncommittedUpdates.length = 0;
       return true;
     });
 
     this.db.rollbackHook(() => {
-      this.uncommittedUpdatedTables.clear();
+      this.uncommittedUpdates.length = 0;
     });
   }
 

packages/node/tests/PowerSyncDatabase.test.ts

@@ -1,9 +1,8 @@
 import * as path from 'node:path';
 import { Worker } from 'node:worker_threads';
-
 import { vi, expect, test } from 'vitest';
 import { AppSchema, databaseTest, tempDirectoryTest } from './utils';
-import { PowerSyncDatabase } from '../lib';
+import { PowerSyncDatabase, RowUpdateType, WatchOnChangeEvent } from '../lib';
 import { WorkerOpener } from '../lib/db/options';
 
 test('validates options', async () => {
@@ -65,28 +64,46 @@ databaseTest('runs queries on multiple threads', async ({ database }) => {
 databaseTest('can watch tables', async ({ database }) => {
   const fn = vi.fn();
   const disposeWatch = database.onChangeWithCallback(
-    {
-      onChange: () => {
-        fn();
-      }
-    },
+    { onChange: (event: WatchOnChangeEvent) => fn(event) },
     { tables: ['todos'], throttleMs: 0 }
   );
 
   await database.execute('INSERT INTO todos (id, content) VALUES (uuid(), ?)', ['first']);
   await expect.poll(() => fn).toHaveBeenCalledOnce();
+  expect(fn).toHaveBeenNthCalledWith(1, {
+    changedTables: ['ps_data__todos'],
+    update: {
+      tables: ['ps_data__todos'],
+      rawUpdates: [{ table: 'ps_data__todos', opType: RowUpdateType.SQLITE_INSERT, rowId: 1n }],
+      groupedUpdates: {
+        ps_data__todos: [{ table: 'ps_data__todos', opType: RowUpdateType.SQLITE_INSERT, rowId: 1n }]
+      }
+    }
+  });
 
   await database.writeTransaction(async (tx) => {
     await tx.execute('INSERT INTO todos (id, content) VALUES (uuid(), ?)', ['second']);
   });
   await expect.poll(() => fn).toHaveBeenCalledTimes(2);
+  expect(fn).toHaveBeenNthCalledWith(2, {
+    changedTables: ['ps_data__todos'],
+    update: {
+      tables: ['ps_data__todos'],
+      rawUpdates: [{ table: 'ps_data__todos', opType: RowUpdateType.SQLITE_INSERT, rowId: 2n }],
+      groupedUpdates: {
+        ps_data__todos: [{ table: 'ps_data__todos', opType: RowUpdateType.SQLITE_INSERT, rowId: 2n }]
+      }
+    }
+  });
 
+  // Assert that rolled back changes aren't emitted
   await database.writeTransaction(async (tx) => {
     await tx.execute('DELETE FROM todos;');
     await tx.rollback();
   });
   await expect.poll(() => fn).toHaveBeenCalledTimes(2);
 
+  // Assert that unwatched changes aren't emitted
   disposeWatch();
   await database.execute('INSERT INTO todos (id, content) VALUES (uuid(), ?)', ['fourth']);
   await expect.poll(() => fn).toHaveBeenCalledTimes(2);