diff --git a/lib/reducers/message-reducer.js b/lib/reducers/message-reducer.js index 40d5d8f4b..9aa0a277f 100644 --- a/lib/reducers/message-reducer.js +++ b/lib/reducers/message-reducer.js @@ -1,1535 +1,1534 @@ // @flow import invariant from 'invariant'; import _difference from 'lodash/fp/difference'; import _flow from 'lodash/fp/flow'; import _isEqual from 'lodash/fp/isEqual'; import _keyBy from 'lodash/fp/keyBy'; import _map from 'lodash/fp/map'; import _mapKeys from 'lodash/fp/mapKeys'; import _mapValues from 'lodash/fp/mapValues'; import _omit from 'lodash/fp/omit'; import _omitBy from 'lodash/fp/omitBy'; import _pick from 'lodash/fp/pick'; import _pickBy from 'lodash/fp/pickBy'; import _uniq from 'lodash/fp/uniq'; import { createEntryActionTypes, saveEntryActionTypes, deleteEntryActionTypes, restoreEntryActionTypes, } from '../actions/entry-actions'; import { fetchMessagesBeforeCursorActionTypes, fetchMostRecentMessagesActionTypes, sendTextMessageActionTypes, sendMultimediaMessageActionTypes, saveMessagesActionType, processMessagesActionType, messageStorePruneActionType, createLocalMessageActionType, fetchSingleMostRecentMessagesFromThreadsActionTypes, setMessageStoreMessages, } from '../actions/message-actions'; import { changeThreadSettingsActionTypes, deleteThreadActionTypes, leaveThreadActionTypes, newThreadActionTypes, removeUsersFromThreadActionTypes, changeThreadMemberRolesActionTypes, joinThreadActionTypes, } from '../actions/thread-actions'; import { updateMultimediaMessageMediaActionType } from '../actions/upload-actions'; import { logOutActionTypes, deleteAccountActionTypes, logInActionTypes, registerActionTypes, } from '../actions/user-actions'; import { locallyUniqueToRealizedThreadIDsSelector } from '../selectors/thread-selectors'; import { messageID, combineTruncationStatuses, sortMessageInfoList, sortMessageIDs, mergeThreadMessageInfos, } from '../shared/message-utils'; import { threadHasPermission, threadInChatList, threadIsPending, } from '../shared/thread-utils'; import threadWatcher from '../shared/thread-watcher'; import { unshimMessageInfos } from '../shared/unshim-utils'; import type { Media, Image } from '../types/media-types'; import { type RawMessageInfo, type LocalMessageInfo, type MessageStore, type ReplaceMessageOperation, type MessageStoreOperation, type MessageTruncationStatus, type ThreadMessageInfo, type MessageTruncationStatuses, messageTruncationStatus, messageTypes, defaultNumberPerThread, } from '../types/message-types'; import type { RawImagesMessageInfo } from '../types/messages/images'; import type { RawMediaMessageInfo } from '../types/messages/media'; import { type BaseAction, rehydrateActionType } from '../types/redux-types'; import { processServerRequestsActionType } from '../types/request-types'; import { fullStateSyncActionType, incrementalStateSyncActionType, } from '../types/socket-types'; import { type RawThreadInfo, threadPermissions } from '../types/thread-types'; import { updateTypes, type ClientUpdateInfo, processUpdatesActionType, } from '../types/update-types'; import { setNewSessionActionType } from '../utils/action-utils'; -import { isDev } from '../utils/dev-utils'; import { translateClientDBMessageInfosToRawMessageInfos } from '../utils/message-ops-utils'; import { assertObjectsAreEqual } from '../utils/objects'; -const PROCESSED_MSG_STORE_INVARIANTS_DISABLED = !isDev; +const PROCESSED_MSG_STORE_INVARIANTS_DISABLED = false; const _mapValuesWithKeys = _mapValues.convert({ cap: false }); // Input must already be ordered! function threadsToMessageIDsFromMessageInfos( orderedMessageInfos: $ReadOnlyArray, ): { [threadID: string]: string[] } { const threads: { [threadID: string]: string[] } = {}; for (const messageInfo of orderedMessageInfos) { const key = messageID(messageInfo); if (!threads[messageInfo.threadID]) { threads[messageInfo.threadID] = [key]; } else { threads[messageInfo.threadID].push(key); } } return threads; } function threadIsWatched( threadID: string, threadInfo: ?RawThreadInfo, watchedIDs: $ReadOnlyArray, ) { return ( threadIsPending(threadID) || (threadInfo && threadHasPermission(threadInfo, threadPermissions.VISIBLE) && (threadInChatList(threadInfo) || watchedIDs.includes(threadID))) ); } function assertMessageStoreMessagesAreEqual( processedMessageStore: MessageStore, expectedMessageStore: MessageStore, location: string, ) { if (PROCESSED_MSG_STORE_INVARIANTS_DISABLED) { return; } assertObjectsAreEqual( processedMessageStore.messages, expectedMessageStore.messages, `MessageStore.messages - ${location}`, ); } type FreshMessageStoreResult = { +messageStoreOperations: $ReadOnlyArray, +messageStore: MessageStore, }; function freshMessageStore( messageInfos: $ReadOnlyArray, truncationStatus: { [threadID: string]: MessageTruncationStatus }, currentAsOf: number, threadInfos: { +[threadID: string]: RawThreadInfo }, ): FreshMessageStoreResult { const unshimmed = unshimMessageInfos(messageInfos); const orderedMessageInfos = sortMessageInfoList(unshimmed); const messages = _keyBy(messageID)(orderedMessageInfos); const messageStoreOperations = orderedMessageInfos.map(messageInfo => ({ type: 'replace', payload: { id: messageID(messageInfo), messageInfo }, })); const threadsToMessageIDs = threadsToMessageIDsFromMessageInfos( orderedMessageInfos, ); const lastPruned = Date.now(); const threads = _mapValuesWithKeys( (messageIDs: string[], threadID: string) => ({ messageIDs, startReached: truncationStatus[threadID] === messageTruncationStatus.EXHAUSTIVE, lastNavigatedTo: 0, lastPruned, }), )(threadsToMessageIDs); const watchedIDs = threadWatcher.getWatchedIDs(); for (const threadID in threadInfos) { const threadInfo = threadInfos[threadID]; if ( threads[threadID] || !threadIsWatched(threadID, threadInfo, watchedIDs) ) { continue; } threads[threadID] = { messageIDs: [], startReached: false, lastNavigatedTo: 0, lastPruned, }; } return { messageStoreOperations, messageStore: { messages, threads, local: {}, currentAsOf }, }; } type ReassignmentResult = { +messageStoreOperations: $ReadOnlyArray, +messageStore: MessageStore, +reassignedThreadIDs: $ReadOnlyArray, }; function reassignMessagesToRealizedThreads( messageStore: MessageStore, threadInfos: { +[threadID: string]: RawThreadInfo }, ): ReassignmentResult { const locallyUniqueToRealizedThreadIDs = locallyUniqueToRealizedThreadIDsSelector( threadInfos, ); const messageStoreOperations: MessageStoreOperation[] = []; const messages = {}; for (const storeMessageID in messageStore.messages) { const message = messageStore.messages[storeMessageID]; const newThreadID = locallyUniqueToRealizedThreadIDs.get(message.threadID); messages[storeMessageID] = newThreadID ? { ...message, threadID: newThreadID, time: threadInfos[newThreadID]?.creationTime ?? message.time, } : message; if (!newThreadID) { continue; } const updateMsgOperation: ReplaceMessageOperation = { type: 'replace', payload: { id: storeMessageID, messageInfo: messages[storeMessageID] }, }; messageStoreOperations.push(updateMsgOperation); } const threads = {}; const reassignedThreadIDs = []; for (const threadID in messageStore.threads) { const threadMessageInfo = messageStore.threads[threadID]; const newThreadID = locallyUniqueToRealizedThreadIDs.get(threadID); if (!newThreadID) { threads[threadID] = threadMessageInfo; continue; } const realizedThread = messageStore.threads[newThreadID]; if (!realizedThread) { reassignedThreadIDs.push(newThreadID); threads[newThreadID] = threadMessageInfo; continue; } threads[newThreadID] = mergeThreadMessageInfos( threadMessageInfo, realizedThread, messages, ); } return { messageStoreOperations, messageStore: { ...messageStore, threads, messages, }, reassignedThreadIDs, }; } type MergeNewMessagesResult = { +messageStoreOperations: $ReadOnlyArray, +messageStore: MessageStore, }; // oldMessageStore is from the old state // newMessageInfos, truncationStatus come from server function mergeNewMessages( oldMessageStore: MessageStore, newMessageInfos: $ReadOnlyArray, truncationStatus: { [threadID: string]: MessageTruncationStatus }, threadInfos: { +[threadID: string]: RawThreadInfo }, actionType: *, ): MergeNewMessagesResult { const { messageStoreOperations: updateWithLatestThreadInfosOps, messageStore: updatedMessageStore, reassignedThreadIDs, } = updateMessageStoreWithLatestThreadInfos(oldMessageStore, threadInfos); const messageStoreAfterUpdateOps = processMessageStoreOperations( oldMessageStore, updateWithLatestThreadInfosOps, ); assertMessageStoreMessagesAreEqual( messageStoreAfterUpdateOps, updatedMessageStore, `${actionType}| reassignment and filtering`, ); const localIDsToServerIDs: Map = new Map(); const watchedThreadIDs = [ ...threadWatcher.getWatchedIDs(), ...reassignedThreadIDs, ]; const unshimmedNewMessages = unshimMessageInfos(newMessageInfos); const unshimmedNewMessagesOfWatchedThreads = unshimmedNewMessages.filter( msg => threadIsWatched( msg.threadID, threadInfos[msg.threadID], watchedThreadIDs, ), ); const orderedNewMessageInfos = _flow( _map((messageInfo: RawMessageInfo) => { const { id: inputID } = messageInfo; invariant(inputID, 'new messageInfos should have serverID'); invariant( !threadIsPending(messageInfo.threadID), 'new messageInfos should have realized thread id', ); const currentMessageInfo = updatedMessageStore.messages[inputID]; if ( messageInfo.type === messageTypes.TEXT || messageInfo.type === messageTypes.IMAGES || messageInfo.type === messageTypes.MULTIMEDIA ) { const { localID: inputLocalID } = messageInfo; const currentLocalMessageInfo = inputLocalID ? updatedMessageStore.messages[inputLocalID] : null; if (currentMessageInfo && currentMessageInfo.localID) { // If the client already has a RawMessageInfo with this serverID, keep // any localID associated with the existing one. This is because we // use localIDs as React keys and changing React keys leads to loss of // component state. (The conditional below is for Flow) if (messageInfo.type === messageTypes.TEXT) { messageInfo = { ...messageInfo, localID: currentMessageInfo.localID, }; } else if (messageInfo.type === messageTypes.MULTIMEDIA) { messageInfo = ({ ...messageInfo, localID: currentMessageInfo.localID, }: RawMediaMessageInfo); } else { messageInfo = ({ ...messageInfo, localID: currentMessageInfo.localID, }: RawImagesMessageInfo); } } else if (currentLocalMessageInfo && currentLocalMessageInfo.localID) { // If the client has a RawMessageInfo with this localID, but not with // the serverID, that means the message creation succeeded but the // success action never got processed. We set a key in // localIDsToServerIDs here to fix the messageIDs for the rest of the // MessageStore too. (The conditional below is for Flow) invariant(inputLocalID, 'inputLocalID should be set'); localIDsToServerIDs.set(inputLocalID, inputID); if (messageInfo.type === messageTypes.TEXT) { messageInfo = { ...messageInfo, localID: currentLocalMessageInfo.localID, }; } else if (messageInfo.type === messageTypes.MULTIMEDIA) { messageInfo = ({ ...messageInfo, localID: currentLocalMessageInfo.localID, }: RawMediaMessageInfo); } else { messageInfo = ({ ...messageInfo, localID: currentLocalMessageInfo.localID, }: RawImagesMessageInfo); } } else { // If neither the serverID nor the localID from the delivered // RawMessageInfo exists in the client store, then this message is // brand new to us. Ignore any localID provided by the server. // (The conditional below is for Flow) const { localID, ...rest } = messageInfo; if (rest.type === messageTypes.TEXT) { messageInfo = { ...rest }; } else if (rest.type === messageTypes.MULTIMEDIA) { messageInfo = ({ ...rest }: RawMediaMessageInfo); } else { messageInfo = ({ ...rest }: RawImagesMessageInfo); } } } else if ( currentMessageInfo && messageInfo.time > currentMessageInfo.time ) { // When thick threads will be introduced it will be possible for two // clients to create the same message (e.g. when they create the same // sidebar at the same time). We're going to use deterministic ids for // messages which should be unique within a thread and we have to find // a way for clients to agree which message to keep. We can't rely on // always choosing incoming messages nor messages from the store, // because a message that is in one user's store, will be send to // another user. One way to deal with it is to always choose a message // which is older, according to its timestamp. We can use this strategy // only for messages that can start a thread, because for other types // it might break the "contiguous" property of message ids (we can // consider selecting younger messages in that case, but for now we use // an invariant). invariant( messageInfo.type === messageTypes.CREATE_SIDEBAR || messageInfo.type === messageTypes.CREATE_THREAD || messageInfo.type === messageTypes.SIDEBAR_SOURCE, `Two different messages of type ${messageInfo.type} with the same ` + 'id found', ); return currentMessageInfo; } return _isEqual(messageInfo)(currentMessageInfo) ? currentMessageInfo : messageInfo; }), sortMessageInfoList, )(unshimmedNewMessagesOfWatchedThreads); const newMessageOps: MessageStoreOperation[] = []; const threadsToMessageIDs = threadsToMessageIDsFromMessageInfos( orderedNewMessageInfos, ); const oldMessageInfosToCombine = []; const threadsThatNeedMessageIDsResorted = []; const lastPruned = Date.now(); const local = {}; const threads = _flow( _mapValuesWithKeys((messageIDs: string[], threadID: string) => { const oldThread = updatedMessageStore.threads[threadID]; const truncate = truncationStatus[threadID]; if (!oldThread) { return { messageIDs, startReached: truncate === messageTruncationStatus.EXHAUSTIVE, lastNavigatedTo: 0, lastPruned, }; } let oldMessageIDsUnchanged = true; const oldMessageIDs = oldThread.messageIDs.map(oldID => { const newID = localIDsToServerIDs.get(oldID); if (newID !== null && newID !== undefined) { oldMessageIDsUnchanged = false; return newID; } return oldID; }); if (truncate === messageTruncationStatus.TRUNCATED) { // If the result set in the payload isn't contiguous with what we have // now, that means we need to dump what we have in the state and replace // it with the result set. We do this to achieve our two goals for the // messageStore: currentness and contiguousness. newMessageOps.push({ type: 'remove_messages_for_threads', payload: { threadIDs: [threadID] }, }); return { messageIDs, startReached: false, lastNavigatedTo: oldThread.lastNavigatedTo, lastPruned: oldThread.lastPruned, }; } const oldNotInNew = _difference(oldMessageIDs)(messageIDs); for (const id of oldNotInNew) { const oldMessageInfo = updatedMessageStore.messages[id]; invariant(oldMessageInfo, `could not find ${id} in messageStore`); oldMessageInfosToCombine.push(oldMessageInfo); const localInfo = updatedMessageStore.local[id]; if (localInfo) { local[id] = localInfo; } } const startReached = oldThread.startReached || truncate === messageTruncationStatus.EXHAUSTIVE; if (_difference(messageIDs)(oldMessageIDs).length === 0) { if (startReached === oldThread.startReached && oldMessageIDsUnchanged) { return oldThread; } return { messageIDs: oldMessageIDs, startReached, lastNavigatedTo: oldThread.lastNavigatedTo, lastPruned: oldThread.lastPruned, }; } const mergedMessageIDs = [...messageIDs, ...oldNotInNew]; threadsThatNeedMessageIDsResorted.push(threadID); return { messageIDs: mergedMessageIDs, startReached, lastNavigatedTo: oldThread.lastNavigatedTo, lastPruned: oldThread.lastPruned, }; }), _pickBy(thread => !!thread), )(threadsToMessageIDs); for (const threadID in updatedMessageStore.threads) { if (threads[threadID]) { continue; } let thread = updatedMessageStore.threads[threadID]; const truncate = truncationStatus[threadID]; if (truncate === messageTruncationStatus.EXHAUSTIVE) { thread = { ...thread, startReached: true, }; } threads[threadID] = thread; for (const id of thread.messageIDs) { const messageInfo = updatedMessageStore.messages[id]; if (messageInfo) { oldMessageInfosToCombine.push(messageInfo); } const localInfo = updatedMessageStore.local[id]; if (localInfo) { local[id] = localInfo; } } } const messages = _flow( sortMessageInfoList, _keyBy(messageID), )([...orderedNewMessageInfos, ...oldMessageInfosToCombine]); const newMessages = _keyBy(messageID)(orderedNewMessageInfos); for (const id in newMessages) { newMessageOps.push({ type: 'replace', payload: { id, messageInfo: newMessages[id] }, }); } if (localIDsToServerIDs.size > 0) { newMessageOps.push({ type: 'remove', payload: { ids: [...localIDsToServerIDs.keys()] }, }); } for (const threadID of threadsThatNeedMessageIDsResorted) { threads[threadID].messageIDs = sortMessageIDs(messages)( threads[threadID].messageIDs, ); } const currentAsOf = Math.max( orderedNewMessageInfos.length > 0 ? orderedNewMessageInfos[0].time : 0, updatedMessageStore.currentAsOf, ); const mergedMessageStore = { messages, threads, local, currentAsOf }; const processedMessageStore = processMessageStoreOperations( updatedMessageStore, newMessageOps, ); assertMessageStoreMessagesAreEqual( processedMessageStore, mergedMessageStore, `${actionType}|processed`, ); return { messageStoreOperations: [ ...updateWithLatestThreadInfosOps, ...newMessageOps, ], messageStore: mergedMessageStore, }; } type UpdateMessageStoreWithLatestThreadInfosResult = { +messageStoreOperations: $ReadOnlyArray, +messageStore: MessageStore, +reassignedThreadIDs: $ReadOnlyArray, }; function updateMessageStoreWithLatestThreadInfos( messageStore: MessageStore, threadInfos: { +[id: string]: RawThreadInfo }, ): UpdateMessageStoreWithLatestThreadInfosResult { const messageStoreOperations: MessageStoreOperation[] = []; const { messageStore: reassignedMessageStore, messageStoreOperations: reassignMessagesOps, reassignedThreadIDs, } = reassignMessagesToRealizedThreads(messageStore, threadInfos); messageStoreOperations.push(...reassignMessagesOps); const watchedIDs = [...threadWatcher.getWatchedIDs(), ...reassignedThreadIDs]; const watchedThreadInfos = _pickBy((threadInfo: RawThreadInfo) => threadIsWatched(threadInfo.id, threadInfo, watchedIDs), )(threadInfos); const filteredThreads = _pick(Object.keys(watchedThreadInfos))( reassignedMessageStore.threads, ); const messageIDsToRemove = []; const threadsToRemoveMessagesFrom = []; for (const threadID in reassignedMessageStore.threads) { if (watchedThreadInfos[threadID]) { continue; } threadsToRemoveMessagesFrom.push(threadID); for (const id of reassignedMessageStore.threads[threadID].messageIDs) { messageIDsToRemove.push(id); } } for (const threadID in threadInfos) { const threadInfo = threadInfos[threadID]; if ( threadIsWatched(threadID, threadInfo, watchedIDs) && !filteredThreads[threadID] ) { filteredThreads[threadID] = { messageIDs: [], // We can conclude that startReached, since no messages were returned. startReached: true, lastNavigatedTo: 0, lastPruned: Date.now(), }; } } messageStoreOperations.push({ type: 'remove_messages_for_threads', payload: { threadIDs: threadsToRemoveMessagesFrom }, }); return { messageStoreOperations, messageStore: { messages: _omit(messageIDsToRemove)(reassignedMessageStore.messages), threads: filteredThreads, local: _omit(messageIDsToRemove)(reassignedMessageStore.local), currentAsOf: reassignedMessageStore.currentAsOf, }, reassignedThreadIDs, }; } function ensureRealizedThreadIDIsUsedWhenPossible( payload: T, threadInfos: { +[id: string]: RawThreadInfo }, ): T { const locallyUniqueToRealizedThreadIDs = locallyUniqueToRealizedThreadIDsSelector( threadInfos, ); const realizedThreadID = locallyUniqueToRealizedThreadIDs.get( payload.threadID, ); return realizedThreadID ? { ...payload, threadID: realizedThreadID } : payload; } function processMessageStoreOperations( messageStore: MessageStore, messageStoreOperations: $ReadOnlyArray, ): MessageStore { if (messageStoreOperations.length === 0) { return messageStore; } let processedMessages = { ...messageStore.messages }; for (const operation of messageStoreOperations) { if (operation.type === 'replace') { processedMessages[operation.payload.id] = operation.payload.messageInfo; } else if (operation.type === 'remove') { for (const id of operation.payload.ids) { delete processedMessages[id]; } } else if (operation.type === 'remove_messages_for_threads') { for (const msgID in processedMessages) { if ( operation.payload.threadIDs.includes( processedMessages[msgID].threadID, ) ) { delete processedMessages[msgID]; } } } else if (operation.type === 'rekey') { processedMessages[operation.payload.to] = processedMessages[operation.payload.from]; delete processedMessages[operation.payload.from]; } else if (operation.type === 'remove_all') { processedMessages = {}; } } return { ...messageStore, messages: processedMessages }; } type ReduceMessageStoreResult = { +messageStoreOperations: $ReadOnlyArray, +messageStore: MessageStore, }; function reduceMessageStore( messageStore: MessageStore, action: BaseAction, newThreadInfos: { +[id: string]: RawThreadInfo }, ): ReduceMessageStoreResult { if (action.type === logInActionTypes.success) { const messagesResult = action.payload.messagesResult; const { messageStoreOperations, messageStore: freshStore, } = freshMessageStore( messagesResult.messageInfos, messagesResult.truncationStatus, messagesResult.currentAsOf, newThreadInfos, ); const processedMessageStore = processMessageStoreOperations( messageStore, messageStoreOperations, ); assertMessageStoreMessagesAreEqual( processedMessageStore, freshStore, action.type, ); return { messageStoreOperations, messageStore: freshStore }; } else if (action.type === incrementalStateSyncActionType) { if ( action.payload.messagesResult.rawMessageInfos.length === 0 && action.payload.updatesResult.newUpdates.length === 0 ) { return { messageStoreOperations: [], messageStore }; } const messagesResult = mergeUpdatesWithMessageInfos( action.payload.messagesResult.rawMessageInfos, action.payload.updatesResult.newUpdates, action.payload.messagesResult.truncationStatuses, ); const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, messagesResult.rawMessageInfos, messagesResult.truncationStatuses, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if (action.type === processUpdatesActionType) { if (action.payload.updatesResult.newUpdates.length === 0) { return { messageStoreOperations: [], messageStore }; } const messagesResult = mergeUpdatesWithMessageInfos( [], action.payload.updatesResult.newUpdates, ); const { messageStoreOperations, messageStore: newMessageStore, } = mergeNewMessages( messageStore, messagesResult.rawMessageInfos, messagesResult.truncationStatuses, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: { messages: newMessageStore.messages, threads: newMessageStore.threads, local: newMessageStore.local, currentAsOf: messageStore.currentAsOf, }, }; } else if ( action.type === fullStateSyncActionType || action.type === processMessagesActionType ) { const { messagesResult } = action.payload; const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, messagesResult.rawMessageInfos, messagesResult.truncationStatuses, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if ( action.type === fetchSingleMostRecentMessagesFromThreadsActionTypes.success ) { const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, action.payload.rawMessageInfos, action.payload.truncationStatuses, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if ( action.type === fetchMessagesBeforeCursorActionTypes.success || action.type === fetchMostRecentMessagesActionTypes.success ) { const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, action.payload.rawMessageInfos, { [action.payload.threadID]: action.payload.truncationStatus }, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if ( action.type === logOutActionTypes.success || action.type === deleteAccountActionTypes.success || action.type === deleteThreadActionTypes.success || action.type === leaveThreadActionTypes.success || action.type === setNewSessionActionType ) { const { messageStoreOperations, messageStore: filteredMessageStore, } = updateMessageStoreWithLatestThreadInfos(messageStore, newThreadInfos); const processedMessageStore = processMessageStoreOperations( messageStore, messageStoreOperations, ); assertMessageStoreMessagesAreEqual( processedMessageStore, filteredMessageStore, action.type, ); return { messageStoreOperations, messageStore: filteredMessageStore }; } else if (action.type === newThreadActionTypes.success) { const messagesResult = mergeUpdatesWithMessageInfos( action.payload.newMessageInfos, action.payload.updatesResult.newUpdates, ); const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, messagesResult.rawMessageInfos, messagesResult.truncationStatuses, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if (action.type === registerActionTypes.success) { const truncationStatuses = {}; for (const messageInfo of action.payload.rawMessageInfos) { truncationStatuses[messageInfo.threadID] = messageTruncationStatus.EXHAUSTIVE; } const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, action.payload.rawMessageInfos, truncationStatuses, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if ( action.type === changeThreadSettingsActionTypes.success || action.type === removeUsersFromThreadActionTypes.success || action.type === changeThreadMemberRolesActionTypes.success ) { const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, action.payload.newMessageInfos, { [action.payload.threadID]: messageTruncationStatus.UNCHANGED }, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if ( action.type === createEntryActionTypes.success || action.type === saveEntryActionTypes.success ) { const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, action.payload.newMessageInfos, { [action.payload.threadID]: messageTruncationStatus.UNCHANGED }, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if (action.type === deleteEntryActionTypes.success) { const payload = action.payload; if (payload) { const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, payload.newMessageInfos, { [payload.threadID]: messageTruncationStatus.UNCHANGED }, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } } else if (action.type === restoreEntryActionTypes.success) { const { threadID } = action.payload; const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, action.payload.newMessageInfos, { [threadID]: messageTruncationStatus.UNCHANGED }, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if (action.type === joinThreadActionTypes.success) { const messagesResult = mergeUpdatesWithMessageInfos( action.payload.rawMessageInfos, action.payload.updatesResult.newUpdates, ); const { messageStoreOperations, messageStore: mergedMessageStore, } = mergeNewMessages( messageStore, messagesResult.rawMessageInfos, messagesResult.truncationStatuses, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: mergedMessageStore }; } else if ( action.type === sendTextMessageActionTypes.started || action.type === sendMultimediaMessageActionTypes.started ) { const payload = ensureRealizedThreadIDIsUsedWhenPossible( action.payload, newThreadInfos, ); const { localID, threadID } = payload; invariant(localID, `localID should be set on ${action.type}`); const messageStoreOperations = [ { type: 'replace', payload: { id: localID, messageInfo: payload }, }, ]; const processedMessageStore = processMessageStoreOperations( messageStore, messageStoreOperations, ); const now = Date.now(); const messageIDs = messageStore.threads[threadID]?.messageIDs ?? []; if (messageStore.messages[localID]) { const messages = { ...messageStore.messages, [localID]: payload }; const local = _pickBy( (localInfo: LocalMessageInfo, key: string) => key !== localID, )(messageStore.local); const thread = messageStore.threads[threadID]; const threads = { ...messageStore.threads, [threadID]: { messageIDs: sortMessageIDs(messages)(messageIDs), startReached: thread?.startReached ?? true, lastNavigatedTo: thread?.lastNavigatedTo ?? now, lastPruned: thread?.lastPruned ?? now, }, }; const newMessageStore = { ...messageStore, messages, threads, local }; assertMessageStoreMessagesAreEqual( processedMessageStore, newMessageStore, action.type, ); return { messageStoreOperations, messageStore: newMessageStore }; } for (const existingMessageID of messageIDs) { const existingMessageInfo = messageStore.messages[existingMessageID]; if (existingMessageInfo && existingMessageInfo.localID === localID) { return { messageStoreOperations: [], messageStore }; } } const threadState: ThreadMessageInfo = messageStore.threads[threadID] ? { ...messageStore.threads[threadID], messageIDs: [localID, ...messageIDs], } : { messageIDs: [localID], startReached: true, lastNavigatedTo: now, lastPruned: now, }; const newMessageStore = { messages: { ...messageStore.messages, [localID]: payload, }, threads: { ...messageStore.threads, [threadID]: threadState, }, local: messageStore.local, currentAsOf: messageStore.currentAsOf, }; assertMessageStoreMessagesAreEqual( processedMessageStore, newMessageStore, action.type, ); return { messageStoreOperations, messageStore: newMessageStore }; } else if ( action.type === sendTextMessageActionTypes.failed || action.type === sendMultimediaMessageActionTypes.failed ) { const { localID } = action.payload; return { messageStoreOperations: [], messageStore: { messages: messageStore.messages, threads: messageStore.threads, local: { ...messageStore.local, [localID]: { sendFailed: true }, }, currentAsOf: messageStore.currentAsOf, }, }; } else if ( action.type === sendTextMessageActionTypes.success || action.type === sendMultimediaMessageActionTypes.success ) { const { payload } = action; invariant( !threadIsPending(payload.threadID), 'Successful message action should have realized thread id', ); const replaceMessageKey = (messageKey: string) => messageKey === payload.localID ? payload.serverID : messageKey; let newMessages; const messageStoreOperations = []; if (messageStore.messages[payload.serverID]) { // If somehow the serverID got in there already, we'll just update the // serverID message and scrub the localID one newMessages = _omitBy( (messageInfo: RawMessageInfo) => messageInfo.type === messageTypes.TEXT && !messageInfo.id && messageInfo.localID === payload.localID, )(messageStore.messages); messageStoreOperations.push({ type: 'remove', payload: { ids: [payload.localID] }, }); } else if (messageStore.messages[payload.localID]) { // The normal case, the localID message gets replaced by the serverID one newMessages = _mapKeys(replaceMessageKey)(messageStore.messages); messageStoreOperations.push({ type: 'rekey', payload: { from: payload.localID, to: payload.serverID }, }); } else { // Well this is weird, we probably got deauthorized between when the // action was dispatched and when we ran this reducer... return { messageStoreOperations, messageStore }; } const newMessage = { ...newMessages[payload.serverID], id: payload.serverID, localID: payload.localID, time: payload.time, }; newMessages[payload.serverID] = newMessage; messageStoreOperations.push({ type: 'replace', payload: { id: payload.serverID, messageInfo: newMessage }, }); const threadID = payload.threadID; const newMessageIDs = _flow( _uniq, sortMessageIDs(newMessages), )(messageStore.threads[threadID].messageIDs.map(replaceMessageKey)); const currentAsOf = payload.interface === 'socket' ? Math.max(payload.time, messageStore.currentAsOf) : messageStore.currentAsOf; const local = _pickBy( (localInfo: LocalMessageInfo, key: string) => key !== payload.localID, )(messageStore.local); const updatedMessageStore = { messages: newMessages, threads: { ...messageStore.threads, [threadID]: { ...messageStore.threads[threadID], messageIDs: newMessageIDs, }, }, local, currentAsOf, }; const processedMessageStore = processMessageStoreOperations( messageStore, messageStoreOperations, ); assertMessageStoreMessagesAreEqual( processedMessageStore, updatedMessageStore, action.type, ); return { messageStoreOperations, messageStore: updatedMessageStore }; } else if (action.type === saveMessagesActionType) { const truncationStatuses = {}; for (const messageInfo of action.payload.rawMessageInfos) { truncationStatuses[messageInfo.threadID] = messageTruncationStatus.UNCHANGED; } const { messageStoreOperations, messageStore: newMessageStore, } = mergeNewMessages( messageStore, action.payload.rawMessageInfos, truncationStatuses, newThreadInfos, action.type, ); return { messageStoreOperations, messageStore: { messages: newMessageStore.messages, threads: newMessageStore.threads, local: newMessageStore.local, // We avoid bumping currentAsOf because notifs may include a contracted // RawMessageInfo, so we want to make sure we still fetch it currentAsOf: messageStore.currentAsOf, }, }; } else if (action.type === messageStorePruneActionType) { const now = Date.now(); const messageIDsToPrune = []; const newThreads = { ...messageStore.threads }; for (const threadID of action.payload.threadIDs) { let thread = newThreads[threadID]; if (!thread) { continue; } thread = { ...thread, lastPruned: now }; const newMessageIDs = [...thread.messageIDs]; const removed = newMessageIDs.splice(defaultNumberPerThread); if (removed.length > 0) { thread = { ...thread, messageIDs: newMessageIDs, startReached: false, }; } for (const id of removed) { messageIDsToPrune.push(id); } newThreads[threadID] = thread; } const prunedMessageStore = { messages: _omit(messageIDsToPrune)(messageStore.messages), threads: newThreads, local: _omit(messageIDsToPrune)(messageStore.local), currentAsOf: messageStore.currentAsOf, }; const messageStoreOperations = [ { type: 'remove', payload: { ids: messageIDsToPrune }, }, ]; const processedMessageStore = processMessageStoreOperations( messageStore, messageStoreOperations, ); assertMessageStoreMessagesAreEqual( processedMessageStore, prunedMessageStore, action.type, ); return { messageStoreOperations, messageStore: prunedMessageStore }; } else if (action.type === updateMultimediaMessageMediaActionType) { const { messageID: id, currentMediaID, mediaUpdate } = action.payload; const message = messageStore.messages[id]; invariant(message, `message with ID ${id} could not be found`); invariant( message.type === messageTypes.IMAGES || message.type === messageTypes.MULTIMEDIA, `message with ID ${id} is not multimedia`, ); let updatedMessage; let replaced = false; if (message.type === messageTypes.IMAGES) { const media: Image[] = []; for (const singleMedia of message.media) { if (singleMedia.id !== currentMediaID) { media.push(singleMedia); } else { let updatedMedia: Image = { id: mediaUpdate.id ?? singleMedia.id, type: 'photo', uri: mediaUpdate.uri ?? singleMedia.uri, dimensions: mediaUpdate.dimensions ?? singleMedia.dimensions, }; if ( 'localMediaSelection' in singleMedia && !('localMediaSelection' in mediaUpdate) ) { updatedMedia = { ...updatedMedia, localMediaSelection: singleMedia.localMediaSelection, }; } else if (mediaUpdate.localMediaSelection) { updatedMedia = { ...updatedMedia, localMediaSelection: mediaUpdate.localMediaSelection, }; } media.push(updatedMedia); replaced = true; } } updatedMessage = { ...message, media }; } else { const media: Media[] = []; for (const singleMedia of message.media) { if (singleMedia.id !== currentMediaID) { media.push(singleMedia); } else if ( singleMedia.type === 'photo' && mediaUpdate.type === 'photo' ) { media.push({ ...singleMedia, ...mediaUpdate }); replaced = true; } else if ( singleMedia.type === 'video' && mediaUpdate.type === 'video' ) { media.push({ ...singleMedia, ...mediaUpdate }); replaced = true; } } updatedMessage = { ...message, media }; } invariant( replaced, `message ${id} did not contain media with ID ${currentMediaID}`, ); const updatedMessageStore = { ...messageStore, messages: { ...messageStore.messages, [id]: updatedMessage, }, }; const messageStoreOperations = [ { type: 'replace', payload: { id, messageInfo: updatedMessage, }, }, ]; const processedMessageStore = processMessageStoreOperations( messageStore, messageStoreOperations, ); assertMessageStoreMessagesAreEqual( processedMessageStore, updatedMessageStore, action.type, ); return { messageStoreOperations, messageStore: updatedMessageStore }; } else if (action.type === createLocalMessageActionType) { const messageInfo = ensureRealizedThreadIDIsUsedWhenPossible( action.payload, newThreadInfos, ); const { localID, threadID } = messageInfo; const messageIDs = messageStore.threads[messageInfo.threadID]?.messageIDs ?? []; const now = Date.now(); const threadState: ThreadMessageInfo = messageStore.threads[threadID] ? { ...messageStore.threads[threadID], messageIDs: [localID, ...messageIDs], } : { messageIDs: [localID], startReached: true, lastNavigatedTo: now, lastPruned: now, }; const updatedMessageStore = { ...messageStore, messages: { ...messageStore.messages, [localID]: messageInfo, }, threads: { ...messageStore.threads, [threadID]: threadState, }, }; const messageStoreOperations = [ { type: 'replace', payload: { id: localID, messageInfo }, }, ]; const processedMessageStore = processMessageStoreOperations( messageStore, messageStoreOperations, ); assertMessageStoreMessagesAreEqual( updatedMessageStore, processedMessageStore, action.type, ); return { messageStoreOperations, messageStore: updatedMessageStore }; } else if (action.type === rehydrateActionType) { // When starting the app on native, we filter out any local-only multimedia // messages because the relevant context is no longer available const { messages, threads, local } = messageStore; const newMessages = {}; let newThreads = threads, newLocal = local; const messageIDsToBeRemoved = []; for (const id in messages) { const message = messages[id]; if ( (message.type !== messageTypes.IMAGES && message.type !== messageTypes.MULTIMEDIA) || message.id ) { newMessages[id] = message; continue; } messageIDsToBeRemoved.push(id); const { threadID } = message; newThreads = { ...newThreads, [threadID]: { ...newThreads[threadID], messageIDs: newThreads[threadID].messageIDs.filter( curMessageID => curMessageID !== id, ), }, }; newLocal = _pickBy( (localInfo: LocalMessageInfo, key: string) => key !== id, )(newLocal); } if (newThreads === threads) { return { messageStoreOperations: [], messageStore }; } const newMessageStore = { ...messageStore, messages: newMessages, threads: newThreads, local: newLocal, }; const messageStoreOperations: MessageStoreOperation[] = [ { type: 'remove', payload: { ids: messageIDsToBeRemoved }, }, ]; const processedMessageStore = processMessageStoreOperations( messageStore, messageStoreOperations, ); assertMessageStoreMessagesAreEqual( processedMessageStore, newMessageStore, action.type, ); return { messageStoreOperations, messageStore: newMessageStore }; } else if (action.type === processServerRequestsActionType) { const { messageStoreOperations, messageStore: messageStoreAfterReassignment, } = reassignMessagesToRealizedThreads(messageStore, newThreadInfos); const processedMessageStore = processMessageStoreOperations( messageStore, messageStoreOperations, ); assertMessageStoreMessagesAreEqual( processedMessageStore, messageStoreAfterReassignment, action.type, ); return { messageStoreOperations, messageStore: messageStoreAfterReassignment, }; } else if (action.type === setMessageStoreMessages) { return { messageStoreOperations: [], messageStore: { ...messageStore, messages: translateClientDBMessageInfosToRawMessageInfos( action.payload, ), }, }; } return { messageStoreOperations: [], messageStore }; } type MergedUpdatesWithMessages = { +rawMessageInfos: $ReadOnlyArray, +truncationStatuses: MessageTruncationStatuses, }; function mergeUpdatesWithMessageInfos( messageInfos: $ReadOnlyArray, newUpdates: $ReadOnlyArray, truncationStatuses?: MessageTruncationStatuses, ): MergedUpdatesWithMessages { const messageIDs = new Set(messageInfos.map(messageInfo => messageInfo.id)); const mergedMessageInfos = [...messageInfos]; const mergedTruncationStatuses = { ...truncationStatuses }; for (const updateInfo of newUpdates) { if (updateInfo.type !== updateTypes.JOIN_THREAD) { continue; } for (const messageInfo of updateInfo.rawMessageInfos) { if (messageIDs.has(messageInfo.id)) { continue; } mergedMessageInfos.push(messageInfo); messageIDs.add(messageInfo.id); } mergedTruncationStatuses[ updateInfo.threadInfo.id ] = combineTruncationStatuses( updateInfo.truncationStatus, mergedTruncationStatuses[updateInfo.threadInfo.id], ); } return { rawMessageInfos: mergedMessageInfos, truncationStatuses: mergedTruncationStatuses, }; } export { freshMessageStore, reduceMessageStore }; diff --git a/lib/reducers/thread-reducer.js b/lib/reducers/thread-reducer.js index aa4d84d90..ba531417b 100644 --- a/lib/reducers/thread-reducer.js +++ b/lib/reducers/thread-reducer.js @@ -1,604 +1,603 @@ // @flow import _isEqual from 'lodash/fp/isEqual'; import { setThreadUnreadStatusActionTypes, updateActivityActionTypes, } from '../actions/activity-actions'; import { saveMessagesActionType } from '../actions/message-actions'; import { changeThreadSettingsActionTypes, deleteThreadActionTypes, newThreadActionTypes, removeUsersFromThreadActionTypes, changeThreadMemberRolesActionTypes, joinThreadActionTypes, leaveThreadActionTypes, setThreadStoreActionType, } from '../actions/thread-actions'; import { logOutActionTypes, deleteAccountActionTypes, logInActionTypes, registerActionTypes, updateSubscriptionActionTypes, } from '../actions/user-actions'; import type { BaseAction } from '../types/redux-types'; import { type ClientThreadInconsistencyReportCreationRequest, reportTypes, } from '../types/report-types'; import { serverRequestTypes, processServerRequestsActionType, } from '../types/request-types'; import { fullStateSyncActionType, incrementalStateSyncActionType, } from '../types/socket-types'; import type { RawThreadInfo, ThreadStore, ThreadStoreOperation, } from '../types/thread-types'; import { updateTypes, type ClientUpdateInfo, processUpdatesActionType, } from '../types/update-types'; import { actionLogger } from '../utils/action-logger'; import { setNewSessionActionType } from '../utils/action-utils'; import { getConfig } from '../utils/config'; -import { isDev } from '../utils/dev-utils'; import { assertObjectsAreEqual } from '../utils/objects'; import { sanitizeActionSecrets } from '../utils/sanitization'; -const PROCESSED_THREAD_STORE_INVARIANTS_DISABLED = !isDev; +const PROCESSED_THREAD_STORE_INVARIANTS_DISABLED = false; function reduceThreadUpdates( threadInfos: { +[id: string]: RawThreadInfo }, payload: { +updatesResult: { +newUpdates: $ReadOnlyArray, ... }, ... }, ): { +threadStoreOperations: $ReadOnlyArray, +threadInfos: { +[id: string]: RawThreadInfo }, } { const updatedThreadInfos = { ...threadInfos }; let someThreadUpdated = false; const threadOperations: ThreadStoreOperation[] = []; for (const update of payload.updatesResult.newUpdates) { if ( (update.type === updateTypes.UPDATE_THREAD || update.type === updateTypes.JOIN_THREAD) && !_isEqual(threadInfos[update.threadInfo.id])(update.threadInfo) ) { someThreadUpdated = true; updatedThreadInfos[update.threadInfo.id] = update.threadInfo; threadOperations.push({ type: 'replace', payload: { id: update.threadInfo.id, threadInfo: update.threadInfo, }, }); } else if ( update.type === updateTypes.UPDATE_THREAD_READ_STATUS && threadInfos[update.threadID] && threadInfos[update.threadID].currentUser.unread !== update.unread ) { someThreadUpdated = true; const updatedThread = { ...threadInfos[update.threadID], currentUser: { ...threadInfos[update.threadID].currentUser, unread: update.unread, }, }; updatedThreadInfos[update.threadID] = updatedThread; threadOperations.push({ type: 'replace', payload: { id: update.threadID, threadInfo: updatedThread, }, }); } else if ( update.type === updateTypes.DELETE_THREAD && threadInfos[update.threadID] ) { someThreadUpdated = true; delete updatedThreadInfos[update.threadID]; threadOperations.push({ type: 'remove', payload: { ids: [update.threadID], }, }); } else if (update.type === updateTypes.DELETE_ACCOUNT) { for (const threadID in threadInfos) { const threadInfo = threadInfos[threadID]; const newMembers = threadInfo.members.filter( member => member.id !== update.deletedUserID, ); if (newMembers.length < threadInfo.members.length) { someThreadUpdated = true; const updatedThread = { ...threadInfo, members: newMembers, }; updatedThreadInfos[threadID] = updatedThread; threadOperations.push({ type: 'replace', payload: { id: threadID, threadInfo: updatedThread, }, }); } } } } if (!someThreadUpdated) { return { threadStoreOperations: [], threadInfos }; } return { threadStoreOperations: threadOperations, threadInfos: updatedThreadInfos, }; } const emptyArray = []; function findInconsistencies( action: BaseAction, beforeStateCheck: { +[id: string]: RawThreadInfo }, afterStateCheck: { +[id: string]: RawThreadInfo }, ): ClientThreadInconsistencyReportCreationRequest[] { if (_isEqual(beforeStateCheck)(afterStateCheck)) { return emptyArray; } return [ { type: reportTypes.THREAD_INCONSISTENCY, platformDetails: getConfig().platformDetails, beforeAction: beforeStateCheck, action: sanitizeActionSecrets(action), pushResult: afterStateCheck, lastActions: actionLogger.interestingActionSummaries, time: Date.now(), }, ]; } function reduceThreadInfos( state: ThreadStore, action: BaseAction, ): { threadStore: ThreadStore, newThreadInconsistencies: $ReadOnlyArray, threadStoreOperations: $ReadOnlyArray, } { if ( action.type === logInActionTypes.success || action.type === registerActionTypes.success || action.type === fullStateSyncActionType ) { const newThreadInfos = action.payload.threadInfos; const updatedStore = { threadInfos: newThreadInfos, }; const threadStoreOperations = [ { type: 'remove_all', }, ...Object.keys(newThreadInfos).map((id: string) => ({ type: 'replace', payload: { id, threadInfo: newThreadInfos[id] }, })), ]; const processedStore = processThreadStoreOperations( state, threadStoreOperations, ); assertThreadStoreThreadsAreEqual(processedStore, updatedStore, action.type); return { threadStore: updatedStore, newThreadInconsistencies: [], threadStoreOperations, }; } else if ( action.type === logOutActionTypes.success || action.type === deleteAccountActionTypes.success || (action.type === setNewSessionActionType && action.payload.sessionChange.cookieInvalidated) ) { if (Object.keys(state.threadInfos).length === 0) { return { threadStore: state, newThreadInconsistencies: [], threadStoreOperations: [], }; } const updatedStore = { threadInfos: {}, }; const threadStoreOperations = [ { type: 'remove_all', }, ]; const processedStore = processThreadStoreOperations( state, threadStoreOperations, ); assertThreadStoreThreadsAreEqual(processedStore, updatedStore, action.type); return { threadStore: updatedStore, newThreadInconsistencies: [], threadStoreOperations, }; } else if ( action.type === joinThreadActionTypes.success || action.type === leaveThreadActionTypes.success || action.type === deleteThreadActionTypes.success || action.type === changeThreadSettingsActionTypes.success || action.type === removeUsersFromThreadActionTypes.success || action.type === changeThreadMemberRolesActionTypes.success || action.type === incrementalStateSyncActionType || action.type === processUpdatesActionType || action.type === newThreadActionTypes.success ) { if (action.payload.updatesResult.newUpdates.length === 0) { return { threadStore: state, newThreadInconsistencies: [], threadStoreOperations: [], }; } const { threadStoreOperations, threadInfos: newThreadInfos, } = reduceThreadUpdates(state.threadInfos, action.payload); const updatedStore = { threadInfos: newThreadInfos, }; const processedStore = processThreadStoreOperations( state, threadStoreOperations, ); assertThreadStoreThreadsAreEqual(processedStore, updatedStore, action.type); return { threadStore: updatedStore, newThreadInconsistencies: [], threadStoreOperations, }; } else if (action.type === updateSubscriptionActionTypes.success) { const newThreadInfo = { ...state.threadInfos[action.payload.threadID], currentUser: { ...state.threadInfos[action.payload.threadID].currentUser, subscription: action.payload.subscription, }, }; const updatedStore = { threadInfos: { ...state.threadInfos, [action.payload.threadID]: newThreadInfo, }, }; const threadStoreOperations = [ { type: 'replace', payload: { id: action.payload.threadID, threadInfo: newThreadInfo, }, }, ]; const processedStore = processThreadStoreOperations( state, threadStoreOperations, ); assertThreadStoreThreadsAreEqual(processedStore, updatedStore, action.type); return { threadStore: updatedStore, newThreadInconsistencies: [], threadStoreOperations, }; } else if (action.type === saveMessagesActionType) { const threadIDToMostRecentTime = new Map(); for (const messageInfo of action.payload.rawMessageInfos) { const current = threadIDToMostRecentTime.get(messageInfo.threadID); if (!current || current < messageInfo.time) { threadIDToMostRecentTime.set(messageInfo.threadID, messageInfo.time); } } const changedThreadInfos = {}; for (const [threadID, mostRecentTime] of threadIDToMostRecentTime) { const threadInfo = state.threadInfos[threadID]; if ( !threadInfo || threadInfo.currentUser.unread || action.payload.updatesCurrentAsOf > mostRecentTime ) { continue; } changedThreadInfos[threadID] = { ...state.threadInfos[threadID], currentUser: { ...state.threadInfos[threadID].currentUser, unread: true, }, }; } if (Object.keys(changedThreadInfos).length !== 0) { const updatedStore = { threadInfos: { ...state.threadInfos, ...changedThreadInfos, }, }; const threadStoreOperations = Object.keys(changedThreadInfos).map(id => ({ type: 'replace', payload: { id, threadInfo: changedThreadInfos[id], }, })); const processedStore = processThreadStoreOperations( state, threadStoreOperations, ); assertThreadStoreThreadsAreEqual( processedStore, updatedStore, action.type, ); return { threadStore: updatedStore, newThreadInconsistencies: [], threadStoreOperations, }; } } else if (action.type === processServerRequestsActionType) { const checkStateRequest = action.payload.serverRequests.find( candidate => candidate.type === serverRequestTypes.CHECK_STATE, ); if (!checkStateRequest || !checkStateRequest.stateChanges) { return { threadStore: state, newThreadInconsistencies: [], threadStoreOperations: [], }; } const { rawThreadInfos, deleteThreadIDs } = checkStateRequest.stateChanges; if (!rawThreadInfos && !deleteThreadIDs) { return { threadStore: state, newThreadInconsistencies: [], threadStoreOperations: [], }; } const newThreadInfos = { ...state.threadInfos }; const threadStoreOperations: ThreadStoreOperation[] = []; if (rawThreadInfos) { for (const rawThreadInfo of rawThreadInfos) { newThreadInfos[rawThreadInfo.id] = rawThreadInfo; threadStoreOperations.push({ type: 'replace', payload: { id: rawThreadInfo.id, threadInfo: rawThreadInfo, }, }); } } if (deleteThreadIDs) { for (const deleteThreadID of deleteThreadIDs) { delete newThreadInfos[deleteThreadID]; } threadStoreOperations.push({ type: 'remove', payload: { ids: deleteThreadIDs, }, }); } const newThreadInconsistencies = findInconsistencies( action, state.threadInfos, newThreadInfos, ); const updatedStore = { threadInfos: newThreadInfos, }; const processedStore = processThreadStoreOperations( state, threadStoreOperations, ); assertThreadStoreThreadsAreEqual(processedStore, updatedStore, action.type); return { threadStore: updatedStore, newThreadInconsistencies, threadStoreOperations, }; } else if (action.type === updateActivityActionTypes.success) { const updatedThreadInfos = {}; for (const setToUnread of action.payload.result.unfocusedToUnread) { const threadInfo = state.threadInfos[setToUnread]; if (threadInfo && !threadInfo.currentUser.unread) { updatedThreadInfos[setToUnread] = { ...threadInfo, currentUser: { ...threadInfo.currentUser, unread: true, }, }; } } if (Object.keys(updatedThreadInfos).length === 0) { return { threadStore: state, newThreadInconsistencies: [], threadStoreOperations: [], }; } const updatedStore = { threadInfos: { ...state.threadInfos, ...updatedThreadInfos }, }; const threadStoreOperations = Object.keys(updatedThreadInfos).map(id => ({ type: 'replace', payload: { id, threadInfo: updatedThreadInfos[id], }, })); const processedStore = processThreadStoreOperations( state, threadStoreOperations, ); assertThreadStoreThreadsAreEqual(processedStore, updatedStore, action.type); return { threadStore: updatedStore, newThreadInconsistencies: [], threadStoreOperations, }; } else if (action.type === setThreadUnreadStatusActionTypes.started) { const { threadID, unread } = action.payload; const updatedThreadInfo = { ...state.threadInfos[threadID], currentUser: { ...state.threadInfos[threadID].currentUser, unread, }, }; const updatedStore = { threadInfos: { ...state.threadInfos, [threadID]: updatedThreadInfo, }, }; const threadStoreOperations = [ { type: 'replace', payload: { id: threadID, threadInfo: updatedThreadInfo, }, }, ]; const processedStore = processThreadStoreOperations( state, threadStoreOperations, ); assertThreadStoreThreadsAreEqual(processedStore, updatedStore, action.type); return { threadStore: updatedStore, newThreadInconsistencies: [], threadStoreOperations, }; } else if (action.type === setThreadUnreadStatusActionTypes.success) { const { threadID, resetToUnread } = action.payload; const currentUser = state.threadInfos[threadID].currentUser; if (!resetToUnread || currentUser.unread) { return { threadStore: state, newThreadInconsistencies: [], threadStoreOperations: [], }; } const updatedUser = { ...currentUser, unread: true, }; const updatedThread = { ...state.threadInfos[threadID], currentUser: updatedUser, }; const updatedStore = { threadInfos: { ...state.threadInfos, [threadID]: updatedThread, }, }; const threadStoreOperations = [ { type: 'replace', payload: { id: threadID, threadInfo: updatedThread, }, }, ]; const processedStore = processThreadStoreOperations( state, threadStoreOperations, ); assertThreadStoreThreadsAreEqual(processedStore, updatedStore, action.type); return { threadStore: updatedStore, newThreadInconsistencies: [], threadStoreOperations, }; } else if (action.type === setThreadStoreActionType) { return { threadStore: action.payload, newThreadInconsistencies: [], threadStoreOperations: [], }; } return { threadStore: state, newThreadInconsistencies: [], threadStoreOperations: [], }; } function processThreadStoreOperations( threadStore: ThreadStore, threadStoreOperations: $ReadOnlyArray, ): ThreadStore { if (threadStoreOperations.length === 0) { return threadStore; } let processedThreads = { ...threadStore.threadInfos }; for (const operation of threadStoreOperations) { if (operation.type === 'replace') { processedThreads[operation.payload.id] = operation.payload.threadInfo; } else if (operation.type === 'remove') { for (const id of operation.payload.ids) { delete processedThreads[id]; } } else if (operation.type === 'remove_all') { processedThreads = {}; } } return { ...threadStore, threadInfos: processedThreads }; } function assertThreadStoreThreadsAreEqual( processedThreadStore: ThreadStore, expectedThreadStore: ThreadStore, actionType: string, ) { if (PROCESSED_THREAD_STORE_INVARIANTS_DISABLED) { return; } assertObjectsAreEqual( processedThreadStore.threadInfos, expectedThreadStore.threadInfos, `ThreadStore.threadInfos - ${actionType}`, ); } export { reduceThreadInfos, processThreadStoreOperations, assertThreadStoreThreadsAreEqual, }; diff --git a/native/redux/redux-setup.js b/native/redux/redux-setup.js index bd86e6611..d1daf335e 100644 --- a/native/redux/redux-setup.js +++ b/native/redux/redux-setup.js @@ -1,509 +1,522 @@ // @flow +import { isEqual } from 'lodash/lang'; import { AppState as NativeAppState, Platform, Alert } from 'react-native'; import ExitApp from 'react-native-exit-app'; import Orientation from 'react-native-orientation-locker'; import { createStore, applyMiddleware, type Store, compose } from 'redux'; import { persistStore, persistReducer } from 'redux-persist'; import thunk from 'redux-thunk'; import { setDeviceTokenActionTypes } from 'lib/actions/device-actions'; import { logOutActionTypes, deleteAccountActionTypes, logInActionTypes, sqliteOpFailure, } from 'lib/actions/user-actions'; import baseReducer from 'lib/reducers/master-reducer'; import { processThreadStoreOperations, assertThreadStoreThreadsAreEqual, } from 'lib/reducers/thread-reducer'; import { invalidSessionDowngrade, invalidSessionRecovery, } from 'lib/shared/account-utils'; import { defaultEnabledApps } from 'lib/types/enabled-apps'; import { defaultCalendarFilters } from 'lib/types/filter-types'; import type { Dispatch, BaseAction } from 'lib/types/redux-types'; import type { SetSessionPayload } from 'lib/types/session-types'; import { defaultConnectionInfo, incrementalStateSyncActionType, } from 'lib/types/socket-types'; import type { ThreadStoreOperation } from 'lib/types/thread-types'; import { updateTypes } from 'lib/types/update-types'; import { reduxLoggerMiddleware } from 'lib/utils/action-logger'; import { setNewSessionActionType } from 'lib/utils/action-utils'; -import { convertMessageStoreOperationsToClientDBOperations } from 'lib/utils/message-ops-utils'; +import { + convertMessageStoreOperationsToClientDBOperations, + translateClientDBMessageInfosToRawMessageInfos, +} from 'lib/utils/message-ops-utils'; import { convertThreadStoreOperationsToClientDBOperations } from 'lib/utils/thread-ops-utils'; import { defaultNavInfo } from '../navigation/default-state'; import { getGlobalNavContext } from '../navigation/icky-global'; import { activeMessageListSelector } from '../navigation/nav-selectors'; import { defaultNotifPermissionAlertInfo } from '../push/alerts'; import { reduceThreadIDsToNotifIDs } from '../push/reducer'; import reactotron from '../reactotron'; import { defaultDeviceCameraInfo } from '../types/camera'; import { defaultConnectivityInfo } from '../types/connectivity'; import { defaultGlobalThemeInfo } from '../types/themes'; import { defaultURLPrefix, natNodeServer, setCustomServer, getDevServerHostname, } from '../utils/url-utils'; import { resetUserStateActionType, recordNotifPermissionAlertActionType, recordAndroidNotificationActionType, clearAndroidNotificationsActionType, rescindAndroidNotificationActionType, updateDimensionsActiveType, updateConnectivityActiveType, updateThemeInfoActionType, updateDeviceCameraInfoActionType, updateDeviceOrientationActionType, updateThreadLastNavigatedActionType, backgroundActionTypes, setReduxStateActionType, type Action, } from './action-types'; import { remoteReduxDevServerConfig } from './dev-tools'; import { defaultDimensionsInfo } from './dimensions-updater.react'; import { persistConfig, setPersistor } from './persist'; import type { AppState } from './state-types'; const defaultState = ({ navInfo: defaultNavInfo, currentUserInfo: null, entryStore: { entryInfos: {}, daysToEntries: {}, lastUserInteractionCalendar: 0, }, threadStore: { threadInfos: {}, }, userStore: { userInfos: {}, inconsistencyReports: [], }, messageStore: { messages: {}, threads: {}, local: {}, currentAsOf: 0, }, updatesCurrentAsOf: 0, loadingStatuses: {}, calendarFilters: defaultCalendarFilters, cookie: null, deviceToken: null, dataLoaded: false, urlPrefix: defaultURLPrefix, customServer: natNodeServer, threadIDsToNotifIDs: {}, notifPermissionAlertInfo: defaultNotifPermissionAlertInfo, connection: defaultConnectionInfo(Platform.OS), watchedThreadIDs: [], lifecycleState: 'active', enabledApps: defaultEnabledApps, reportStore: { enabledReports: { crashReports: __DEV__, inconsistencyReports: __DEV__, mediaReports: __DEV__, }, queuedReports: [], }, nextLocalID: 0, _persist: null, dimensions: defaultDimensionsInfo, connectivity: defaultConnectivityInfo, globalThemeInfo: defaultGlobalThemeInfo, deviceCameraInfo: defaultDeviceCameraInfo, deviceOrientation: Orientation.getInitialOrientation(), frozen: false, }: AppState); function reducer(state: AppState = defaultState, action: Action) { if (action.type === setReduxStateActionType) { return action.payload.state; } if ( (action.type === setNewSessionActionType && invalidSessionDowngrade( state, action.payload.sessionChange.currentUserInfo, action.payload.preRequestUserState, )) || (action.type === logOutActionTypes.success && invalidSessionDowngrade( state, action.payload.currentUserInfo, action.payload.preRequestUserState, )) || (action.type === deleteAccountActionTypes.success && invalidSessionDowngrade( state, action.payload.currentUserInfo, action.payload.preRequestUserState, )) ) { return state; } if ( (action.type === setNewSessionActionType && action.payload.sessionChange.currentUserInfo && invalidSessionRecovery( state, action.payload.sessionChange.currentUserInfo, action.payload.source, )) || (action.type === logInActionTypes.success && invalidSessionRecovery( state, action.payload.currentUserInfo, action.payload.source, )) ) { return state; } const threadIDsToNotifIDs = reduceThreadIDsToNotifIDs( state.threadIDsToNotifIDs, action, ); state = { ...state, threadIDsToNotifIDs }; if ( action.type === recordAndroidNotificationActionType || action.type === clearAndroidNotificationsActionType || action.type === rescindAndroidNotificationActionType ) { return state; } if (action.type === setCustomServer) { return { ...state, customServer: action.payload, }; } else if (action.type === recordNotifPermissionAlertActionType) { return { ...state, notifPermissionAlertInfo: { totalAlerts: state.notifPermissionAlertInfo.totalAlerts + 1, lastAlertTime: action.payload.time, }, }; } else if (action.type === resetUserStateActionType) { const cookie = state.cookie && state.cookie.startsWith('anonymous=') ? state.cookie : null; const currentUserInfo = state.currentUserInfo && state.currentUserInfo.anonymous ? state.currentUserInfo : null; return { ...state, currentUserInfo, cookie, }; } else if (action.type === updateDimensionsActiveType) { return { ...state, dimensions: { ...state.dimensions, ...action.payload, }, }; } else if (action.type === updateConnectivityActiveType) { return { ...state, connectivity: action.payload, }; } else if (action.type === updateThemeInfoActionType) { return { ...state, globalThemeInfo: { ...state.globalThemeInfo, ...action.payload, }, }; } else if (action.type === updateDeviceCameraInfoActionType) { return { ...state, deviceCameraInfo: { ...state.deviceCameraInfo, ...action.payload, }, }; } else if (action.type === updateDeviceOrientationActionType) { return { ...state, deviceOrientation: action.payload, }; } else if (action.type === setDeviceTokenActionTypes.success) { return { ...state, deviceToken: action.payload, }; } else if (action.type === updateThreadLastNavigatedActionType) { const { threadID, time } = action.payload; if (state.messageStore.threads[threadID]) { state = { ...state, messageStore: { ...state.messageStore, threads: { ...state.messageStore.threads, [threadID]: { ...state.messageStore.threads[threadID], lastNavigatedTo: time, }, }, }, }; } return state; } if (action.type === setNewSessionActionType) { sessionInvalidationAlert(action.payload); state = { ...state, cookie: action.payload.sessionChange.cookie, }; } else if (action.type === incrementalStateSyncActionType) { let wipeDeviceToken = false; for (const update of action.payload.updatesResult.newUpdates) { if ( update.type === updateTypes.BAD_DEVICE_TOKEN && update.deviceToken === state.deviceToken ) { wipeDeviceToken = true; break; } } if (wipeDeviceToken) { state = { ...state, deviceToken: null, }; } } const baseReducerResult = baseReducer(state, (action: BaseAction)); state = baseReducerResult.state; const { storeOperations } = baseReducerResult; const { threadStoreOperations, messageStoreOperations } = storeOperations; const fixUnreadActiveThreadResult = fixUnreadActiveThread(state, action); state = fixUnreadActiveThreadResult.state; const threadStoreOperationsWithUnreadFix = [ ...threadStoreOperations, ...fixUnreadActiveThreadResult.threadStoreOperations, ]; const convertedThreadStoreOperations = convertThreadStoreOperationsToClientDBOperations( threadStoreOperationsWithUnreadFix, ); const convertedMessageStoreOperations = convertMessageStoreOperationsToClientDBOperations( messageStoreOperations, ); + + if (convertedMessageStoreOperations.length > 0) { + global.CommCoreModule.processMessageStoreOperationsSync( + convertedMessageStoreOperations, + ); + } + + const messages = global.CommCoreModule.getAllMessagesSync(); + const rawMsgsFromSQLite = translateClientDBMessageInfosToRawMessageInfos( + messages, + ); + + if (!isEqual(rawMsgsFromSQLite, state.messageStore.messages)) { + Alert.alert(`${action.type}: NOT EQUAL`); + } + (async () => { try { const promises = []; if (convertedThreadStoreOperations.length > 0) { promises.push( global.CommCoreModule.processThreadStoreOperations( convertedThreadStoreOperations, ), ); } - if (convertedMessageStoreOperations.length > 0) { - promises.push( - global.CommCoreModule.processMessageStoreOperations( - convertedMessageStoreOperations, - ), - ); - } await Promise.all(promises); } catch { dispatch({ type: setNewSessionActionType, payload: { sessionChange: { cookie: null, cookieInvalidated: false, currentUserInfo: state.currentUserInfo, }, preRequestUserState: { currentUserInfo: state.currentUserInfo, sessionID: undefined, cookie: state.cookie, }, error: null, source: sqliteOpFailure, }, }); await persistor.flush(); ExitApp.exitApp(); } })(); return state; } function sessionInvalidationAlert(payload: SetSessionPayload) { if ( !payload.sessionChange.cookieInvalidated || !payload.preRequestUserState || !payload.preRequestUserState.currentUserInfo || payload.preRequestUserState.currentUserInfo.anonymous ) { return; } if (payload.error === 'client_version_unsupported') { const app = Platform.select({ ios: 'App Store', android: 'Play Store', }); Alert.alert( 'App out of date', "Your app version is pretty old, and the server doesn't know how to " + `speak to it anymore. Please use the ${app} app to update!`, [{ text: 'OK' }], { cancelable: true }, ); } else { Alert.alert( 'Session invalidated', "We're sorry, but your session was invalidated by the server. " + 'Please log in again.', [{ text: 'OK' }], { cancelable: true }, ); } } // Makes sure a currently focused thread is never unread. Note that we consider // a backgrounded NativeAppState to actually be active if it last changed to // inactive more than 10 seconds ago. This is because there is a delay when // NativeAppState is updating in response to a foreground, and actions don't get // processed more than 10 seconds after a backgrounding anyways. However we // don't consider this for action types that can be expected to happen while the // app is backgrounded. type FixUnreadActiveThreadResult = { +state: AppState, +threadStoreOperations: $ReadOnlyArray, }; function fixUnreadActiveThread( state: AppState, action: *, ): FixUnreadActiveThreadResult { const navContext = getGlobalNavContext(); const activeThread = activeMessageListSelector(navContext); if ( !activeThread || !state.threadStore.threadInfos[activeThread]?.currentUser.unread || (NativeAppState.currentState !== 'active' && (appLastBecameInactive + 10000 >= Date.now() || backgroundActionTypes.has(action.type))) ) { return { state, threadStoreOperations: [] }; } const updatedActiveThreadInfo = { ...state.threadStore.threadInfos[activeThread], currentUser: { ...state.threadStore.threadInfos[activeThread].currentUser, unread: false, }, }; const updatedState = { ...state, threadStore: { ...state.threadStore, threadInfos: { ...state.threadStore.threadInfos, [activeThread]: updatedActiveThreadInfo, }, }, }; const threadStoreOperations = [ { type: 'replace', payload: { id: activeThread, threadInfo: updatedActiveThreadInfo, }, }, ]; const processedStore = processThreadStoreOperations( state.threadStore, threadStoreOperations, ); assertThreadStoreThreadsAreEqual( processedStore, updatedState.threadStore, action.type, ); return { state: updatedState, threadStoreOperations }; } let appLastBecameInactive = 0; function appBecameInactive() { appLastBecameInactive = Date.now(); } const middlewares = [thunk, reduxLoggerMiddleware]; if (__DEV__) { const createDebugger = require('redux-flipper').default; middlewares.push(createDebugger()); } const middleware = applyMiddleware(...middlewares); let composeFunc = compose; if (__DEV__ && global.HermesInternal) { const { composeWithDevTools } = require('remote-redux-devtools/src'); composeFunc = composeWithDevTools({ name: 'Redux', hostname: getDevServerHostname(), ...remoteReduxDevServerConfig, }); } else if (global.__REDUX_DEVTOOLS_EXTENSION_COMPOSE__) { composeFunc = global.__REDUX_DEVTOOLS_EXTENSION_COMPOSE__({ name: 'Redux', }); } let enhancers; if (reactotron) { enhancers = composeFunc(middleware, reactotron.createEnhancer()); } else { enhancers = composeFunc(middleware); } const store: Store = createStore( persistReducer(persistConfig, reducer), defaultState, enhancers, ); const persistor = persistStore(store); setPersistor(persistor); const unsafeDispatch: any = store.dispatch; const dispatch: Dispatch = unsafeDispatch; export { store, dispatch, appBecameInactive };