fix(realtime): ensure custom JWT token is set before channel subscriptions

Author: mandarini

packages/core/realtime-js/src/RealtimeClient.ts

@@ -186,7 +186,6 @@ export default class RealtimeClient {
     }
 
     this._setConnectionState('connecting')
-    this._setAuthSafely('connect')
 
     // Establish WebSocket connection
     if (this.transport) {
@@ -253,11 +252,13 @@ export default class RealtimeClient {
         this._setConnectionState('disconnected')
       }
 
-      // Close the WebSocket connection
-      if (code) {
-        this.conn.close(code, reason ?? '')
-      } else {
-        this.conn.close()
+      // Close the WebSocket connection if close method exists
+      if (typeof this.conn.close === 'function') {
+        if (code) {
+          this.conn.close(code, reason ?? '')
+        } else {
+          this.conn.close()
+        }
       }
 
       this._teardownConnection()
@@ -608,7 +609,11 @@ export default class RealtimeClient {
 
     // Wait for any pending auth operations before flushing send buffer
     // This ensures channel join messages include the correct access token
-    this._waitForAuthIfNeeded()
+    const authPromise =
+      this._authPromise ||
+      (this.accessToken && !this.accessTokenValue ? this.setAuth() : Promise.resolve())
+
+    authPromise
       .then(() => {
         this.flushSendBuffer()
       })

packages/core/supabase-js/test/integration.test.ts

@@ -363,31 +363,45 @@ describe('Storage API', () => {
 describe('Custom JWT', () => {
   describe('Realtime', () => {
     test('will connect with a properly signed jwt token', async () => {
-      const jwtToken = sign({ sub: '1234567890' }, JWT_SECRET, { expiresIn: '1h' })
+      const jwtToken = sign(
+        {
+          sub: '1234567890',
+          role: 'anon',
+          iss: 'supabase-demo',
+        },
+        JWT_SECRET,
+        { expiresIn: '1h' }
+      )
       const supabaseWithCustomJwt = createClient(SUPABASE_URL, ANON_KEY, {
         accessToken: () => Promise.resolve(jwtToken),
+        realtime: {
+          ...(wsTransport && { transport: wsTransport }),
+        },
       })
 
-      // Wait for subscription using Promise to avoid polling
-      await new Promise<void>((resolve, reject) => {
-        const timeout = setTimeout(() => {
-          reject(new Error('Timeout waiting for subscription'))
-        }, 10000)
-
-        supabaseWithCustomJwt.channel('test-channel').subscribe((status, err) => {
-          if (status === 'SUBSCRIBED') {
-            clearTimeout(timeout)
-            // Verify token was set
-            expect(supabaseWithCustomJwt.realtime.accessTokenValue).toBe(jwtToken)
-            resolve()
-          } else if (status === 'CHANNEL_ERROR' || status === 'TIMED_OUT') {
-            clearTimeout(timeout)
-            reject(err || new Error(`Subscription failed with status: ${status}`))
-          }
+      try {
+        // Wait for subscription using Promise to avoid polling
+        await new Promise<void>((resolve, reject) => {
+          const timeout = setTimeout(() => {
+            reject(new Error('Timeout waiting for subscription'))
+          }, 10000)
+
+          supabaseWithCustomJwt.channel('test-channel').subscribe((status, err) => {
+            if (status === 'SUBSCRIBED') {
+              clearTimeout(timeout)
+              // Verify token was set
+              expect(supabaseWithCustomJwt.realtime.accessTokenValue).toBe(jwtToken)
+              resolve()
+            } else if (status === 'CHANNEL_ERROR' || status === 'TIMED_OUT') {
+              clearTimeout(timeout)
+              reject(err || new Error(`Subscription failed with status: ${status}`))
+            }
+          })
         })
-      })
-
-      await supabaseWithCustomJwt.removeAllChannels()
+      } finally {
+        // Always cleanup channels and connection, even if test fails
+        await supabaseWithCustomJwt.removeAllChannels()
+      }
     }, 15000)
   })
 })