fix: prevent multiple web workers from starting (#490)

Author: filipecabaco

src/RealtimeClient.ts

@@ -562,36 +562,47 @@ export default class RealtimeClient {
     this.flushSendBuffer()
     this.reconnectTimer.reset()
     if (!this.worker) {
-      this.heartbeatTimer && clearInterval(this.heartbeatTimer)
-      this.heartbeatTimer = setInterval(
-        () => this.sendHeartbeat(),
-        this.heartbeatIntervalMs
-      )
+      this._startHeartbeat()
     } else {
-      if (this.workerUrl) {
-        this.log('worker', `starting worker for from ${this.workerUrl}`)
-      } else {
-        this.log('worker', `starting default worker`)
-      }
-      const objectUrl = this._workerObjectUrl(this.workerUrl!)
-      this.workerRef = new Worker(objectUrl)
-      this.workerRef.onerror = (error) => {
-        this.log('worker', 'worker error', (error as ErrorEvent).message)
-        this.workerRef!.terminate()
-      }
-      this.workerRef.onmessage = (event) => {
-        if (event.data.event === 'keepAlive') {
-          this.sendHeartbeat()
-        }
+      if (!this.workerRef) {
+        this._startWorkerHeartbeat()
       }
-      this.workerRef.postMessage({
-        event: 'start',
-        interval: this.heartbeatIntervalMs,
-      })
     }
+
     this.stateChangeCallbacks.open.forEach((callback) => callback())
   }
+  /** @internal */
+  private _startHeartbeat() {
+    this.heartbeatTimer && clearInterval(this.heartbeatTimer)
+    this.heartbeatTimer = setInterval(
+      () => this.sendHeartbeat(),
+      this.heartbeatIntervalMs
+    )
+  }
 
+  /** @internal */
+  private _startWorkerHeartbeat() {
+    if (this.workerUrl) {
+      this.log('worker', `starting worker for from ${this.workerUrl}`)
+    } else {
+      this.log('worker', `starting default worker`)
+    }
+    const objectUrl = this._workerObjectUrl(this.workerUrl!)
+    this.workerRef = new Worker(objectUrl)
+    this.workerRef.onerror = (error) => {
+      this.log('worker', 'worker error', (error as ErrorEvent).message)
+      this.workerRef!.terminate()
+    }
+    this.workerRef.onmessage = (event) => {
+      if (event.data.event === 'keepAlive') {
+        this.sendHeartbeat()
+      }
+    }
+    this.workerRef.postMessage({
+      event: 'start',
+      interval: this.heartbeatIntervalMs,
+    })
+  }
   /** @internal */
   private _onConnClose(event: any) {
     this.log('transport', 'close', event)

test/channel.test.ts renamed to test/RealtimeChannel.test.ts

@@ -1,21 +1,11 @@
 import assert from 'assert'
 import sinon from 'sinon'
 import crypto from 'crypto'
-import {
-  describe,
-  beforeEach,
-  afterEach,
-  test,
-  beforeAll,
-  afterAll,
-  vi,
-  it,
-} from 'vitest'
+import { describe, beforeEach, afterEach, test, vi } from 'vitest'
 
 import RealtimeClient from '../src/RealtimeClient'
 import RealtimeChannel from '../src/RealtimeChannel'
 import { Response } from '@supabase/node-fetch'
-import Worker from 'web-worker'
 import { Server, WebSocket } from 'mock-socket'
 import { CHANNEL_STATES } from '../src/lib/constants'
 import Push from '../src/lib/push'
@@ -1416,37 +1406,3 @@ describe('trigger', () => {
     assert.equal(client.accessTokenValue, '123')
   })
 })
-
-describe('worker', () => {
-  let client: RealtimeClient
-  let mockServer: Server
-
-  beforeAll(() => {
-    window.Worker = Worker
-    projectRef = randomProjectRef()
-    url = `wss://${projectRef}/socket`
-    mockServer = new Server(url)
-  })
-
-  afterAll(() => {
-    // @ts-ignore - Deliberately removing Worker to clean up test environment
-    window.Worker = undefined
-    mockServer.close()
-  })
-
-  beforeEach(() => {
-    client = new RealtimeClient('ws://localhost:8080/socket', {
-      worker: true,
-      workerUrl: 'https://realtime.supabase.com/worker.js',
-      heartbeatIntervalMs: 10,
-    })
-  })
-
-  test('sets worker flag', () => {
-    assert.ok(client.worker)
-  })
-
-  test('sets worker URL', () => {
-    assert.equal(client.workerUrl, 'https://realtime.supabase.com/worker.js')
-  })
-})

test/client.test.ts renamed to test/RealtimeClient.test.ts

@@ -1,23 +1,30 @@
 import assert from 'assert'
-import { describe, beforeEach, afterEach, test, vi, expect } from 'vitest'
+import {
+  afterAll,
+  afterEach,
+  beforeAll,
+  beforeEach,
+  describe,
+  expect,
+  test,
+  vi,
+} from 'vitest'
 import { Server, WebSocket as MockWebSocket } from 'mock-socket'
 import WebSocket from 'ws'
 import sinon from 'sinon'
 import crypto from 'crypto'
-
-import RealtimeClient, {
-  HeartbeatStatus,
-  RealtimeMessage,
-} from '../src/RealtimeClient'
+import RealtimeClient, { HeartbeatStatus } from '../src/RealtimeClient'
 import jwt from 'jsonwebtoken'
 import { CHANNEL_STATES } from '../src/lib/constants'
+import path from 'path'
 
 function generateJWT(exp: string): string {
   return jwt.sign({}, 'your-256-bit-secret', {
     algorithm: 'HS256',
     expiresIn: exp || '1h',
   })
 }
+import Worker from 'web-worker'
 
 let socket: RealtimeClient
 let randomProjectRef = () => crypto.randomUUID()
@@ -853,3 +860,44 @@ describe('log operations', () => {
     )
   })
 })
+
+describe('worker', () => {
+  let mockServer: Server
+  let client: RealtimeClient
+  const workerPath = path.join(__dirname, 'test_worker.js')
+  beforeAll(() => {
+    window.Worker = Worker
+    projectRef = randomProjectRef()
+    url = `wss://${projectRef}/socket`
+    mockServer = new Server(url)
+  })
+
+  afterAll(() => {
+    // @ts-ignore - Deliberately removing Worker to clean up test environment
+    window.Worker = undefined
+    mockServer.close()
+  })
+
+  beforeEach(() => {
+    client = new RealtimeClient('ws://localhost:8080/socket', {
+      worker: true,
+      workerUrl: workerPath,
+      heartbeatIntervalMs: 10,
+    })
+  })
+  test('sets worker flag', () => {
+    assert.ok(client.worker)
+  })
+
+  test('sets worker URL', () => {
+    assert.equal(client.workerUrl, workerPath)
+  })
+
+  test('ensures single worker ref is started even with multiple connect calls', () => {
+    client._onConnOpen()
+    let ref = client.workerRef
+
+    client._onConnOpen()
+    assert.ok(ref === client.workerRef)
+  })
+})

test/presence.test.ts renamed to test/RealtimePresence.test.ts

@@ -0,0 +1,5 @@
+addEventListener('message', (e) => {
+  if (e.data.event === 'start') {
+    setInterval(() => postMessage({ event: 'keepAlive' }), e.data.interval)
+  }
+})