cleanup test code

Author: stevensJourney

packages/web/src/db/adapters/AsyncDatabaseConnection.ts

@@ -17,6 +17,11 @@ export type ProxiedQueryResult = Omit<QueryResult, 'rows'> & {
  */
 export type OnTableChangeCallback = (event: BatchedUpdateNotification) => void;
 
+/**
+ * Thrown when an underlying database connection is closed.
+ * This is particularly relevant when worker connections are marked as closed while
+ * operations are still in progress.
+ */
 export class ConnectionClosedError extends Error {
   constructor(message: string) {
     super(message);

packages/web/src/db/adapters/LockedAsyncDatabaseAdapter.ts

@@ -57,7 +57,6 @@ export class LockedAsyncDatabaseAdapter
   private _config: ResolvedWebSQLOpenOptions | null = null;
   protected pendingAbortControllers: Set<AbortController>;
   protected requiresHolds: boolean | null;
-  protected requiresReOpen: boolean;
   protected databaseOpenPromise: Promise<void> | null = null;
 
   closing: boolean;
@@ -71,7 +70,6 @@ export class LockedAsyncDatabaseAdapter
     this.closed = false;
     this.closing = false;
     this.requiresHolds = null;
-    this.requiresReOpen = false;
     // Set the name if provided. We can query for the name if not available yet
     this.debugMode = options.debugMode ?? false;
     if (this.debugMode) {

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

@@ -17,6 +17,13 @@ export class MockSyncService {
   private nextId = 0;
   private automaticResponse: AutomaticResponseConfig | null = null;
 
+  /**
+   * A Static instance of the mock sync service.
+   * This can be used directly for non-worker environments.
+   * A proxy is required for worker environments.
+   */
+  static readonly GLOBAL_INSTANCE = new 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.
@@ -253,43 +260,19 @@ export class MockSyncService {
   }
 }
 
-/**
- * Global mock service instance (only available in shared worker context)
- */
-let globalMockService: MockSyncService | null = null;
-
-/**
- * Get or create the global mock service instance
- */
-export function getMockSyncService(): MockSyncService | null {
-  // Only available in shared worker context
-  if (typeof SharedWorkerGlobalScope === 'undefined') {
-    return null;
-  }
-
-  if (!globalMockService) {
-    globalMockService = new MockSyncService();
-  }
-
-  return globalMockService;
-}
-
 /**
  * Set up message handler for the mock service on a MessagePort
  */
 export function setupMockServiceMessageHandler(port: MessagePort) {
-  const service = getMockSyncService();
-  if (!service) {
-    return;
-  }
-
   port.addEventListener('message', (event: MessageEvent<MockSyncServiceMessage>) => {
     const message = event.data;
 
     if (!message || typeof message !== 'object' || !('type' in message)) {
       return;
     }
 
+    const service = MockSyncService.GLOBAL_INSTANCE;
+
     try {
       switch (message.type) {
         case 'getPendingRequests': {

packages/web/tests/mocks/MockWebRemote.ts

@@ -11,34 +11,7 @@ import {
   SocketSyncStreamOptions
 } from '@powersync/common';
 import { serialize, type BSON } from 'bson';
-import { getMockSyncService, setupMockServiceMessageHandler } from '../utils/MockSyncService';
-
-/**
- * Check if we're running in a shared worker context
- */
-function isSharedWorkerContext(): boolean {
-  const isSharedWorker =
-    typeof SharedWorkerGlobalScope !== 'undefined' &&
-    typeof self !== 'undefined' &&
-    (self as any).constructor?.name === 'SharedWorkerGlobalScope';
-  return isSharedWorker;
-}
-
-/**
- * Mock fetch provider that returns 401 for non-stream requests
- */
-class MockFetchProvider extends FetchImplementationProvider {
-  getFetch(): FetchImplementation {
-    // Return a mock fetch that always returns 401
-    return async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
-      const response = new Response(null, {
-        status: 401,
-        statusText: 'Unauthorized'
-      });
-      return response;
-    };
-  }
-}
+import { MockSyncService, setupMockServiceMessageHandler } from '../../src/worker/sync/MockSyncServiceWorker';
 
 /**
  * Mock fetch provider that intercepts all requests and routes them to the mock sync service.
@@ -53,96 +26,61 @@ class MockSyncServiceFetchProvider extends FetchImplementationProvider {
   getFetch(): FetchImplementation {
     return async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
       const request = new Request(input, init);
-      const mockService = getMockSyncService();
 
-      if (mockService) {
-        // Read the request body (if any)
-        let body: any = null;
-        try {
-          if (request.body) {
-            const clonedRequest = request.clone();
-            body = await clonedRequest.json().catch(() => {
-              // If JSON parsing fails, try text
-              return clonedRequest.text().catch(() => null);
-            });
-          }
-        } catch (e) {
-          // Body might not be readable, that's okay
+      const mockService = MockSyncService.GLOBAL_INSTANCE;
+
+      // Read the request body (if any)
+      let body: any = null;
+      try {
+        if (request.body) {
+          const clonedRequest = request.clone();
+          body = await clonedRequest.json().catch(() => {
+            // If JSON parsing fails, try text
+            return clonedRequest.text().catch(() => null);
+          });
         }
-
-        // Extract headers from the request
-        const headers: Record<string, string> = {};
-        request.headers.forEach((value, key) => {
-          headers[key] = value;
-        });
-
-        // Register as a pending request and wait for client to create response
-        return await mockService.registerPendingRequest(request.url, request.method, headers, body, request.signal);
+      } catch (e) {
+        // Body might not be readable, that's okay
       }
 
-      // Fallback if mock service is not available
-      return new Response(null, {
-        status: 401,
-        statusText: 'Unauthorized'
+      // Extract headers from the request
+      const headers: Record<string, string> = {};
+      request.headers.forEach((value, key) => {
+        headers[key] = value;
       });
+
+      // Register as a pending request and wait for client to create response
+      return await mockService.registerPendingRequest(request.url, request.method, headers, body, request.signal);
     };
   }
 }
 
-// Track if we've already set up the onconnect handler
-let onconnectHandlerSetup = false;
-
 /**
- * Create the mock sync service globally upfront when in shared worker context.
- * This ensures the service is available before any requests come in.
+ * Check if we're running in a shared worker context
  */
-function initializeMockSyncService() {
-  if (!isSharedWorkerContext()) {
-    return;
-  }
-
-  // Create the mock service upfront (getMockSyncService creates it if it doesn't exist)
-  getMockSyncService();
+function isSharedWorkerContext(): boolean {
+  const isSharedWorker =
+    typeof SharedWorkerGlobalScope !== 'undefined' &&
+    typeof self !== 'undefined' &&
+    (self as any).constructor?.name === 'SharedWorkerGlobalScope';
+  return isSharedWorker;
 }
 
-/**
- * Set up the onconnect listener lazily when the first WebRemote is created.
- * This ensures deterministic initialization order regardless of import order.
- * The mock service is created upfront, but message handlers are only set up on connect.
- */
-function setupOnconnectHandlerIfNeeded() {
-  if (onconnectHandlerSetup || !isSharedWorkerContext()) {
-    return;
-  }
-
-  // Create the mock sync service globally upfront
-  initializeMockSyncService();
-
+if (isSharedWorkerContext()) {
   const _self: SharedWorkerGlobalScope = self as any;
+  console.log('MockWebRemote: setting up connect listener');
 
-  // Store the original onconnect if it exists (may be set by SharedSyncImplementation.worker.ts)
-  const originalOnConnect = _self.onconnect;
-
-  _self.onconnect = async function (event: MessageEvent) {
+  /**
+   * This listener should be called in tandem with the shared sync worker's listener.
+   */
+  _self.addEventListener('connect', async function (event: MessageEvent) {
+    console.log('MockWebRemote: connect listener called');
     const port = event.ports[0];
 
-    // Get the mock service (already created upfront) and set up message handler lazily
-    const mockService = getMockSyncService();
-    if (mockService) {
-      // Set up message handler for the mock service on this port
-      // Tests can create a separate SharedWorker connection to access this
-      setupMockServiceMessageHandler(port);
-    }
-
-    // Call the original handler if it exists (this will set up WorkerClient)
-    // Note: The mock service message handler and WorkerClient can coexist
-    // since they use different message types
-    if (originalOnConnect) {
-      await originalOnConnect.call(this, event);
-    }
-  };
-
-  onconnectHandlerSetup = true;
+    // Set up message handler for the mock service on this port
+    // Tests can create a separate SharedWorker connection to access this
+    setupMockServiceMessageHandler(port);
+  });
 }
 
 export class WebRemote extends AbstractRemote {
@@ -153,12 +91,8 @@ export class WebRemote extends AbstractRemote {
     protected logger: ILogger = DEFAULT_REMOTE_LOGGER,
     options?: Partial<AbstractRemoteOptions>
   ) {
-    // Lazy initialize the onconnect handler when the first WebRemote is created
-    // This ensures deterministic initialization order regardless of import order
-    setupOnconnectHandlerIfNeeded();
-
     // Use mock service fetch provider if we're in a shared worker context
-    const fetchProvider = isSharedWorkerContext() ? new MockSyncServiceFetchProvider() : new MockFetchProvider();
+    const fetchProvider = new MockSyncServiceFetchProvider();
 
     super(connector, logger, {
       ...(options ?? {}),

packages/web/tests/multiple_instances.test.ts

@@ -5,7 +5,7 @@ import { beforeAll, describe, expect, it, onTestFinished, vi } from 'vitest';
 import { LockedAsyncDatabaseAdapter } from '../src/db/adapters/LockedAsyncDatabaseAdapter';
 import { WebDBAdapter } from '../src/db/adapters/WebDBAdapter';
 import { WorkerWrappedAsyncDatabaseConnection } from '../src/db/adapters/WorkerWrappedAsyncDatabaseConnection';
-import { getMockSyncServiceFromWorker } from './utils/MockSyncService';
+import { getMockSyncServiceFromWorker } from './utils/MockSyncServiceClient';
 import { createTestConnector, sharedMockSyncServiceTest } from './utils/mockSyncServiceTest';
 import { generateTestDb, testSchema } from './utils/testDb';