Try testing sync with mock fetch

simolus3 · simolus3 · commit 4164c4147caf · 2025-04-01T13:26:37.000+02:00
diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts
@@ -32,7 +32,7 @@ import {
   type PowerSyncConnectionOptions,
   type RequiredAdditionalConnectionOptions
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
-import { FULL_SYNC_PRIORITY } from 'src/db/crud/SyncProgress.js';
+import { FULL_SYNC_PRIORITY } from '../db/crud/SyncProgress.js';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
diff --git a/packages/node/package.json b/packages/node/package.json
@@ -54,9 +54,10 @@
     "comlink": "^4.4.2"
   },
   "devDependencies": {
+    "@powersync/drizzle-driver": "workspace:*",
     "@types/async-lock": "^1.4.0",
     "drizzle-orm": "^0.35.2",
-    "@powersync/drizzle-driver": "workspace:*",
+    "fetch-mock": "^12.5.2",
     "rollup": "4.14.3",
     "typescript": "^5.5.3",
     "vitest": "^3.0.5"
diff --git a/packages/node/src/db/PowerSyncDatabase.ts b/packages/node/src/db/PowerSyncDatabase.ts
@@ -1,8 +1,10 @@
 import {
   AbstractPowerSyncDatabase,
+  AbstractRemoteOptions,
   AbstractStreamingSyncImplementation,
   BucketStorageAdapter,
   DBAdapter,
+  DEFAULT_REMOTE_LOGGER,
   PowerSyncBackendConnector,
   PowerSyncDatabaseOptions,
   PowerSyncDatabaseOptionsWithSettings,
@@ -18,6 +20,12 @@ import { NodeSQLOpenOptions } from './options.js';
 
 export type NodePowerSyncDatabaseOptions = PowerSyncDatabaseOptions & {
   database: DBAdapter | SQLOpenFactory | NodeSQLOpenOptions;
+  /**
+   * Options to override how the SDK will connect to the sync service.
+   * 
+   * This option is intended to be used for internal tests.
+   */
+  remoteOptions?: Partial<AbstractRemoteOptions>,
 };
 
 /**
@@ -57,7 +65,11 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
   protected generateSyncStreamImplementation(
     connector: PowerSyncBackendConnector
   ): AbstractStreamingSyncImplementation {
-    const remote = new NodeRemote(connector);
+    const remote = new NodeRemote(
+      connector,
+      DEFAULT_REMOTE_LOGGER,
+      (this.options as NodePowerSyncDatabaseOptions).remoteOptions,
+    );
 
     return new NodeStreamingSyncImplementation({
       adapter: this.bucketStorageAdapter,
diff --git a/packages/node/src/sync/stream/NodeRemote.ts b/packages/node/src/sync/stream/NodeRemote.ts
@@ -28,8 +28,8 @@ export class NodeRemote extends AbstractRemote {
     options?: Partial<AbstractRemoteOptions>
   ) {
     super(connector, logger, {
+      fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider(),
       ...(options ?? {}),
-      fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider()
     });
   }
 
diff --git a/packages/node/tests/PowerSyncDatabase.test.ts b/packages/node/tests/PowerSyncDatabase.test.ts
@@ -1,5 +1,4 @@
 import * as path from 'node:path';
-import * as fs from 'node:fs/promises';
 import { Worker } from 'node:worker_threads';
 
 import { vi, expect, test } from 'vitest';
diff --git a/packages/node/tests/sync.test.ts b/packages/node/tests/sync.test.ts
@@ -0,0 +1,102 @@
+import { describe, vi, expect, beforeEach } from 'vitest';
+
+import { connectedDatabaseTest, MockSyncService, TestConnector, waitForSyncStatus } from "./utils";
+import { AbstractPowerSyncDatabase, BucketChecksum, OplogEntryJSON } from '@powersync/common';
+
+describe('Sync', () => {
+    describe('reports progress', () => {
+        let lastOpId = 0;
+
+        beforeEach(() => {
+            lastOpId = 0;
+        });
+
+        function pushDataLine(service: MockSyncService, bucket: string, amount: number) {
+            const data: OplogEntryJSON[] = [];
+            for (let i = 0; i < amount; i++) {
+                data.push({
+                    op_id: `${++lastOpId}`,
+                    op: 'PUT',
+                    object_type: bucket,
+                    object_id: `${lastOpId}`,
+                    checksum: 0,
+                    data: '{}',
+                });
+            }
+            
+            service.pushLine({data: {
+                bucket,
+                data,
+            }});
+        }
+
+        function pushCheckpointComplete(service: MockSyncService, priority?: number) {
+            if (priority != null) {
+                service.pushLine({
+                    partial_checkpoint_complete: {
+                        last_op_id: `${lastOpId}`,
+                        priority,
+                    },
+                });
+            } else {
+                service.pushLine({
+                    checkpoint_complete: {
+                        last_op_id: `${lastOpId}`,
+                    },
+                });
+            }
+        }
+
+        connectedDatabaseTest('without priorities', async ({database, syncService}) => {
+            await database.connect(new TestConnector());
+            await vi.waitFor(() => expect(syncService.connectedListeners).toBe(1));
+
+            syncService.pushLine({checkpoint: {
+                last_op_id: '10',
+                buckets: [bucket('a', 10)]
+            }});
+
+            await waitForProgress(database, [0, 10]);
+
+            pushDataLine(syncService, 'a', 10);
+            await waitForProgress(database, [10, 10]);
+
+            pushCheckpointComplete(syncService);
+            await waitForSyncStatus(database, (s) => s.downloadProgress == null);
+
+            // Emit new data, progress should be 0/2 instead of 10/12
+            syncService.pushLine({checkpoint_diff: {
+                last_op_id: '12',
+                updated_buckets: [bucket('a', 12)],
+                removed_buckets: [],
+                write_checkpoint: '',
+            }});
+            await waitForProgress(database, [0, 2]);
+            pushDataLine(syncService, 'a', 2);
+            await waitForProgress(database, [2, 2]);
+            pushCheckpointComplete(syncService);
+            await waitForSyncStatus(database, (s) => s.downloadProgress == null);
+        });
+    });
+});
+
+function bucket(name: string, count: number, priority: number = 3): BucketChecksum {
+  return {
+    bucket: name,
+    count,
+    checksum: 0,
+    priority,
+  };
+}
+
+async function waitForProgress(database: AbstractPowerSyncDatabase, total: [number, number]) {
+    await waitForSyncStatus(database, (status) => {
+        const progress = status.downloadProgress;
+        if (!progress) {
+            return false;
+        }
+
+        const untilCompletion = progress.untilCompletion;
+        return untilCompletion.completed == total[0] && untilCompletion.total == total[1];
+    });
+}
diff --git a/packages/node/tests/utils.ts b/packages/node/tests/utils.ts
@@ -1,8 +1,9 @@
 import os from 'node:os';
 import fs from 'node:fs/promises';
 import path from 'node:path';
+import { defaultFetchMockConfig, FetchMock } from 'fetch-mock';
 import { test } from 'vitest';
-import { column, PowerSyncDatabase, Schema, Table } from '../lib';
+import { AbstractPowerSyncDatabase, AbstractRemoteOptions, column, NodePowerSyncDatabaseOptions, PowerSyncBackendConnector, PowerSyncCredentials, PowerSyncDatabase, Schema, StreamingSyncLine, SyncStatus, Table } from '../lib';
 
 export async function createTempDir() {
   const ostmpdir = os.tmpdir();
@@ -37,15 +38,115 @@ export const tempDirectoryTest = test.extend<{ tmpdir: string }>({
   }
 });
 
-export const databaseTest = tempDirectoryTest.extend<{ database: PowerSyncDatabase }>({
-  database: async ({ tmpdir }, use) => {
+function createDatabaseFixture(options: Partial<NodePowerSyncDatabaseOptions> = {}) {
+  return async ({ tmpdir }, use) => {
     const database = new PowerSyncDatabase({
+      ...options,
       schema: AppSchema,
       database: {
         dbFilename: 'test.db',
         dbLocation: tmpdir
       }
     });
     await use(database);
-  }
+    await database.close();
+  };
+}
+
+export const databaseTest = tempDirectoryTest.extend<{ database: PowerSyncDatabase }>({
+  database: async ({tmpdir}, use) => {
+    await createDatabaseFixture()({tmpdir}, use);
+  },
+});
+
+// TODO: Unify this with the test setup for the web SDK.
+export const mockSyncServiceTest = tempDirectoryTest.extend<{syncService: MockSyncService}>({
+  syncService: async ({}, use) => {
+    // Don't install global fetch mocks, we want tests to be isolated!
+    const fetchMock = new FetchMock(defaultFetchMockConfig);
+    const listeners: ReadableStreamDefaultController<StreamingSyncLine>[] = [];
+
+    fetchMock.route('path:/sync/stream', async () => {
+      let thisController: ReadableStreamDefaultController<StreamingSyncLine> | null = null;
+
+      const syncLines = new ReadableStream<StreamingSyncLine>({
+        start(controller) {
+          thisController = controller;
+          listeners.push(controller);
+        },
+        cancel() {
+          listeners.splice(listeners.indexOf(thisController!), 1);
+        },
+      });
+
+
+      const encoder = new TextEncoder();
+      const asLines = new TransformStream<StreamingSyncLine, Uint8Array>({
+        transform: (chunk, controller) => {
+          const line = `${JSON.stringify(chunk)}\n`;
+          controller.enqueue(encoder.encode(line));
+        },
+      });
+
+      return new Response(syncLines.pipeThrough(asLines), {status: 200});
+    });
+    fetchMock.catch(404);
+
+    await use({
+      clientOptions: {
+        fetchImplementation: fetchMock.fetchHandler.bind(fetchMock),
+      },
+      get connectedListeners() {
+        return listeners.length;
+      },
+      pushLine(line) {
+        for (const listener of listeners) {
+          listener.enqueue(line);
+        }
+      },
+    });
+  },
+});
+
+export const connectedDatabaseTest = mockSyncServiceTest.extend<{ database: PowerSyncDatabase }>({
+  database: async ({ tmpdir, syncService }, use) => {
+    const fixture = createDatabaseFixture({remoteOptions: syncService.clientOptions});
+    await fixture({ tmpdir }, use);
+  },
 });
+
+export interface MockSyncService {
+  clientOptions: Partial<AbstractRemoteOptions>,
+  pushLine: (line: StreamingSyncLine) => void,
+  connectedListeners: number,
+}
+
+export class TestConnector implements PowerSyncBackendConnector {
+  async fetchCredentials(): Promise<PowerSyncCredentials> {
+    return {
+      endpoint: '',
+      token: ''
+    };
+  }
+  async uploadData(database: AbstractPowerSyncDatabase): Promise<void> {
+    const tx = await database.getNextCrudTransaction();
+    await tx?.complete();
+  }
+}
+
+export function waitForSyncStatus(database: AbstractPowerSyncDatabase, matcher: (status: SyncStatus) => boolean): Promise<void> {
+  return new Promise((resolve) => {
+    if (matcher(database.currentStatus)) {
+      return resolve();
+    }
+
+    const dispose = database.registerListener({
+      statusChanged: (status) => {
+        if (matcher(status)) {
+          dispose();
+          resolve();
+        }
+      }
+    });
+  });
+}
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml

Original file line number	Diff line number	Diff line change
`@@ -28,8 +28,8 @@ export class NodeRemote extends AbstractRemote {`
`28`	`28`	`options?: Partial<AbstractRemoteOptions>`
`29`	`29`	`) {`
`30`	`30`	`super(connector, logger, {`
	`31`	`+ fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider(),`
`31`	`32`	`...(options ?? {}),`
`32`		`- fetchImplementation: options?.fetchImplementation ?? new NodeFetchProvider()`
`33`	`33`	`});`
`34`	`34`	`}`
`35`	`35`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,4 @@`
`1`	`1`	`import * as path from 'node:path';`
`2`		`-import * as fs from 'node:fs/promises';`
`3`	`2`	`import { Worker } from 'node:worker_threads';`
`4`	`3`
`5`	`4`	`import { vi, expect, test } from 'vitest';`