Fix clientId mismatch on login

.github/workflows/tests.yml

@@ -6,8 +6,6 @@ on:
     branches:
       - master
       - '*.x'
-  schedule:
-    - cron: '0 0 * * 0' # Once in a week run tests to check assertions against updated(if any) ably-js package.
 
 jobs:
   tests:

src/channel/ably/attach.ts

@@ -7,6 +7,7 @@ let channelAttachAuthorized = false;
  */
 export const beforeChannelAttach = (ablyClient, authorize: Function) => {
     const dummyRealtimeChannel = ablyClient.channels.get('dummy');
+    dummyRealtimeChannel.__proto__.authorizeChannel = authorize;
     if (channelAttachAuthorized) {
         return;
     }
@@ -22,7 +23,7 @@ export const beforeChannelAttach = (ablyClient, authorize: Function) => {
         this.authorizing = true;
         const bindedInternalAttach = internalAttach.bind(this);
 
-        authorize(this, (error) => {
+        this.authorizeChannel(this, (error) => {
             this.authorizing = false;
             if (error) {
                 if (errCallback) {

src/channel/ably/auth.ts

@@ -4,12 +4,13 @@ import { SequentialAuthTokenRequestExecuter } from './token-request';
 import { AblyChannel } from '../ably-channel';
 import { AblyConnector } from '../../connector/ably-connector';
 import { AblyPresenceChannel } from '../ably-presence-channel';
-import { AuthOptions, ChannelStateChange, ClientOptions } from '../../../typings/ably';
+import { AblyRealtime, AuthOptions, ChannelStateChange, ClientOptions, TokenDetails } from '../../../typings/ably';
 
 export class AblyAuth {
     authEndpoint: string;
     authHeaders: any;
     authRequestExecuter: SequentialAuthTokenRequestExecuter;
+    ablyConnector: AblyConnector;
 
     expiredAuthChannels = new Set<string>();
     setExpired = (channelName: string) => this.expiredAuthChannels.add(channelName);
@@ -47,7 +48,8 @@ export class AblyAuth {
         return await httpRequestAsync(postOptions);
     };
 
-    constructor(options) {
+    constructor(ablyConnector: AblyConnector, options) {
+        this.ablyConnector = ablyConnector;
         const {
             authEndpoint,
             auth: { headers },
@@ -59,8 +61,14 @@ export class AblyAuth {
         this.authRequestExecuter = new SequentialAuthTokenRequestExecuter(token, requestTokenFn ?? this.requestToken);
     }
 
-    enableAuthorizeBeforeChannelAttach = (ablyConnector: AblyConnector) => {
-        const ablyClient: any = ablyConnector.ably;
+    ablyClient = () => this.ablyConnector.ably as AblyRealtime | any;
+
+    existingToken = () => this.ablyClient().auth.tokenDetails as TokenDetails;
+
+    getChannel = name => this.ablyConnector.channels[name];
+
+    enableAuthorizeBeforeChannelAttach = () => {
+        const ablyClient = this.ablyClient()
         ablyClient.auth.getTimestamp(this.options.queryTime, () => void 0); // generates serverTimeOffset in the background
 
         beforeChannelAttach(ablyClient, (realtimeChannel, errorCallback) => {
@@ -71,7 +79,7 @@ export class AblyAuth {
             }
 
             // Use cached token if has channel capability and is not expired
-            const tokenDetails = ablyClient.auth.tokenDetails;
+            const tokenDetails = this.existingToken();
             if (tokenDetails && !this.isExpired(channelName)) {
                 const capability = parseJwt(tokenDetails.token).payload['x-ably-capability'];
                 const tokenHasChannelCapability = capability.includes(`${channelName}"`);
@@ -87,10 +95,9 @@ export class AblyAuth {
                 .request(channelName)
                 .then(({ token: jwtToken, info }) => {
                     // get upgraded token with channel access
-                    const echoChannel = ablyConnector.channels[channelName];
+                    const echoChannel = this.getChannel(channelName);
                     this.setPresenceInfo(echoChannel, info);
-                    ablyClient.auth.authorize(
-                        null,
+                    this.tryAuthorizeOnSameConnection(
                         { ...this.options, token: toTokenDetails(jwtToken) },
                         (err, _tokenDetails) => {
                             if (err) {
@@ -106,6 +113,37 @@ export class AblyAuth {
         });
     };
 
+    allowReconnectOnUserLogin = () => {
+        const ablyConnection = this.ablyClient().connection
+
+        const connectionFailedCallback = stateChange => {
+            if (stateChange.reason.code == 40102) { // 40102 denotes mismatched clientId
+                ablyConnection.off(connectionFailedCallback);
+                console.warn("User login detected, re-connecting again!")
+                this.onClientIdChanged();
+            }
+        }
+        ablyConnection.on('failed', connectionFailedCallback);
+    }
+
+    /**
+     * This will be called when (guest)user logs in and new clientId is returned in the jwt token.
+     * If client tries to authenticate with new clientId on same connection, ably server returns
+     * error and connection goes into failed state.
+     * See https://github.com/ably/laravel-broadcaster/issues/45 for more details.
+     * There's a separate test case added for user login flow => ably-user-login.test.ts.
+     */
+    onClientIdChanged = () => {
+        this.ablyClient().connect();
+        for (const ablyChannel of Object.values(this.ablyConnector.channels)) {
+            ablyChannel.channel.attach(ablyChannel._alertErrorListeners);
+        }
+    }
+
+    tryAuthorizeOnSameConnection = (authOptions?: AuthOptions, callback?: (error, TokenDetails) => void) => {
+        this.ablyClient().auth.authorize(null, authOptions, callback)
+    }
+
     onChannelFailed = (echoAblyChannel: AblyChannel) => (stateChange: ChannelStateChange) => {
         // channel capability rejected https://help.ably.io/error/40160
         if (stateChange.reason?.code == 40160) {
@@ -123,8 +161,7 @@ export class AblyAuth {
             .request(channelName)
             .then(({ token: jwtToken, info }) => {
                 this.setPresenceInfo(echoAblyChannel, info);
-                echoAblyChannel.ably.auth.authorize(
-                    null,
+                this.tryAuthorizeOnSameConnection(
                     { ...this.options, token: toTokenDetails(jwtToken) as any },
                     (err, _tokenDetails) => {
                         if (err) {

src/connector/ably-connector.ts

@@ -1,7 +1,7 @@
 import { Connector } from './connector';
 
 import { AblyChannel, AblyPrivateChannel, AblyPresenceChannel, AblyAuth } from './../channel';
-import { AblyRealtime } from '../../typings/ably';
+import { AblyRealtime, TokenDetails } from '../../typings/ably';
 
 /**
  * This class creates a connector to Ably.
@@ -35,13 +35,14 @@ export class AblyConnector extends Connector {
         if (typeof this.options.client !== 'undefined') {
             this.ably = this.options.client;
         } else {
-            this.ablyAuth = new AblyAuth(this.options);
+            this.ablyAuth = new AblyAuth(this, this.options);
             if (!this.options.agents) {
                 this.options.agents = {};
             }
             this.options.agents['laravel-echo'] = AblyConnector.LIB_VERSION;
             this.ably = new Ably.Realtime({ ...this.ablyAuth.options, ...this.options });
-            this.ablyAuth.enableAuthorizeBeforeChannelAttach(this);
+            this.ablyAuth.enableAuthorizeBeforeChannelAttach();
+            this.ablyAuth.allowReconnectOnUserLogin()
         }
     }
 

tests/ably/ably-user-login.test.ts

@@ -0,0 +1,184 @@
+import { setup, tearDown } from './setup/sandbox';
+import Echo from '../../src/echo';
+import { MockAuthServer } from './setup/mock-auth-server';
+import { AblyChannel, AblyPrivateChannel } from '../../src/channel';
+import * as Ably from 'ably';
+import waitForExpect from 'wait-for-expect';
+
+jest.setTimeout(20000);
+describe('AblyUserLogin', () => {
+    let testApp: any;
+    let mockAuthServer: MockAuthServer;
+    let echo: Echo;
+
+    beforeAll(async () => {
+        global.Ably = Ably;
+        testApp = await setup();
+        mockAuthServer = new MockAuthServer(testApp.keys[0].keyStr);
+    });
+
+    afterAll(async () => {
+        return await tearDown(testApp);
+    });
+
+    beforeEach(() => {
+        mockAuthServer.clientId = null;
+        echo = new Echo({
+            broadcaster: 'ably',
+            useTls: true,
+            environment: 'sandbox',
+            requestTokenFn: mockAuthServer.getSignedToken,
+            echoMessages: true, // https://docs.ably.io/client-lib-development-guide/features/#TO3h
+        });
+    });
+
+    afterEach((done) => {
+        echo.disconnect();
+        echo.connector.ably.connection.once('closed', () => {
+            done();
+        });
+    });
+
+    test('user logs in without previous (guest) channels', async () => {
+        let connectionStates : Array<any>= []
+        // Initial clientId is null
+        expect(mockAuthServer.clientId).toBeNull();
+        await waitForExpect(() => {
+            expect(echo.connector.ably.connection.state).toBe('connected')
+        });
+
+        // Track all connection state changes
+        echo.connector.ably.connection.on(stateChange => {
+            connectionStates.push(stateChange.current)
+        });
+
+        // Making sure client is still anonymous
+        expect(mockAuthServer.clientId).toBeNull();
+        expect(echo.connector.ablyAuth.existingToken().clientId).toBeNull()
+
+        // Set server clientId to [email protected], user logs in for next request
+        mockAuthServer.clientId = '[email protected]'
+        const privateChannel = echo.private('test') as AblyPrivateChannel; // Requests new token        
+        await new Promise((resolve) => privateChannel.subscribed(resolve));
+
+        // Connection goes into failed state and then reconnects again
+        await waitForExpect(() => {
+            expect(connectionStates).toStrictEqual(['failed', 'connecting', 'connected'])
+        });
+        expect(privateChannel.channel.state).toBe('attached');
+
+        expect(echo.connector.ablyAuth.existingToken().clientId).toBe('[email protected]');
+    });
+
+    test('user logs in with previous (guest) channels', async () => {
+        let connectionStates : Array<any>= []
+        let publicChannelStates : Array<any>= []
+
+        // Initial clientId is null
+        expect(mockAuthServer.clientId).toBeNull();
+        await waitForExpect(() => {
+            expect(echo.connector.ably.connection.state).toBe('connected')
+        });
+
+        // Track all connection state changes
+        echo.connector.ably.connection.on(stateChange => {
+            connectionStates.push(stateChange.current)
+        })
+
+        // Subscribe to a public channel as a guest user
+        const publicChannel = echo.channel('test1') as AblyChannel;
+        await new Promise((resolve) => publicChannel.subscribed(resolve));
+        publicChannel.channel.on(stateChange => {
+            publicChannelStates.push(stateChange.current)
+        })
+
+        // Making sure client is still anonymous
+        expect(mockAuthServer.clientId).toBeNull();
+        expect(echo.connector.ablyAuth.existingToken().clientId).toBeNull();
+
+        // Set server clientId to [email protected], user logs in for next request
+        mockAuthServer.clientId = '[email protected]';
+        const privateChannel = echo.private('test') as AblyPrivateChannel; // requests new token for channel `test`
+
+        // Since new clientId is returned in the new token, ably returns mismatched error for given auth request
+        const privateChannelErr : Error = await new Promise(resolve => privateChannel.error(resolve));
+        expect(privateChannelErr.message).toContain('Mismatched clientId for existing connection');
+
+        // Reconnects again and starts explicit attach for all channels
+        await waitForExpect(() => {
+            expect(echo.connector.ably.connection.state).toBe('connected')
+        });
+
+        // Attaches both public and private channel
+        await Promise.all([ 
+            new Promise(resolve => publicChannel.subscribed(resolve)),
+            new Promise(resolve => privateChannel.subscribed(resolve))
+        ]);
+
+        await waitForExpect(() => {
+            expect(connectionStates).toStrictEqual(['failed', 'connecting', 'connected']);
+        });
+        await waitForExpect(() => {
+            expect(publicChannelStates).toStrictEqual(['failed', 'attaching', 'attached']);
+        });
+        expect(privateChannel.channel.state).toBe('attached');
+
+        expect(echo.connector.ablyAuth.existingToken().clientId).toBe('[email protected]');
+    });
+
+    test('user logs in and then logs out', async() => {
+        let connectionStates : Array<any>= []
+        let privateChannelStates : Array<any>= []
+
+        // Initial clientId is null
+        expect(mockAuthServer.clientId).toBeNull();
+        await waitForExpect(() => {
+            expect(echo.connector.ably.connection.state).toBe('connected');
+        });
+
+        // Track all connection state changes
+        echo.connector.ably.connection.on(stateChange => {
+            connectionStates.push(stateChange.current)
+        });
+
+        // Making sure client is still anonymous
+        expect(mockAuthServer.clientId).toBeNull();
+        expect(echo.connector.ablyAuth.existingToken().clientId).toBeNull()
+
+        // Set server clientId to [email protected], so user logs in for next request
+        mockAuthServer.clientId = '[email protected]'
+        const privateChannel = echo.private('test') as AblyPrivateChannel; // Requests new token     
+        privateChannel.channel.on(statechange => {
+            privateChannelStates.push(statechange.current)
+        })
+        const privateChannel1ErrPromise = new Promise((resolve) => privateChannel.error(resolve))
+
+        await new Promise((resolve) => privateChannel.subscribed(resolve)); // successful attach 
+        await waitForExpect(() => {
+            expect(connectionStates).toStrictEqual(['failed', 'connecting', 'connected'])
+        });
+
+        // Logout user by setting clientId to null
+        mockAuthServer.clientId = null
+        const privateChannel2 = echo.private('test1') as AblyPrivateChannel; // Requests new token with null clientId
+
+        // Receives error on both channels
+        const privateChannel1Err = await privateChannel1ErrPromise as any;
+        const privateChannel2Err = await new Promise((resolve) => privateChannel2.error(resolve)) as any;
+
+        const errMsg = 'Mismatched clientId for existing connection'
+        expect(privateChannel1Err.message).toContain(errMsg);
+        expect(privateChannel2Err.message).toContain(errMsg);
+
+        await waitForExpect(() => {
+            expect(privateChannelStates).toStrictEqual(['attaching', 'attached', 'failed']);
+        });
+
+        // Connection transitions to failed state
+        await waitForExpect(() => {
+            expect(connectionStates).toStrictEqual(['failed', 'connecting', 'connected', 'failed'])
+        });
+
+        expect(echo.connector.ablyAuth.existingToken().clientId).toBeNull();
+    });
+});

tests/ably/setup/mock-auth-server.ts

@@ -8,7 +8,7 @@ export class MockAuthServer {
     keyName: string;
     keySecret: string;
     ablyClient: Ably.Rest;
-    clientId = '[email protected]';
+    clientId: string | null = '[email protected]';
     userInfo = { id: '[email protected]', name: 'sacOO7' };
 
     shortLived: channels;