diff --git a/keyserver/src/socket/socket.js b/keyserver/src/socket/socket.js index 5dc0eb7f1..93479ea25 100644 --- a/keyserver/src/socket/socket.js +++ b/keyserver/src/socket/socket.js @@ -1,844 +1,834 @@ // @flow import type { $Request } from 'express'; import invariant from 'invariant'; import _debounce from 'lodash/debounce.js'; import t from 'tcomb'; import type { TUnion } from 'tcomb'; import WebSocket from 'ws'; import { baseLegalPolicies } from 'lib/facts/policies.js'; import { mostRecentMessageTimestamp } from 'lib/shared/message-utils.js'; import { serverRequestSocketTimeout, serverResponseTimeout, } from 'lib/shared/timeouts.js'; import { mostRecentUpdateTimestamp } from 'lib/shared/update-utils.js'; import type { Shape } from 'lib/types/core.js'; import { endpointIsSocketSafe } from 'lib/types/endpoints.js'; import { defaultNumberPerThread } from 'lib/types/message-types.js'; import { redisMessageTypes, type RedisMessage } from 'lib/types/redis-types.js'; import { serverRequestTypes } from 'lib/types/request-types.js'; import { cookieSources, sessionCheckFrequency, stateCheckInactivityActivationInterval, } from 'lib/types/session-types.js'; import { type ClientSocketMessage, type InitialClientSocketMessage, type ResponsesClientSocketMessage, type ServerStateSyncFullSocketPayload, type ServerServerSocketMessage, type ErrorServerSocketMessage, type AuthErrorServerSocketMessage, type PingClientSocketMessage, type AckUpdatesClientSocketMessage, type APIRequestClientSocketMessage, clientSocketMessageTypes, stateSyncPayloadTypes, serverSocketMessageTypes, serverServerSocketMessageValidator, } from 'lib/types/socket-types.js'; import { ServerError } from 'lib/utils/errors.js'; import { values } from 'lib/utils/objects.js'; import { promiseAll } from 'lib/utils/promises.js'; import SequentialPromiseResolver from 'lib/utils/sequential-promise-resolver.js'; import sleep from 'lib/utils/sleep.js'; import { tShape, tCookie } from 'lib/utils/validation-utils.js'; import { RedisSubscriber } from './redis.js'; import { clientResponseInputValidator, processClientResponses, initializeSession, checkState, } from './session-utils.js'; import { fetchUpdateInfosWithRawUpdateInfos } from '../creators/update-creator.js'; import { deleteActivityForViewerSession } from '../deleters/activity-deleters.js'; import { deleteCookie } from '../deleters/cookie-deleters.js'; import { deleteUpdatesBeforeTimeTargetingSession } from '../deleters/update-deleters.js'; import { jsonEndpoints } from '../endpoints.js'; import { fetchEntryInfos } from '../fetchers/entry-fetchers.js'; import { fetchMessageInfosSince, getMessageFetchResultFromRedisMessages, } from '../fetchers/message-fetchers.js'; import { fetchThreadInfos } from '../fetchers/thread-fetchers.js'; import { fetchUpdateInfos } from '../fetchers/update-fetchers.js'; import { fetchCurrentUserInfo, fetchKnownUserInfos, } from '../fetchers/user-fetchers.js'; import { newEntryQueryInputValidator, verifyCalendarQueryThreadIDs, } from '../responders/entry-responders.js'; import { handleAsyncPromise } from '../responders/handlers.js'; import { fetchViewerForSocket, extendCookieLifespan, createNewAnonymousCookie, isCookieMissingSignedIdentityKeysBlob, } from '../session/cookies.js'; import { Viewer } from '../session/viewer.js'; import { commitSessionUpdate } from '../updaters/session-updaters.js'; import { assertSecureRequest } from '../utils/security-utils.js'; import { checkInputValidator, checkClientSupported, policiesValidator, validateOutput, } from '../utils/validation-utils.js'; const clientSocketMessageInputValidator: TUnion = t.union([ tShape({ type: t.irreducible( 'clientSocketMessageTypes.INITIAL', x => x === clientSocketMessageTypes.INITIAL, ), id: t.Number, payload: tShape({ sessionIdentification: tShape({ cookie: t.maybe(tCookie), sessionID: t.maybe(t.String), }), sessionState: tShape({ calendarQuery: newEntryQueryInputValidator, messagesCurrentAsOf: t.Number, updatesCurrentAsOf: t.Number, watchedIDs: t.list(t.String), }), clientResponses: t.list(clientResponseInputValidator), }), }), tShape({ type: t.irreducible( 'clientSocketMessageTypes.RESPONSES', x => x === clientSocketMessageTypes.RESPONSES, ), id: t.Number, payload: tShape({ clientResponses: t.list(clientResponseInputValidator), }), }), tShape({ type: t.irreducible( 'clientSocketMessageTypes.PING', x => x === clientSocketMessageTypes.PING, ), id: t.Number, }), tShape({ type: t.irreducible( 'clientSocketMessageTypes.ACK_UPDATES', x => x === clientSocketMessageTypes.ACK_UPDATES, ), id: t.Number, payload: tShape({ currentAsOf: t.Number, }), }), tShape({ type: t.irreducible( 'clientSocketMessageTypes.API_REQUEST', x => x === clientSocketMessageTypes.API_REQUEST, ), id: t.Number, payload: tShape({ endpoint: t.String, input: t.maybe(t.Object), }), }), ]); function onConnection(ws: WebSocket, req: $Request) { assertSecureRequest(req); new Socket(ws, req); } type StateCheckConditions = { activityRecentlyOccurred: boolean, stateCheckOngoing: boolean, }; class Socket { ws: WebSocket; httpRequest: $Request; viewer: ?Viewer; redis: ?RedisSubscriber; redisPromiseResolver: SequentialPromiseResolver; stateCheckConditions: StateCheckConditions = { activityRecentlyOccurred: true, stateCheckOngoing: false, }; stateCheckTimeoutID: ?TimeoutID; constructor(ws: WebSocket, httpRequest: $Request) { this.ws = ws; this.httpRequest = httpRequest; ws.on('message', this.onMessage); ws.on('close', this.onClose); this.resetTimeout(); this.redisPromiseResolver = new SequentialPromiseResolver(this.sendMessage); } onMessage = async ( messageString: string | Buffer | ArrayBuffer | Array, ) => { invariant(typeof messageString === 'string', 'message should be string'); let clientSocketMessage: ?ClientSocketMessage; try { this.resetTimeout(); const messageObject = JSON.parse(messageString); clientSocketMessage = checkInputValidator( clientSocketMessageInputValidator, messageObject, ); if (clientSocketMessage.type === clientSocketMessageTypes.INITIAL) { if (this.viewer) { // This indicates that the user sent multiple INITIAL messages. throw new ServerError('socket_already_initialized'); } this.viewer = await fetchViewerForSocket( this.httpRequest, clientSocketMessage, ); if (!this.viewer) { // This indicates that the cookie was invalid, but the client is using // cookieSources.HEADER and thus can't accept a new cookie over // WebSockets. See comment under catch block for socket_deauthorized. throw new ServerError('socket_deauthorized'); } } const { viewer } = this; if (!viewer) { // This indicates a non-INITIAL message was sent by the client before // the INITIAL message. throw new ServerError('socket_uninitialized'); } if (viewer.sessionChanged) { // This indicates that the cookie was invalid, and we've assigned a new // anonymous one. throw new ServerError('socket_deauthorized'); } if (!viewer.loggedIn) { // This indicates that the specified cookie was an anonymous one. throw new ServerError('not_logged_in'); } await checkClientSupported( viewer, clientSocketMessageInputValidator, clientSocketMessage, ); await policiesValidator(viewer, baseLegalPolicies); const serverResponses = await this.handleClientSocketMessage( clientSocketMessage, ); if (!this.redis) { this.redis = new RedisSubscriber( { userID: viewer.userID, sessionID: viewer.session }, this.onRedisMessage, ); } if (viewer.sessionChanged) { // This indicates that something has caused the session to change, which // shouldn't happen from inside a WebSocket since we can't handle cookie // invalidation. throw new ServerError('session_mutated_from_socket'); } if (clientSocketMessage.type !== clientSocketMessageTypes.PING) { handleAsyncPromise(extendCookieLifespan(viewer.cookieID)); } for (const response of serverResponses) { this.sendMessage(response); } if (clientSocketMessage.type === clientSocketMessageTypes.INITIAL) { this.onSuccessfulConnection(); } } catch (error) { console.warn(error); if (!(error instanceof ServerError)) { const errorMessage: ErrorServerSocketMessage = { type: serverSocketMessageTypes.ERROR, message: error.message, }; const responseTo = clientSocketMessage ? clientSocketMessage.id : null; if (responseTo !== null) { errorMessage.responseTo = responseTo; } this.markActivityOccurred(); this.sendMessage(errorMessage); return; } invariant(clientSocketMessage, 'should be set'); const responseTo = clientSocketMessage.id; if (error.message === 'socket_deauthorized') { const authErrorMessage: AuthErrorServerSocketMessage = { type: serverSocketMessageTypes.AUTH_ERROR, responseTo, message: error.message, }; if (this.viewer) { // viewer should only be falsey for cookieSources.HEADER (web) // clients. Usually if the cookie is invalid we construct a new // anonymous Viewer with a new cookie, and then pass the cookie down // in the error. But we can't pass HTTP cookies in WebSocket messages. authErrorMessage.sessionChange = { cookie: this.viewer.cookiePairString, currentUserInfo: { id: this.viewer.cookieID, anonymous: true, }, }; } this.sendMessage(authErrorMessage); this.ws.close(4100, error.message); return; } else if (error.message === 'client_version_unsupported') { const { viewer } = this; invariant(viewer, 'should be set'); const promises = {}; promises.deleteCookie = deleteCookie(viewer.cookieID); if (viewer.cookieSource !== cookieSources.BODY) { promises.anonymousViewerData = createNewAnonymousCookie({ platformDetails: error.platformDetails, deviceToken: viewer.deviceToken, }); } const { anonymousViewerData } = await promiseAll(promises); const authErrorMessage: AuthErrorServerSocketMessage = { type: serverSocketMessageTypes.AUTH_ERROR, responseTo, message: error.message, }; if (anonymousViewerData) { // It is normally not safe to pass the result of // createNewAnonymousCookie to the Viewer constructor. That is because // createNewAnonymousCookie leaves several fields of // AnonymousViewerData unset, and consequently Viewer will throw when // access is attempted. It is only safe here because we can guarantee // that only cookiePairString and cookieID are accessed on anonViewer // below. const anonViewer = new Viewer(anonymousViewerData); authErrorMessage.sessionChange = { cookie: anonViewer.cookiePairString, currentUserInfo: { id: anonViewer.cookieID, anonymous: true, }, }; } this.sendMessage(authErrorMessage); this.ws.close(4101, error.message); return; } if (error.payload) { this.sendMessage({ type: serverSocketMessageTypes.ERROR, responseTo, message: error.message, payload: error.payload, }); } else { this.sendMessage({ type: serverSocketMessageTypes.ERROR, responseTo, message: error.message, }); } if (error.message === 'not_logged_in') { this.ws.close(4102, error.message); } else if (error.message === 'session_mutated_from_socket') { this.ws.close(4103, error.message); } else { this.markActivityOccurred(); } } }; onClose = async () => { this.clearStateCheckTimeout(); this.resetTimeout.cancel(); this.debouncedAfterActivity.cancel(); if (this.viewer && this.viewer.hasSessionInfo) { await deleteActivityForViewerSession(this.viewer); } if (this.redis) { this.redis.quit(); this.redis = null; } }; sendMessage = (message: ServerServerSocketMessage) => { invariant( this.ws.readyState > 0, "shouldn't send message until connection established", ); if (this.ws.readyState === 1) { const { viewer } = this; const validatedMessage = validateOutput( viewer?.platformDetails, serverServerSocketMessageValidator, message, ); this.ws.send(JSON.stringify(validatedMessage)); } }; async handleClientSocketMessage( message: ClientSocketMessage, ): Promise { const resultPromise = (async () => { if (message.type === clientSocketMessageTypes.INITIAL) { this.markActivityOccurred(); return await this.handleInitialClientSocketMessage(message); } else if (message.type === clientSocketMessageTypes.RESPONSES) { this.markActivityOccurred(); return await this.handleResponsesClientSocketMessage(message); } else if (message.type === clientSocketMessageTypes.PING) { return this.handlePingClientSocketMessage(message); } else if (message.type === clientSocketMessageTypes.ACK_UPDATES) { this.markActivityOccurred(); return await this.handleAckUpdatesClientSocketMessage(message); } else if (message.type === clientSocketMessageTypes.API_REQUEST) { this.markActivityOccurred(); return await this.handleAPIRequestClientSocketMessage(message); } return []; })(); const timeoutPromise = (async () => { await sleep(serverResponseTimeout); throw new ServerError('socket_response_timeout'); })(); return await Promise.race([resultPromise, timeoutPromise]); } async handleInitialClientSocketMessage( message: InitialClientSocketMessage, ): Promise { const { viewer } = this; invariant(viewer, 'should be set'); const responses = []; const { sessionState, clientResponses } = message.payload; const { calendarQuery, updatesCurrentAsOf: oldUpdatesCurrentAsOf, messagesCurrentAsOf: oldMessagesCurrentAsOf, watchedIDs, } = sessionState; await verifyCalendarQueryThreadIDs(calendarQuery); const sessionInitializationResult = await initializeSession( viewer, calendarQuery, oldUpdatesCurrentAsOf, ); const threadCursors = {}; for (const watchedThreadID of watchedIDs) { threadCursors[watchedThreadID] = null; } const messageSelectionCriteria = { threadCursors, joinedThreads: true, newerThan: oldMessagesCurrentAsOf, }; const [fetchMessagesResult, { serverRequests, activityUpdateResult }] = await Promise.all([ fetchMessageInfosSince( viewer, messageSelectionCriteria, defaultNumberPerThread, ), processClientResponses(viewer, clientResponses), ]); const messagesResult = { rawMessageInfos: fetchMessagesResult.rawMessageInfos, truncationStatuses: fetchMessagesResult.truncationStatuses, currentAsOf: mostRecentMessageTimestamp( fetchMessagesResult.rawMessageInfos, oldMessagesCurrentAsOf, ), }; - if ( - viewer.userAgent?.includes('Electron') && - viewer.platform === 'web' && - !serverRequests.find( - request => request.type === serverRequestTypes.PLATFORM_DETAILS, - ) - ) { - serverRequests.push({ type: serverRequestTypes.PLATFORM_DETAILS }); - } - const isCookieMissingSignedIdentityKeysBlobPromise = isCookieMissingSignedIdentityKeysBlob(viewer.cookieID); if (!sessionInitializationResult.sessionContinued) { const [threadsResult, entriesResult, currentUserInfo, knownUserInfos] = await Promise.all([ fetchThreadInfos(viewer), fetchEntryInfos(viewer, [calendarQuery]), fetchCurrentUserInfo(viewer), fetchKnownUserInfos(viewer), ]); const payload: ServerStateSyncFullSocketPayload = { type: stateSyncPayloadTypes.FULL, messagesResult, threadInfos: threadsResult.threadInfos, currentUserInfo, rawEntryInfos: entriesResult.rawEntryInfos, userInfos: values(knownUserInfos), updatesCurrentAsOf: oldUpdatesCurrentAsOf, }; if (viewer.sessionChanged) { // If initializeSession encounters, // sessionIdentifierTypes.BODY_SESSION_ID but the session // is unspecified or expired, // it will set a new sessionID and specify viewer.sessionChanged const { sessionID } = viewer; invariant( sessionID !== null && sessionID !== undefined, 'should be set', ); payload.sessionID = sessionID; viewer.sessionChanged = false; } responses.push({ type: serverSocketMessageTypes.STATE_SYNC, responseTo: message.id, payload, }); } else { const { sessionUpdate, deltaEntryInfoResult } = sessionInitializationResult; const promises = {}; promises.deleteExpiredUpdates = deleteUpdatesBeforeTimeTargetingSession( viewer, oldUpdatesCurrentAsOf, ); promises.fetchUpdateResult = fetchUpdateInfos( viewer, oldUpdatesCurrentAsOf, calendarQuery, ); promises.sessionUpdate = commitSessionUpdate(viewer, sessionUpdate); const { fetchUpdateResult } = await promiseAll(promises); const { updateInfos, userInfos } = fetchUpdateResult; const newUpdatesCurrentAsOf = mostRecentUpdateTimestamp( [...updateInfos], oldUpdatesCurrentAsOf, ); const updatesResult = { newUpdates: updateInfos, currentAsOf: newUpdatesCurrentAsOf, }; responses.push({ type: serverSocketMessageTypes.STATE_SYNC, responseTo: message.id, payload: { type: stateSyncPayloadTypes.INCREMENTAL, messagesResult, updatesResult, deltaEntryInfos: deltaEntryInfoResult.rawEntryInfos, deletedEntryIDs: deltaEntryInfoResult.deletedEntryIDs, userInfos: values(userInfos), }, }); } const signedIdentityKeysBlobMissing = await isCookieMissingSignedIdentityKeysBlobPromise; if (signedIdentityKeysBlobMissing) { serverRequests.push({ type: serverRequestTypes.SIGNED_IDENTITY_KEYS_BLOB, }); } if (serverRequests.length > 0 || clientResponses.length > 0) { // We send this message first since the STATE_SYNC triggers the client's // connection status to shift to "connected", and we want to make sure the // client responses are cleared from Redux before that happens responses.unshift({ type: serverSocketMessageTypes.REQUESTS, responseTo: message.id, payload: { serverRequests }, }); } if (activityUpdateResult) { // Same reason for unshifting as above responses.unshift({ type: serverSocketMessageTypes.ACTIVITY_UPDATE_RESPONSE, responseTo: message.id, payload: activityUpdateResult, }); } return responses; } async handleResponsesClientSocketMessage( message: ResponsesClientSocketMessage, ): Promise { const { viewer } = this; invariant(viewer, 'should be set'); const { clientResponses } = message.payload; const { stateCheckStatus } = await processClientResponses( viewer, clientResponses, ); const serverRequests = []; if (stateCheckStatus && stateCheckStatus.status !== 'state_check') { const { sessionUpdate, checkStateRequest } = await checkState( viewer, stateCheckStatus, viewer.calendarQuery, ); if (sessionUpdate) { await commitSessionUpdate(viewer, sessionUpdate); this.setStateCheckConditions({ stateCheckOngoing: false }); } if (checkStateRequest) { serverRequests.push(checkStateRequest); } } // We send a response message regardless of whether we have any requests, // since we need to ack the client's responses return [ { type: serverSocketMessageTypes.REQUESTS, responseTo: message.id, payload: { serverRequests }, }, ]; } handlePingClientSocketMessage( message: PingClientSocketMessage, ): ServerServerSocketMessage[] { return [ { type: serverSocketMessageTypes.PONG, responseTo: message.id, }, ]; } async handleAckUpdatesClientSocketMessage( message: AckUpdatesClientSocketMessage, ): Promise { const { viewer } = this; invariant(viewer, 'should be set'); const { currentAsOf } = message.payload; await Promise.all([ deleteUpdatesBeforeTimeTargetingSession(viewer, currentAsOf), commitSessionUpdate(viewer, { lastUpdate: currentAsOf }), ]); return []; } async handleAPIRequestClientSocketMessage( message: APIRequestClientSocketMessage, ): Promise { if (!endpointIsSocketSafe(message.payload.endpoint)) { throw new ServerError('endpoint_unsafe_for_socket'); } const { viewer } = this; invariant(viewer, 'should be set'); const responder = jsonEndpoints[message.payload.endpoint]; await policiesValidator(viewer, responder.requiredPolicies); const response = await responder.responder(viewer, message.payload.input); return [ { type: serverSocketMessageTypes.API_RESPONSE, responseTo: message.id, payload: response, }, ]; } onRedisMessage = async (message: RedisMessage) => { try { await this.processRedisMessage(message); } catch (e) { console.warn(e); } }; async processRedisMessage(message: RedisMessage) { if (message.type === redisMessageTypes.START_SUBSCRIPTION) { this.ws.terminate(); } else if (message.type === redisMessageTypes.NEW_UPDATES) { const { viewer } = this; invariant(viewer, 'should be set'); if (message.ignoreSession && message.ignoreSession === viewer.session) { return; } const rawUpdateInfos = message.updates; this.redisPromiseResolver.add( (async () => { const { updateInfos, userInfos } = await fetchUpdateInfosWithRawUpdateInfos(rawUpdateInfos, { viewer, }); if (updateInfos.length === 0) { console.warn( 'could not get any UpdateInfos from redisMessageTypes.NEW_UPDATES', ); return null; } this.markActivityOccurred(); return { type: serverSocketMessageTypes.UPDATES, payload: { updatesResult: { currentAsOf: mostRecentUpdateTimestamp([...updateInfos], 0), newUpdates: updateInfos, }, userInfos: values(userInfos), }, }; })(), ); } else if (message.type === redisMessageTypes.NEW_MESSAGES) { const { viewer } = this; invariant(viewer, 'should be set'); const rawMessageInfos = message.messages; const messageFetchResult = getMessageFetchResultFromRedisMessages( viewer, rawMessageInfos, ); if (messageFetchResult.rawMessageInfos.length === 0) { console.warn( 'could not get any rawMessageInfos from ' + 'redisMessageTypes.NEW_MESSAGES', ); return; } this.redisPromiseResolver.add( (async () => { this.markActivityOccurred(); return { type: serverSocketMessageTypes.MESSAGES, payload: { messagesResult: { rawMessageInfos: messageFetchResult.rawMessageInfos, truncationStatuses: messageFetchResult.truncationStatuses, currentAsOf: mostRecentMessageTimestamp( messageFetchResult.rawMessageInfos, 0, ), }, }, }; })(), ); } } onSuccessfulConnection() { if (this.ws.readyState !== 1) { return; } this.handleStateCheckConditionsUpdate(); } // The Socket will timeout by calling this.ws.terminate() // serverRequestSocketTimeout milliseconds after the last // time resetTimeout is called resetTimeout = _debounce( () => this.ws.terminate(), serverRequestSocketTimeout, ); debouncedAfterActivity = _debounce( () => this.setStateCheckConditions({ activityRecentlyOccurred: false }), stateCheckInactivityActivationInterval, ); markActivityOccurred = () => { if (this.ws.readyState !== 1) { return; } this.setStateCheckConditions({ activityRecentlyOccurred: true }); this.debouncedAfterActivity(); }; clearStateCheckTimeout() { const { stateCheckTimeoutID } = this; if (stateCheckTimeoutID) { clearTimeout(stateCheckTimeoutID); this.stateCheckTimeoutID = null; } } setStateCheckConditions(newConditions: Shape) { this.stateCheckConditions = { ...this.stateCheckConditions, ...newConditions, }; this.handleStateCheckConditionsUpdate(); } get stateCheckCanStart() { return Object.values(this.stateCheckConditions).every(cond => !cond); } handleStateCheckConditionsUpdate() { if (!this.stateCheckCanStart) { this.clearStateCheckTimeout(); return; } if (this.stateCheckTimeoutID) { return; } const { viewer } = this; if (!viewer) { return; } const timeUntilStateCheck = viewer.sessionLastValidated + sessionCheckFrequency - Date.now(); if (timeUntilStateCheck <= 0) { this.initiateStateCheck(); } else { this.stateCheckTimeoutID = setTimeout( this.initiateStateCheck, timeUntilStateCheck, ); } } initiateStateCheck = async () => { this.setStateCheckConditions({ stateCheckOngoing: true }); const { viewer } = this; invariant(viewer, 'should be set'); const { checkStateRequest } = await checkState( viewer, { status: 'state_check' }, viewer.calendarQuery, ); invariant(checkStateRequest, 'should be set'); this.sendMessage({ type: serverSocketMessageTypes.REQUESTS, payload: { serverRequests: [checkStateRequest] }, }); }; } export { onConnection }; diff --git a/lib/socket/socket.react.js b/lib/socket/socket.react.js index 58ff02099..db9206da3 100644 --- a/lib/socket/socket.react.js +++ b/lib/socket/socket.react.js @@ -1,774 +1,772 @@ // @flow import invariant from 'invariant'; import _throttle from 'lodash/throttle.js'; import * as React from 'react'; import ActivityHandler from './activity-handler.react.js'; import APIRequestHandler from './api-request-handler.react.js'; import CalendarQueryHandler from './calendar-query-handler.react.js'; import { InflightRequests } from './inflight-requests.js'; import MessageHandler from './message-handler.react.js'; import ReportHandler from './report-handler.react.js'; import RequestResponseHandler from './request-response-handler.react.js'; import UpdateHandler from './update-handler.react.js'; import { updateActivityActionTypes } from '../actions/activity-actions.js'; import { logOutActionTypes } from '../actions/user-actions.js'; import { unsupervisedBackgroundActionType } from '../reducers/lifecycle-state-reducer.js'; import { pingFrequency, serverRequestSocketTimeout, clientRequestVisualTimeout, clientRequestSocketTimeout, } from '../shared/timeouts.js'; import { logInActionSources, type LogOutResult, } from '../types/account-types.js'; import { isWebPlatform } from '../types/device-types.js'; import type { CalendarQuery } from '../types/entry-types.js'; import { forcePolicyAcknowledgmentActionType } from '../types/policy-types.js'; import type { Dispatch } from '../types/redux-types.js'; import { serverRequestTypes, type ClientClientResponse, type ClientServerRequest, } from '../types/request-types.js'; import { type SessionState, type SessionIdentification, type PreRequestUserState, } from '../types/session-types.js'; import { clientSocketMessageTypes, type ClientClientSocketMessage, serverSocketMessageTypes, type ClientServerSocketMessage, stateSyncPayloadTypes, fullStateSyncActionType, incrementalStateSyncActionType, updateConnectionStatusActionType, type ConnectionInfo, type ClientInitialClientSocketMessage, type ClientResponsesClientSocketMessage, type PingClientSocketMessage, type AckUpdatesClientSocketMessage, type APIRequestClientSocketMessage, type ClientSocketMessageWithoutID, type SocketListener, type ConnectionStatus, setLateResponseActionType, type CommTransportLayer, } from '../types/socket-types.js'; import { actionLogger } from '../utils/action-logger.js'; import type { DispatchActionPromise } from '../utils/action-utils.js'; import { setNewSessionActionType, fetchNewCookieFromNativeCredentials, } from '../utils/action-utils.js'; import { getConfig } from '../utils/config.js'; import { ServerError, SocketTimeout, SocketOffline } from '../utils/errors.js'; import { promiseAll } from '../utils/promises.js'; import sleep from '../utils/sleep.js'; const remainingTimeAfterVisualTimeout = clientRequestSocketTimeout - clientRequestVisualTimeout; export type BaseSocketProps = { +detectUnsupervisedBackgroundRef?: ( detectUnsupervisedBackground: (alreadyClosed: boolean) => boolean, ) => void, }; type Props = { ...BaseSocketProps, // Redux state +active: boolean, +openSocket: () => CommTransportLayer, +getClientResponses: ( activeServerRequests: $ReadOnlyArray, ) => Promise<$ReadOnlyArray>, +activeThread: ?string, +sessionStateFunc: () => SessionState, +sessionIdentification: SessionIdentification, +cookie: ?string, +urlPrefix: string, +connection: ConnectionInfo, +currentCalendarQuery: () => CalendarQuery, +canSendReports: boolean, +frozen: boolean, +preRequestUserState: PreRequestUserState, +noDataAfterPolicyAcknowledgment?: boolean, // Redux dispatch functions +dispatch: Dispatch, +dispatchActionPromise: DispatchActionPromise, // async functions that hit server APIs +logOut: (preRequestUserState: PreRequestUserState) => Promise, +socketCrashLoopRecovery?: () => Promise, }; type State = { +inflightRequests: ?InflightRequests, }; class Socket extends React.PureComponent { state: State = { inflightRequests: null, }; socket: ?CommTransportLayer; nextClientMessageID: number = 0; listeners: Set = new Set(); pingTimeoutID: ?TimeoutID; messageLastReceived: ?number; - initialPlatformDetailsSent: boolean = isWebPlatform( - getConfig().platformDetails.platform, - ); + initialPlatformDetailsSent: boolean = false; reopenConnectionAfterClosing: boolean = false; invalidationRecoveryInProgress: boolean = false; initializedWithUserState: ?PreRequestUserState; failuresAfterPolicyAcknowledgment: number = 0; openSocket(newStatus: ConnectionStatus) { if ( this.props.frozen || (!isWebPlatform(getConfig().platformDetails.platform) && (!this.props.cookie || !this.props.cookie.startsWith('user='))) ) { return; } if (this.socket) { const { status } = this.props.connection; if (status === 'forcedDisconnecting') { this.reopenConnectionAfterClosing = true; return; } else if (status === 'disconnecting' && this.socket.readyState === 1) { this.markSocketInitialized(); return; } else if ( status === 'connected' || status === 'connecting' || status === 'reconnecting' ) { return; } if (this.socket.readyState < 2) { this.socket.close(); console.log(`this.socket seems open, but Redux thinks it's ${status}`); } } this.props.dispatch({ type: updateConnectionStatusActionType, payload: { status: newStatus }, }); const socket = this.props.openSocket(); const openObject = {}; socket.onopen = () => { if (this.socket === socket) { this.initializeSocket(); openObject.initializeMessageSent = true; } }; socket.onmessage = this.receiveMessage; socket.onclose = () => { if (this.socket === socket) { this.onClose(); } }; this.socket = socket; (async () => { await sleep(clientRequestVisualTimeout); if (this.socket !== socket || openObject.initializeMessageSent) { return; } this.setLateResponse(-1, true); await sleep(remainingTimeAfterVisualTimeout); if (this.socket !== socket || openObject.initializeMessageSent) { return; } this.finishClosingSocket(); })(); this.setState({ inflightRequests: new InflightRequests({ timeout: () => { if (this.socket === socket) { this.finishClosingSocket(); } }, setLateResponse: (messageID: number, isLate: boolean) => { if (this.socket === socket) { this.setLateResponse(messageID, isLate); } }, }), }); } markSocketInitialized() { this.props.dispatch({ type: updateConnectionStatusActionType, payload: { status: 'connected' }, }); this.resetPing(); } closeSocket( // This param is a hack. When closing a socket there is a race between this // function and the one to propagate the activity update. We make sure that // the activity update wins the race by passing in this param. activityUpdatePending: boolean, ) { const { status } = this.props.connection; if (status === 'disconnected') { return; } else if (status === 'disconnecting' || status === 'forcedDisconnecting') { this.reopenConnectionAfterClosing = false; return; } this.stopPing(); this.props.dispatch({ type: updateConnectionStatusActionType, payload: { status: 'disconnecting' }, }); if (!activityUpdatePending) { this.finishClosingSocket(); } } forceCloseSocket() { this.stopPing(); const { status } = this.props.connection; if (status !== 'forcedDisconnecting' && status !== 'disconnected') { this.props.dispatch({ type: updateConnectionStatusActionType, payload: { status: 'forcedDisconnecting' }, }); } this.finishClosingSocket(); } finishClosingSocket(receivedResponseTo?: ?number) { const { inflightRequests } = this.state; if ( inflightRequests && !inflightRequests.allRequestsResolvedExcept(receivedResponseTo) ) { return; } if (this.socket && this.socket.readyState < 2) { // If it's not closing already, close it this.socket.close(); } this.socket = null; this.stopPing(); this.setState({ inflightRequests: null }); if (this.props.connection.status !== 'disconnected') { this.props.dispatch({ type: updateConnectionStatusActionType, payload: { status: 'disconnected' }, }); } if (this.reopenConnectionAfterClosing) { this.reopenConnectionAfterClosing = false; if (this.props.active) { this.openSocket('connecting'); } } } reconnect: $Call void, number> = _throttle( () => this.openSocket('reconnecting'), 2000, ); componentDidMount() { if (this.props.detectUnsupervisedBackgroundRef) { this.props.detectUnsupervisedBackgroundRef( this.detectUnsupervisedBackground, ); } if (this.props.active) { this.openSocket('connecting'); } } componentWillUnmount() { this.closeSocket(false); this.reconnect.cancel(); } componentDidUpdate(prevProps: Props) { if (this.props.active && !prevProps.active) { this.openSocket('connecting'); } else if (!this.props.active && prevProps.active) { this.closeSocket(!!prevProps.activeThread); } else if ( this.props.active && prevProps.openSocket !== this.props.openSocket ) { // This case happens when the baseURL/urlPrefix is changed this.reopenConnectionAfterClosing = true; this.forceCloseSocket(); } else if ( this.props.active && this.props.connection.status === 'disconnected' && prevProps.connection.status !== 'disconnected' && !this.invalidationRecoveryInProgress ) { this.reconnect(); } } render(): React.Node { // It's important that APIRequestHandler get rendered first here. This is so // that it is registered with Redux first, so that its componentDidUpdate // processes before the other Handlers. This allows APIRequestHandler to // register itself with action-utils before other Handlers call // dispatchActionPromise in response to the componentDidUpdate triggered by // the same Redux change (state.connection.status). return ( ); } sendMessageWithoutID: (message: ClientSocketMessageWithoutID) => number = message => { const id = this.nextClientMessageID++; // These conditions all do the same thing and the runtime checks are only // necessary for Flow if (message.type === clientSocketMessageTypes.INITIAL) { this.sendMessage( ({ ...message, id }: ClientInitialClientSocketMessage), ); } else if (message.type === clientSocketMessageTypes.RESPONSES) { this.sendMessage( ({ ...message, id }: ClientResponsesClientSocketMessage), ); } else if (message.type === clientSocketMessageTypes.PING) { this.sendMessage(({ ...message, id }: PingClientSocketMessage)); } else if (message.type === clientSocketMessageTypes.ACK_UPDATES) { this.sendMessage(({ ...message, id }: AckUpdatesClientSocketMessage)); } else if (message.type === clientSocketMessageTypes.API_REQUEST) { this.sendMessage(({ ...message, id }: APIRequestClientSocketMessage)); } return id; }; sendMessage(message: ClientClientSocketMessage) { const socket = this.socket; invariant(socket, 'should be set'); socket.send(JSON.stringify(message)); } static messageFromEvent(event: MessageEvent): ?ClientServerSocketMessage { if (typeof event.data !== 'string') { console.log('socket received a non-string message'); return null; } try { return JSON.parse(event.data); } catch (e) { console.log(e); return null; } } receiveMessage: (event: MessageEvent) => Promise = async event => { const message = Socket.messageFromEvent(event); if (!message) { return; } this.failuresAfterPolicyAcknowledgment = 0; const { inflightRequests } = this.state; if (!inflightRequests) { // inflightRequests can be falsey here if we receive a message after we've // begun shutting down the socket. It's possible for a React Native // WebSocket to deliver a message even after close() is called on it. In // this case the message is probably a PONG, which we can safely ignore. // If it's not a PONG, it has to be something server-initiated (like // UPDATES or MESSAGES), since InflightRequests.allRequestsResolvedExcept // will wait for all responses to client-initiated requests to be // delivered before closing a socket. UPDATES and MESSAGES are both // checkpointed on the client, so should be okay to just ignore here and // redownload them later, probably in an incremental STATE_SYNC. return; } // If we receive any message, that indicates that our connection is healthy, // so we can reset the ping timeout. this.resetPing(); inflightRequests.resolveRequestsForMessage(message); const { status } = this.props.connection; if (status === 'disconnecting' || status === 'forcedDisconnecting') { this.finishClosingSocket( // We do this for Flow message.responseTo !== undefined ? message.responseTo : null, ); } for (const listener of this.listeners) { listener(message); } if (message.type === serverSocketMessageTypes.ERROR) { const { message: errorMessage, payload } = message; if (payload) { console.log(`socket sent error ${errorMessage} with payload`, payload); } else { console.log(`socket sent error ${errorMessage}`); } if (errorMessage === 'policies_not_accepted') { this.props.dispatch({ type: forcePolicyAcknowledgmentActionType, payload, }); } } else if (message.type === serverSocketMessageTypes.AUTH_ERROR) { const { sessionChange } = message; const cookie = sessionChange ? sessionChange.cookie : this.props.cookie; this.invalidationRecoveryInProgress = true; const recoverySessionChange = await fetchNewCookieFromNativeCredentials( this.props.dispatch, cookie, this.props.urlPrefix, logInActionSources.socketAuthErrorResolutionAttempt, ); if (!recoverySessionChange && sessionChange) { // This should only happen in the cookieSources.BODY (native) case when // the resolution attempt failed const { cookie: newerCookie, currentUserInfo } = sessionChange; this.props.dispatch({ type: setNewSessionActionType, payload: { sessionChange: { cookieInvalidated: true, currentUserInfo, cookie: newerCookie, }, preRequestUserState: this.initializedWithUserState, error: null, logInActionSource: logInActionSources.socketAuthErrorResolutionAttempt, }, }); } else if (!recoverySessionChange) { this.props.dispatchActionPromise( logOutActionTypes, this.props.logOut(this.props.preRequestUserState), ); } this.invalidationRecoveryInProgress = false; } }; addListener: (listener: SocketListener) => void = listener => { this.listeners.add(listener); }; removeListener: (listener: SocketListener) => void = listener => { this.listeners.delete(listener); }; onClose: () => void = () => { const { status } = this.props.connection; this.socket = null; this.stopPing(); if (this.state.inflightRequests) { this.state.inflightRequests.rejectAll(new Error('socket closed')); this.setState({ inflightRequests: null }); } const handled = this.detectUnsupervisedBackground(true); if (!handled && status !== 'disconnected') { this.props.dispatch({ type: updateConnectionStatusActionType, payload: { status: 'disconnected' }, }); } }; async sendInitialMessage() { const { inflightRequests } = this.state; invariant( inflightRequests, 'inflightRequests falsey inside sendInitialMessage', ); const messageID = this.nextClientMessageID++; const promises = {}; const clientResponses = []; if (!this.initialPlatformDetailsSent) { this.initialPlatformDetailsSent = true; clientResponses.push({ type: serverRequestTypes.PLATFORM_DETAILS, platformDetails: getConfig().platformDetails, }); } const { queuedActivityUpdates } = this.props.connection; if (queuedActivityUpdates.length > 0) { clientResponses.push({ type: serverRequestTypes.INITIAL_ACTIVITY_UPDATES, activityUpdates: queuedActivityUpdates, }); promises.activityUpdateMessage = inflightRequests.fetchResponse( messageID, serverSocketMessageTypes.ACTIVITY_UPDATE_RESPONSE, ); } const sessionState = this.props.sessionStateFunc(); const { sessionIdentification } = this.props; const initialMessage = { type: clientSocketMessageTypes.INITIAL, id: messageID, payload: { clientResponses, sessionState, sessionIdentification, }, }; this.initializedWithUserState = this.props.preRequestUserState; this.sendMessage(initialMessage); promises.stateSyncMessage = inflightRequests.fetchResponse( messageID, serverSocketMessageTypes.STATE_SYNC, ); const { stateSyncMessage, activityUpdateMessage } = await promiseAll( promises, ); if (activityUpdateMessage) { this.props.dispatch({ type: updateActivityActionTypes.success, payload: { activityUpdates: queuedActivityUpdates, result: activityUpdateMessage.payload, }, }); } if (stateSyncMessage.payload.type === stateSyncPayloadTypes.FULL) { const { sessionID, type, ...actionPayload } = stateSyncMessage.payload; this.props.dispatch({ type: fullStateSyncActionType, payload: { ...actionPayload, calendarQuery: sessionState.calendarQuery, }, }); if (sessionID !== null && sessionID !== undefined) { invariant( this.initializedWithUserState, 'initializedWithUserState should be set when state sync received', ); this.props.dispatch({ type: setNewSessionActionType, payload: { sessionChange: { cookieInvalidated: false, sessionID }, preRequestUserState: this.initializedWithUserState, error: null, logInActionSource: undefined, }, }); } } else { const { type, ...actionPayload } = stateSyncMessage.payload; this.props.dispatch({ type: incrementalStateSyncActionType, payload: { ...actionPayload, calendarQuery: sessionState.calendarQuery, }, }); } const currentAsOf = stateSyncMessage.payload.type === stateSyncPayloadTypes.FULL ? stateSyncMessage.payload.updatesCurrentAsOf : stateSyncMessage.payload.updatesResult.currentAsOf; this.sendMessageWithoutID({ type: clientSocketMessageTypes.ACK_UPDATES, payload: { currentAsOf }, }); this.markSocketInitialized(); } initializeSocket: (retriesLeft?: number) => Promise = async ( retriesLeft = 1, ) => { try { await this.sendInitialMessage(); } catch (e) { if (this.props.noDataAfterPolicyAcknowledgment) { this.failuresAfterPolicyAcknowledgment++; } else { this.failuresAfterPolicyAcknowledgment = 0; } if ( this.failuresAfterPolicyAcknowledgment >= 2 && this.props.socketCrashLoopRecovery ) { this.failuresAfterPolicyAcknowledgment = 0; try { await this.props.socketCrashLoopRecovery(); } catch (error) { console.log(error); this.props.dispatchActionPromise( logOutActionTypes, this.props.logOut(this.props.preRequestUserState), ); } return; } console.log(e); const { status } = this.props.connection; if ( e instanceof SocketTimeout || e instanceof SocketOffline || (status !== 'connecting' && status !== 'reconnecting') ) { // This indicates that the socket will be closed. Do nothing, since the // connection status update will trigger a reconnect. } else if ( retriesLeft === 0 || (e instanceof ServerError && e.message !== 'unknown_error') ) { if (e.message === 'not_logged_in') { this.props.dispatchActionPromise( logOutActionTypes, this.props.logOut(this.props.preRequestUserState), ); } else if (this.socket) { this.socket.close(); } } else { await this.initializeSocket(retriesLeft - 1); } } }; stopPing() { if (this.pingTimeoutID) { clearTimeout(this.pingTimeoutID); this.pingTimeoutID = null; } } resetPing() { this.stopPing(); const socket = this.socket; this.messageLastReceived = Date.now(); this.pingTimeoutID = setTimeout(() => { if (this.socket === socket) { this.sendPing(); } }, pingFrequency); } async sendPing() { if (this.props.connection.status !== 'connected') { // This generally shouldn't happen because anything that changes the // connection status should call stopPing(), but it's good to make sure return; } const messageID = this.sendMessageWithoutID({ type: clientSocketMessageTypes.PING, }); try { invariant( this.state.inflightRequests, 'inflightRequests falsey inside sendPing', ); await this.state.inflightRequests.fetchResponse( messageID, serverSocketMessageTypes.PONG, ); } catch (e) {} } setLateResponse: (messageID: number, isLate: boolean) => void = ( messageID, isLate, ) => { this.props.dispatch({ type: setLateResponseActionType, payload: { messageID, isLate }, }); }; cleanUpServerTerminatedSocket() { if (this.socket && this.socket.readyState < 2) { this.socket.close(); } else { this.onClose(); } } detectUnsupervisedBackground: (alreadyClosed: boolean) => boolean = alreadyClosed => { // On native, sometimes the app is backgrounded without the proper // callbacks getting triggered. This leaves us in an incorrect state for // two reasons: // (1) The connection is still considered to be active, causing API // requests to be processed via socket and failing. // (2) We rely on flipping foreground state in Redux to detect activity // changes, and thus won't think we need to update activity. if ( this.props.connection.status !== 'connected' || !this.messageLastReceived || this.messageLastReceived + serverRequestSocketTimeout >= Date.now() || (actionLogger.mostRecentActionTime && actionLogger.mostRecentActionTime + 3000 < Date.now()) ) { return false; } if (!alreadyClosed) { this.cleanUpServerTerminatedSocket(); } this.props.dispatch({ type: unsupervisedBackgroundActionType, payload: null, }); return true; }; } export default Socket;