mattermost-mobile-upstream/app/managers/websocket_manager.ts

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.

import NetInfo, {NetInfoStateType, type NetInfoState, type NetInfoSubscription} from '@react-native-community/netinfo';
import {AppState, type AppStateStatus, type NativeEventSubscription} from 'react-native';
import {BackgroundTimer} from 'react-native-nitro-bg-timer-plus';
import {BehaviorSubject} from 'rxjs';
import {distinctUntilChanged} from 'rxjs/operators';

import {setCurrentUserStatus} from '@actions/local/user';
import {fetchStatusByIds} from '@actions/remote/user';
import {handleFirstConnect, handleReconnect} from '@actions/websocket';
import {handleWebSocketEvent} from '@actions/websocket/event';
import {hasActiveNativeCall} from '@calls/native_call_mappings';
import WebSocketClient from '@client/websocket';
import {General} from '@constants';
import DatabaseManager from '@database/manager';
import SessionAttributesManager from '@managers/session_attributes_manager';
import {getCurrentUserId} from '@queries/servers/system';
import {queryAllUsers} from '@queries/servers/user';
import RenderPermissionsStore from '@store/render_permissions_store';
import {toMilliseconds} from '@utils/datetime';
import {isMainActivity} from '@utils/helpers';
import {logDebug, logError} from '@utils/log';

const WAIT_TO_CLOSE = toMilliseconds({seconds: 15});
const WAIT_UNTIL_NEXT = toMilliseconds({seconds: 5});

class WebsocketManagerSingleton {
    private connectedSubjects: {[serverUrl: string]: BehaviorSubject<WebsocketConnectedState>} = {};

    private clients: Record<string, WebSocketClient> = {};
    private connectionTimerIDs: Record<string, NodeJS.Timeout> = {};
    private isBackgroundTimerRunning = false;
    private netConnected = false;
    private netType: NetInfoStateType = NetInfoStateType.none;
    private networkIdentity = '';
    private previousActiveState: boolean;
    private statusUpdatesIntervalIDs: Record<string, NodeJS.Timeout> = {};
    private backgroundTimerId: ReturnType<typeof BackgroundTimer.setTimeout> | undefined;
    private firstConnectionSynced: Record<string, boolean> = {};

    private appStateSubscription: NativeEventSubscription | undefined;
    private netStateSubscription: NetInfoSubscription | undefined;

    constructor() {
        this.previousActiveState = AppState.currentState === 'active';
    }

    public init = async (serverCredentials: ServerCredential[]) => {
        logDebug('WebSocketManager: Initializing');
        const netInfo = await NetInfo.fetch();
        this.netConnected = Boolean(netInfo.isConnected);
        this.netType = netInfo.type;
        await Promise.all(serverCredentials.map(
            async ({serverUrl, token, preauthSecret}) => {
                try {
                    DatabaseManager.getServerDatabaseAndOperator(serverUrl);
                    await this.createClient(serverUrl, token, preauthSecret);
                } catch (error) {
                    logError('WebsocketManager init error', error);
                }
            },
        ));

        this.appStateSubscription?.remove();
        this.netStateSubscription?.();

        this.appStateSubscription = AppState.addEventListener('change', this.onAppStateChange);
        this.netStateSubscription = NetInfo.addEventListener(this.onNetStateChange);
    };

    public invalidateClient = async (serverUrl: string) => {
        const client = this.clients[serverUrl];

        // Evict from registry and clear reconnect timer before awaiting teardown,
        // so no concurrent code path (e.g. openAll/initializeClient) can reach
        // this client or use stale manager state during the wait.
        clearTimeout(this.connectionTimerIDs[serverUrl]);
        delete this.connectionTimerIDs[serverUrl];
        delete this.clients[serverUrl];
        delete this.firstConnectionSynced[serverUrl];
        this.stopPeriodicStatusUpdates(serverUrl);

        if (client) {
            client.close(true);
            await client.waitForClose();
            client.invalidate();
        }

        // We don't remove the connected subject so any potential client invalidation
        // and subsequent creation of the client can still be observed by the component.
        // Being purist, this is a memory leak, since we never clean any server url,
        // but since this information lives in memory and we don't expect many servers
        // to be added and removed in one single session, this should be fine.
        this.getConnectedSubject(serverUrl).next('not_connected');
    };

    public createClient = async (serverUrl: string, bearerToken: string, preauthSecret?: string) => {
        if (this.clients[serverUrl]) {
            await this.invalidateClient(serverUrl);
        }

        const client = new WebSocketClient(serverUrl, bearerToken, preauthSecret);

        // Guard: callbacks capture the client instance so stale callbacks from
        // an old client that was replaced (but whose late onClose still fires)
        // can be detected and dropped by comparing against this.clients[serverUrl].
        const isCurrentClient = () => this.clients[serverUrl] === client;

        client.setFirstConnectCallback(() => isCurrentClient() && this.onFirstConnect(serverUrl));
        client.setEventCallback((evt: WebSocketMessage) => isCurrentClient() && handleWebSocketEvent(serverUrl, evt));

        //client.setMissedEventsCallback(() => {}) Nothing to do on missedEvents callback
        client.setReconnectCallback(() => isCurrentClient() && this.onReconnect(serverUrl));
        client.setReliableReconnectCallback(() => isCurrentClient() && this.onReliableReconnect(serverUrl));
        client.setCloseCallback((connectFailCount: number) => isCurrentClient() && this.onWebsocketClose(serverUrl, connectFailCount));

        this.clients[serverUrl] = client;

        return this.clients[serverUrl];
    };

    public closeAll = () => {
        for (const url of Object.keys(this.clients)) {
            const client = this.clients[url];
            client.close(true);
            this.getConnectedSubject(url).next('not_connected');
        }
    };

    public openAll = async (groupLabel?: BaseRequestGroupLabel) => {
        let queued = 0;
        for await (const clientUrl of Object.keys(this.clients)) {
            const activeServerUrl = await DatabaseManager.getActiveServerUrl();
            if (clientUrl === activeServerUrl) {
                this.initializeClient(clientUrl, groupLabel);
            } else {
                queued += 1;
                this.getConnectedSubject(clientUrl).next('connecting');
                this.connectionTimerIDs[clientUrl] = setTimeout(() => this.initializeClient(clientUrl, groupLabel), WAIT_UNTIL_NEXT * queued);
            }
        }
    };

    public isConnected = (serverUrl: string): boolean => {
        return this.clients[serverUrl]?.isConnected();
    };

    public observeWebsocketState = (serverUrl: string) => {
        return this.getConnectedSubject(serverUrl).asObservable().pipe(
            distinctUntilChanged(),
        );
    };

    private getConnectedSubject = (serverUrl: string) => {
        if (!this.connectedSubjects[serverUrl]) {
            this.connectedSubjects[serverUrl] = new BehaviorSubject(this.isConnected(serverUrl) ? 'connected' : 'not_connected');
        }

        return this.connectedSubjects[serverUrl];
    };

    private cancelConnectTimers = () => {
        for (const [url, timer] of Object.entries(this.connectionTimerIDs)) {
            clearTimeout(timer);
            delete this.connectionTimerIDs[url];
        }
    };

    public initializeClient = async (serverUrl: string, groupLabel: BaseRequestGroupLabel = 'WebSocket Reconnect') => {
        const client = this.clients[serverUrl];
        clearTimeout(this.connectionTimerIDs[serverUrl]);
        delete this.connectionTimerIDs[serverUrl];
        if (!client) {
            return;
        }
        if (!client.isConnected()) {
            const hasSynced = this.firstConnectionSynced[serverUrl];
            client.initialize({}, !hasSynced);
            if (!hasSynced) {
                const error = await handleFirstConnect(serverUrl, groupLabel);
                if (error) {
                    // This will try to reconnect and try to sync again
                    client.close(false);
                }

                // Makes sure a client still exist, and therefore we haven't been logged out
                if (this.clients[serverUrl]) {
                    this.firstConnectionSynced[serverUrl] = true;
                }
            }
        }
    };

    private onFirstConnect = (serverUrl: string) => {
        this.startPeriodicStatusUpdates(serverUrl);
        this.getConnectedSubject(serverUrl).next('connected');
    };

    private onReconnect = async (serverUrl: string) => {
        this.startPeriodicStatusUpdates(serverUrl);
        this.getConnectedSubject(serverUrl).next('connected');

        const error = await handleReconnect(serverUrl);
        if (error) {
            this.getClient(serverUrl)?.close(false);
        }
    };

    private onReliableReconnect = async (serverUrl: string) => {
        this.getConnectedSubject(serverUrl).next('connected');
    };

    private onWebsocketClose = async (serverUrl: string, connectFailCount: number) => {
        this.getConnectedSubject(serverUrl).next('not_connected');
        if (connectFailCount <= 1) { // First fail
            await setCurrentUserStatus(serverUrl, General.OFFLINE);
            this.stopPeriodicStatusUpdates(serverUrl);
        }
    };

    private startPeriodicStatusUpdates(serverUrl: string) {
        let currentId = this.statusUpdatesIntervalIDs[serverUrl];
        if (currentId != null) {
            clearInterval(currentId);
        }

        const getStatusForUsers = async () => {
            const database = DatabaseManager.serverDatabases[serverUrl];
            if (!database) {
                return;
            }

            const currentUserId = await getCurrentUserId(database.database);
            const userIds = (await queryAllUsers(database.database).fetchIds()).filter((id) => id !== currentUserId);

            fetchStatusByIds(serverUrl, userIds);
        };

        currentId = setInterval(getStatusForUsers, General.STATUS_INTERVAL);
        this.statusUpdatesIntervalIDs[serverUrl] = currentId;
        getStatusForUsers();
    }

    private stopPeriodicStatusUpdates(serverUrl: string) {
        clearInterval(this.statusUpdatesIntervalIDs[serverUrl]);
        delete this.statusUpdatesIntervalIDs[serverUrl];
    }

    private onAppStateChange = (appState: AppStateStatus) => {
        const isMain = isMainActivity();
        if (!isMain) {
            return;
        }

        const isActive = appState === 'active';
        this.handleStateChange(this.netConnected, this.netType, isActive);
    };

    private onNetStateChange = (netState: NetInfoState) => {
        const newState = Boolean(netState.isConnected);
        this.handleStateChange(newState, netState.type, this.previousActiveState);

        // After handleStateChange so a network type switch has already closed the sockets: the
        // revalidation then waits for the reconnect instead of going out mid-switch.
        this.expireNetworkBoundDecisions(netState);
    };

    /**
     * Network session attributes (SSID, interface type) feed ABAC decisions and change with no server
     * event, so render-time decisions taken on the previous network are revalidated. Checked here and
     * not in handleStateChange, which ignores a switch between two networks of the same type.
     * iOS reports no SSID (NetInfo's shouldFetchWiFiSSID is off), so a Wi-Fi to Wi-Fi switch there is
     * only seen through the disconnect between them: any reconnection counts as a new network.
     */
    private expireNetworkBoundDecisions = (netState: NetInfoState) => {
        if (!netState.isConnected) {
            this.networkIdentity = '';
            return;
        }
        const ssid = netState.type === NetInfoStateType.wifi ? netState.details?.ssid : undefined;
        const identity = `${netState.type}|${ssid ?? ''}`;
        if (identity !== this.networkIdentity) {
            this.networkIdentity = identity;

            // Only servers that collect session attributes evaluate the device's network.
            for (const serverUrl of SessionAttributesManager.getEnabledServers()) {
                RenderPermissionsStore.expireServer(serverUrl);
            }
        }
    };

    private handleStateChange = (currentIsConnected: boolean, currentNetType: NetInfoStateType, currentIsActive: boolean) => {
        if (currentIsActive === this.previousActiveState && currentIsConnected === this.netConnected && currentNetType === this.netType) {
            return;
        }

        this.cancelConnectTimers();

        const wentBackground = this.previousActiveState && !currentIsActive;
        const switchedNetworks = currentIsConnected && currentNetType !== this.netType && this.netType !== 'none';
        const previousNetType = this.netType;

        this.previousActiveState = currentIsActive;
        this.netConnected = currentIsConnected;
        this.netType = currentNetType;

        if (!currentIsConnected) {
            logDebug('WebSocketManager: closing all connections, network disconnected', 'netType', currentNetType);
            this.closeAll();
            return;
        }

        if (switchedNetworks) {
            // Close all connections when we switch from (for example) vpn to wifi
            // to ensure we are using the right network and doesn't get stuck on
            // retries.
            logDebug('WebSocketManager: closing all connections, network switched', 'from', previousNetType, 'to', currentNetType);
            this.closeAll();
        }

        if (currentIsActive) {
            if (this.isBackgroundTimerRunning) {
                BackgroundTimer.clearTimeout(this.backgroundTimerId!);
            }
            this.isBackgroundTimerRunning = false;
            if (this.netConnected) {
                this.openAll('WebSocket Reconnect');
            }

            return;
        }

        if (wentBackground) {
            this.startBackgroundCloseTimer();
        }
    };

    public scheduleBackgroundCloseIfNeeded = () => {
        if (this.previousActiveState) {
            return;
        }
        this.startBackgroundCloseTimer();
    };

    private startBackgroundCloseTimer = () => {
        if (this.isBackgroundTimerRunning) {
            return;
        }
        this.isBackgroundTimerRunning = true;
        this.backgroundTimerId = BackgroundTimer.setTimeout(() => {
            this.isBackgroundTimerRunning = false;

            // Skip closing while a native call is active; closeAll would drop
            // the WS that's carrying call lifecycle events.
            if (hasActiveNativeCall()) {
                return;
            }
            this.closeAll();
        }, WAIT_TO_CLOSE);
    };

    public getClient = (serverUrl: string): WebSocketClient | undefined => {
        return this.clients[serverUrl];
    };
}

const WebsocketManager = new WebsocketManagerSingleton();
export default WebsocketManager;