Add getCrudTransactions()

Author: simolus3

.changeset/nice-dragons-smile.md

@@ -0,0 +1,8 @@
+---
+'@powersync/common': minor
+'@powersync/node': minor
+'@powersync/react-native': minor
+'@powersync/web': minor
+---
+
+Add `getCrudTransactions()`, returning an async iterator of transactions. This can be used to batch transactions when uploading CRUD data.

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -632,35 +632,72 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * @returns A transaction of CRUD operations to upload, or null if there are none
    */
   async getNextCrudTransaction(): Promise<CrudTransaction | null> {
-    return await this.readTransaction(async (tx) => {
-      const first = await tx.getOptional<CrudEntryJSON>(
-        `SELECT id, tx_id, data FROM ${PSInternalTable.CRUD} ORDER BY id ASC LIMIT 1`
-      );
+    for await (const transaction of this.getCrudTransactions()) {
+      return transaction;
+    }
 
-      if (!first) {
-        return null;
-      }
-      const txId = first.tx_id;
+    return null;
+  }
 
-      let all: CrudEntry[];
-      if (!txId) {
-        all = [CrudEntry.fromRow(first)];
-      } else {
-        const result = await tx.getAll<CrudEntryJSON>(
-          `SELECT id, tx_id, data FROM ${PSInternalTable.CRUD} WHERE tx_id = ? ORDER BY id ASC`,
-          [txId]
-        );
-        all = result.map((row) => CrudEntry.fromRow(row));
+  /**
+   * Returns an async iterator of completed transactions with local writes against the database.
+   *
+   * This is typically used from the {@link PowerSyncBackendConnector.uploadData} callback. Each entry emitted by the
+   * returned flow is a full transaction containing all local writes made while that transaction was active.
+   *
+   * Unlike {@link getNextCrudTransaction}, which always returns the oldest transaction that hasn't been
+   * {@link CrudTransaction.complete}d yet, this flow can be used to collect multiple transactions. Calling
+   * {@link CrudTransaction.complete} will mark _all_ transactions emitted by the flow until that point as completed.
+   *
+   * This can be used to upload multiple transactions in a single batch, e.g with:
+   *
+   * ```TypeScript
+   * let lastTransaction: CrudTransaction | null = null;
+   * let batch: CrudEntry[] = [];
+   *
+   * for await (const transaction of database.getCrudTransactions()) {
+   *   batch.push(...transaction.crud);
+   *   lastTransaction = transaction;
+   *
+   *   if (batch.length > 10) {
+   *     break;
+   *    }
+   * }
+   * ```
+   *
+   * If there is no local data to upload, the async iterator complete without emitting any items.
+   */
+  async *getCrudTransactions(): AsyncIterable<CrudTransaction> {
+    let lastCrudItemId = -1;
+    const sql = `
+WITH RECURSIVE crud_entries AS (
+  SELECT id, tx_id, data FROM ps_crud WHERE id = (SELECT min(id) FROM ps_crud WHERE id > ?)
+  UNION ALL
+  SELECT ps_crud.id, ps_crud.tx_id, ps_crud.data FROM ps_crud
+    INNER JOIN crud_entries ON crud_entries.id + 1 = rowid
+  WHERE crud_entries.tx_id = ps_crud.tx_id
+)
+SELECT * FROM crud_entries;
+    `;
+
+    while (true) {
+      const nextTransaction = await this.database.getAll<CrudEntryJSON>(sql, [lastCrudItemId]);
+      if (nextTransaction.length == 0) {
+        break;
       }
 
-      const last = all[all.length - 1];
+      const items = nextTransaction.map((row) => CrudEntry.fromRow(row));
+      const last = items[items.length - 1];
+      const txId = last.transactionId;
 
-      return new CrudTransaction(
-        all,
+      yield new CrudTransaction(
+        items,
         async (writeCheckpoint?: string) => this.handleCrudCheckpoint(last.clientId, writeCheckpoint),
         txId
       );
-    });
+
+      lastCrudItemId = last.clientId;
+    }
   }
 
   /**

packages/node/tests/PowerSyncDatabase.test.ts

@@ -3,7 +3,7 @@ import { Worker } from 'node:worker_threads';
 
 import { vi, expect, test } from 'vitest';
 import { AppSchema, databaseTest, tempDirectoryTest } from './utils';
-import { PowerSyncDatabase } from '../lib';
+import { CrudEntry, CrudTransaction, PowerSyncDatabase } from '../lib';
 import { WorkerOpener } from '../lib/db/options';
 
 test('validates options', async () => {
@@ -131,3 +131,39 @@ databaseTest.skip('can watch queries', async ({ database }) => {
   await database.execute('INSERT INTO todos (id, content) VALUES (uuid(), ?)', ['fourth']);
   expect((await query.next()).value.rows).toHaveLength(4);
 });
+
+databaseTest('getCrudTransactions', async ({ database }) => {
+  async function createTransaction(amount: number) {
+    await database.writeTransaction(async (tx) => {
+      for (let i = 0; i < amount; i++) {
+        await tx.execute('insert into todos (id) values (uuid())');
+      }
+    });
+  }
+
+  let iterator = database.getCrudTransactions()[Symbol.asyncIterator]();
+  expect(await iterator.next()).toMatchObject({ done: true });
+
+  await createTransaction(5);
+  await createTransaction(10);
+  await createTransaction(15);
+
+  let lastTransaction: CrudTransaction | null = null;
+  let batch: CrudEntry[] = [];
+
+  // Take the first two transactions via the async generator.
+  for await (const transaction of database.getCrudTransactions()) {
+    batch.push(...transaction.crud);
+    lastTransaction = transaction;
+
+    if (batch.length > 10) {
+      break;
+    }
+  }
+
+  expect(batch).toHaveLength(15);
+  await lastTransaction!.complete();
+
+  const remainingTransaction = await database.getNextCrudTransaction();
+  expect(remainingTransaction?.crud).toHaveLength(15);
+});