// 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-background-timer'; 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 WebSocketClient from '@client/websocket'; import {General} from '@constants'; import DatabaseManager from '@database/manager'; import {getCurrentUserId} from '@queries/servers/system'; import {queryAllUsers} from '@queries/servers/user'; import {toMilliseconds} from '@utils/datetime'; import {isMainActivity} from '@utils/helpers'; import {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} = {}; private clients: Record = {}; private connectionTimerIDs: Record = {}; private isBackgroundTimerRunning = false; private netConnected = false; private netType: NetInfoStateType = NetInfoStateType.none; private previousActiveState: boolean; private statusUpdatesIntervalIDs: Record = {}; private backgroundIntervalId: number | undefined; private firstConnectionSynced: Record = {}; private appStateSubscription: NativeEventSubscription | undefined; private netStateSubscription: NetInfoSubscription | undefined; constructor() { this.previousActiveState = AppState.currentState === 'active'; } public init = async (serverCredentials: ServerCredential[]) => { const netInfo = await NetInfo.fetch(); this.netConnected = Boolean(netInfo.isConnected); this.netType = netInfo.type; serverCredentials.forEach( ({serverUrl, token, preauthSecret}) => { try { DatabaseManager.getServerDatabaseAndOperator(serverUrl); 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 = (serverUrl: string) => { this.clients[serverUrl]?.close(true); this.clients[serverUrl]?.invalidate(); clearTimeout(this.connectionTimerIDs[serverUrl]); delete this.clients[serverUrl]; delete this.firstConnectionSynced[serverUrl]; // 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 = (serverUrl: string, bearerToken: string, preauthSecret?: string) => { if (this.clients[serverUrl]) { this.invalidateClient(serverUrl); } const client = new WebSocketClient(serverUrl, bearerToken, preauthSecret); client.setFirstConnectCallback(() => this.onFirstConnect(serverUrl)); client.setEventCallback((evt: WebSocketMessage) => handleWebSocketEvent(serverUrl, evt)); //client.setMissedEventsCallback(() => {}) Nothing to do on missedEvents callback client.setReconnectCallback(() => this.onReconnect(serverUrl)); client.setReliableReconnectCallback(() => this.onReliableReconnect(serverUrl)); client.setCloseCallback((connectFailCount: number) => 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); client.invalidate(); 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: WebSocketClient = this.clients[serverUrl]; clearTimeout(this.connectionTimerIDs[serverUrl]); delete this.connectionTimerIDs[serverUrl]; 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); }; 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'; this.previousActiveState = currentIsActive; this.netConnected = currentIsConnected; this.netType = currentNetType; if (!currentIsConnected) { 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. this.closeAll(); } if (currentIsActive) { if (this.isBackgroundTimerRunning) { BackgroundTimer.clearInterval(this.backgroundIntervalId!); } this.isBackgroundTimerRunning = false; if (this.netConnected) { this.openAll('WebSocket Reconnect'); } return; } if (wentBackground && !this.isBackgroundTimerRunning) { this.isBackgroundTimerRunning = true; this.backgroundIntervalId = BackgroundTimer.setInterval(() => { this.closeAll(); BackgroundTimer.clearInterval(this.backgroundIntervalId!); this.isBackgroundTimerRunning = false; }, WAIT_TO_CLOSE); } }; public getClient = (serverUrl: string): WebSocketClient | undefined => { return this.clients[serverUrl]; }; } const WebsocketManager = new WebsocketManagerSingleton(); export default WebsocketManager;