Update tests for shared web workers

Author: stevensJourney

packages/web/src/db/PowerSyncDatabase.ts

@@ -1,32 +1,31 @@
 import {
-  type BucketStorageAdapter,
-  type PowerSyncBackendConnector,
-  type PowerSyncCloseOptions,
-  type RequiredAdditionalConnectionOptions,
   AbstractPowerSyncDatabase,
   DBAdapter,
-  DEFAULT_POWERSYNC_CLOSE_OPTIONS,
-  isDBAdapter,
-  isSQLOpenFactory,
   PowerSyncDatabaseOptions,
   PowerSyncDatabaseOptionsWithDBAdapter,
   PowerSyncDatabaseOptionsWithOpenFactory,
   PowerSyncDatabaseOptionsWithSettings,
   SqliteBucketStorage,
-  StreamingSyncImplementation
+  StreamingSyncImplementation,
+  isDBAdapter,
+  isSQLOpenFactory,
+  type BucketStorageAdapter,
+  type PowerSyncBackendConnector,
+  type PowerSyncCloseOptions,
+  type RequiredAdditionalConnectionOptions
 } from '@powersync/common';
 import { Mutex } from 'async-mutex';
 import { getNavigatorLocks } from '../shared/navigator';
+import { WebDBAdapter } from './adapters/WebDBAdapter';
 import { WASQLiteOpenFactory } from './adapters/wa-sqlite/WASQLiteOpenFactory';
 import {
   DEFAULT_WEB_SQL_FLAGS,
   ResolvedWebSQLOpenOptions,
-  resolveWebSQLFlags,
-  WebSQLFlags
+  WebSQLFlags,
+  resolveWebSQLFlags
 } from './adapters/web-sql-flags';
-import { WebDBAdapter } from './adapters/WebDBAdapter';
-import { SharedWebStreamingSyncImplementation } from './sync/SharedWebStreamingSyncImplementation';
 import { SSRStreamingSyncImplementation } from './sync/SSRWebStreamingSyncImplementation';
+import { SharedWebStreamingSyncImplementation } from './sync/SharedWebStreamingSyncImplementation';
 import { WebRemote } from './sync/WebRemote';
 import {
   WebStreamingSyncImplementation,
@@ -160,14 +159,13 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
    * By default the sync stream client is only disconnected if
    * multiple tabs are not enabled.
    */
-  close(options: PowerSyncCloseOptions = DEFAULT_POWERSYNC_CLOSE_OPTIONS): Promise<void> {
+  close(options?: PowerSyncCloseOptions): Promise<void> {
     if (this.unloadListener) {
       window.removeEventListener('unload', this.unloadListener);
     }
-
     return super.close({
       // Don't disconnect by default if multiple tabs are enabled
-      disconnect: options.disconnect ?? !this.resolvedFlags.enableMultiTabs
+      disconnect: options?.disconnect ?? !this.resolvedFlags.enableMultiTabs
     });
   }
 

packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -255,8 +255,6 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
   async dispose(): Promise<void> {
     await this.waitForReady();
 
-    await super.dispose();
-
     await new Promise<void>((resolve) => {
       // Listen for the close acknowledgment from the worker
       this.messagePort.addEventListener('message', (event) => {
@@ -273,6 +271,9 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
       };
       this.messagePort.postMessage(closeMessagePayload);
     });
+
+    await super.dispose();
+
     this.abortOnClose.abort();
 
     // Release the proxy

packages/web/src/worker/sync/MockSyncServiceWorker.ts

@@ -15,7 +15,13 @@ export class MockSyncService {
    * Register a new pending request (called by WebRemote when a sync stream is requested).
    * Returns a promise that resolves when a client creates a response for this request.
    */
-  registerPendingRequest(url: string, method: string, headers: Record<string, string>, body: any): Promise<Response> {
+  registerPendingRequest(
+    url: string,
+    method: string,
+    headers: Record<string, string>,
+    body: any,
+    signal?: AbortSignal
+  ): Promise<Response> {
     const id = `pending-${++this.nextId}`;
 
     let resolveResponse: (response: Response) => void;
@@ -40,6 +46,20 @@ export class MockSyncService {
 
     this.pendingRequests.set(id, pendingRequest);
 
+    signal?.addEventListener('abort', () => {
+      this.pendingRequests.delete(id);
+      rejectResponse(new Error('Request aborted'));
+
+      // if already in active responses, remove it
+      if (this.activeResponses.has(id)) {
+        const response = this.activeResponses.get(id);
+        if (response) {
+          response.stream.close();
+        }
+        this.activeResponses.delete(id);
+      }
+    });
+
     // Return the promise - it will resolve when createResponse is called
     return responsePromise;
   }

packages/web/tests/mockSyncServiceExample.test.ts

@@ -17,28 +17,9 @@ describe('Mock Sync Service Example', { timeout: 100000 }, () => {
   sharedMockSyncServiceTest(
     'should allow mocking sync responses in shared worker',
     { timeout: 100000 },
-    async ({ database, connect }) => {
+    async ({ context: { database, connect } }) => {
       // Call connect to start the sync worker and get the sync service
-      const { syncService, connectionPromise } = await connect();
-
-      // Wait for a pending request to appear
-      let pendingRequestId: string;
-      await vi.waitFor(async () => {
-        const requests = await syncService.getPendingRequests();
-        expect(requests.length).toBeGreaterThan(0);
-        pendingRequestId = requests[0].id;
-      });
-
-      // Get the pending request to inspect it
-      const requests = await syncService.getPendingRequests();
-      expect(requests).toHaveLength(1);
-      expect(requests[0].url).toContain('/sync/stream');
-      expect(requests[0].method).toBe('POST');
-      expect(requests[0].headers).toBeDefined();
-      expect(requests[0].body).toBeDefined();
-
-      // Create a response for the pending request
-      await syncService.createResponse(pendingRequestId!, 200, { 'Content-Type': 'application/json' });
+      const { syncService, syncRequestId } = await connect();
 
       // Push a checkpoint with buckets (following node test pattern)
       const checkpoint: StreamingSyncCheckpoint = {
@@ -56,13 +37,10 @@ describe('Mock Sync Service Example', { timeout: 100000 }, () => {
         }
       };
 
-      await syncService.pushBodyLine(pendingRequestId!, checkpoint);
+      await syncService.pushBodyLine(syncRequestId, checkpoint);
 
       // The connect call should resolve by now
-      await connectionPromise;
-
-      // Push data line with PUT operation to save data (following node test pattern)
-      await syncService.pushBodyLine(pendingRequestId!, {
+      await syncService.pushBodyLine(syncRequestId, {
         data: {
           bucket: 'a',
           data: [
@@ -79,14 +57,14 @@ describe('Mock Sync Service Example', { timeout: 100000 }, () => {
       });
 
       // Push checkpoint_complete to finish the sync
-      await syncService.pushBodyLine(pendingRequestId!, {
+      await syncService.pushBodyLine(syncRequestId, {
         checkpoint_complete: {
           last_op_id: '1'
         }
       });
 
       // Complete the response
-      await syncService.completeResponse(pendingRequestId!);
+      await syncService.completeResponse(syncRequestId);
 
       // Wait for sync to complete and verify the data was saved
       await vi.waitFor(async () => {

packages/web/tests/mocks/MockWebRemote.ts

@@ -10,7 +10,7 @@ import {
   RemoteConnector,
   SocketSyncStreamOptions
 } from '@powersync/common';
-import type { BSON } from 'bson';
+import { serialize, type BSON } from 'bson';
 import { getMockSyncService, setupMockServiceMessageHandler } from '../utils/MockSyncService';
 
 /**
@@ -77,7 +77,7 @@ class MockSyncServiceFetchProvider extends FetchImplementationProvider {
         });
 
         // Register as a pending request and wait for client to create response
-        return await mockService.registerPendingRequest(request.url, request.method, headers, body);
+        return await mockService.registerPendingRequest(request.url, request.method, headers, body, request.signal);
       }
 
       // Fallback if mock service is not available
@@ -189,6 +189,6 @@ export class WebRemote extends AbstractRemote {
     bson?: typeof BSON
   ): Promise<DataStream<T>> {
     // postStreamRaw decodes to strings, so convert back to Uint8Array for the map function
-    return await this.postStreamRaw(options, (line: string) => map(new TextEncoder().encode(line)));
+    return await this.postStreamRaw(options, (line: string) => map(serialize(JSON.parse(line))));
   }
 }