Page MenuHomePhabricator
Differential D5486

[services] Tunnelbroker - Move ffi from `main.rs` to the `cxx_bridge.rs` file
ClosedPublic
Actions

Authored by • max on Oct 26 2022, 9:36 AM.
Tags
None
Referenced Files
Unknown Object (File)
Wed, Apr 9, 4:10 AM
Unknown Object (File)
Wed, Apr 9, 4:10 AM
Unknown Object (File)
Wed, Apr 9, 4:10 AM
Unknown Object (File)
Wed, Apr 9, 4:10 AM
Unknown Object (File)
Wed, Apr 9, 4:10 AM
Unknown Object (File)
Mar 17 2025, 3:10 AM
Unknown Object (File)
Mar 17 2025, 3:10 AM
Unknown Object (File)
Mar 17 2025, 3:10 AM
View All Files
Subscribers
• abosh
ashoat
atul
tomek

Details

Reviewers
• jon
varun
marcin
tomek
Commits
rCOMM89f260e671ae: [services] Tunnelbroker - Move ffi from `main.rs` to the `cxx_bridge.rs` file
Summary

This diff moves the FFI functions from CXX from main.rs to the cxx_bridge.rs file.
The reason for moving the FFI away from the main.rs to its own scope is that the ffi::* functions should be called from the gRPC server handlers (for example in D5487).

Linear task: ENG-2094

Test Plan

The service was successfully built with the CI process or can be built manually using the cargo build command in services/tunnelbroker directory.

Diff Detail

Repository
rCOMM Comm
Branch
D5486
Lint
No Lint Coverage
Unit
No Test Coverage

Event Timeline

• max created this revision.Oct 26 2022, 9:36 AM
• max held this revision as a draft.
Herald added a reviewer: • jon. · View Herald TranscriptOct 26 2022, 9:36 AM
Herald added subscribers: • abosh, atul, tomek, ashoat. · View Herald Transcript
• max added a parent revision: D5477: [services] Tunnelbroker - Adding Tonic gRPC server with empty handlers.Oct 26 2022, 9:36 AM
• max added a child revision: D5487: [services] Tunnelbroker - Adding of gRPC `SessionSignature` implementation.
Harbormaster completed remote builds in B13050: Diff 17920.Oct 26 2022, 9:47 AM
• max published this revision for review.Oct 26 2022, 10:02 AM
• max edited the summary of this revision. (Show Details)
• max edited the test plan for this revision. (Show Details)
• max added reviewers: varun, marcin.
• jon accepted this revision.Oct 26 2022, 1:45 PM

seems logical to me

This revision is now accepted and ready to land.Oct 26 2022, 1:45 PM
• max added a reviewer: tomek.Oct 27 2022, 2:26 AM
This revision now requires review to proceed.Oct 27 2022, 2:26 AM
tomek accepted this revision.Oct 27 2022, 5:23 AM

Ideally, we should update an existing diff instead of introducing a new one, but for now let's keep it.

This revision is now accepted and ready to land.Oct 27 2022, 5:23 AM
• max updated this revision to Diff 18156.Nov 7 2022, 10:10 AM

Rebasing on the master changes.

• max removed a parent revision: D5477: [services] Tunnelbroker - Adding Tonic gRPC server with empty handlers.Nov 7 2022, 10:18 AM
• max removed a child revision: D5487: [services] Tunnelbroker - Adding of gRPC `SessionSignature` implementation.
Harbormaster completed remote builds in B13230: Diff 18156.Nov 7 2022, 10:25 AM
Closed by commit rCOMM89f260e671ae: [services] Tunnelbroker - Move ffi from `main.rs` to the `cxx_bridge.rs` file (authored by • max). · Explain WhyNov 7 2022, 10:31 AM
This revision was automatically updated to reflect the committed changes.
• max added a commit: rCOMM89f260e671ae: [services] Tunnelbroker - Move ffi from `main.rs` to the `cxx_bridge.rs` file.

Revision Contents
Changeset List

PathSize
services/
tunnelbroker/
2 lines
src/
8 lines
11 lines

Diff 18156

View Options

services/tunnelbroker/build.rs

Show First 20 Lines • Show All 130 Lines • ▼ Show 20 Lines) {
return ( return (
threadIsPending(threadID) || threadIsPending(threadID) ||
(threadInfo && (threadInfo &&
threadHasPermission(threadInfo, threadPermissions.VISIBLE) && threadHasPermission(threadInfo, threadPermissions.VISIBLE) &&
(threadInChatList(threadInfo) || watchedIDs.includes(threadID))) (threadInChatList(threadInfo) || watchedIDs.includes(threadID)))
); );
} }
// eslint-disable-next-line no-unused-vars
function assertMessageStoreThreadsAreEqual( function assertMessageStoreThreadsAreEqual(
processedMessageStore: MessageStore, processedMessageStore: MessageStore,
expectedMessageStore: MessageStore, expectedMessageStore: MessageStore,
location: string, location: string,
) { ) {
if (PROCESSED_MSG_STORE_INVARIANTS_DISABLED) { if (PROCESSED_MSG_STORE_INVARIANTS_DISABLED) {
return; return;
} }
▲ Show 20 Lines • Show All 100 Lines • ▼ Show 20 Lines const updateMsgOperation: ReplaceMessageOperation = {
payload: { id: storeMessageID, messageInfo: messages[storeMessageID] }, payload: { id: storeMessageID, messageInfo: messages[storeMessageID] },
}; };
messageStoreOperations.push(updateMsgOperation); messageStoreOperations.push(updateMsgOperation);
} }
const threads = {}; const threads = {};
const reassignedThreadIDs = []; const reassignedThreadIDs = [];
const updatedThreads = {};
const threadsToRemove = [];
tomekUnsubmitted
Not Done
Inline Actions

I guess we can make this a Set

tomek: I guess we can make this a `Set`
kamilAuthorUnsubmitted
Done
Inline Actions

What are the advantages of using Set here?

In the next lines, I iterate through message store threads only once and if the condition is met I add value to the array, which means each value will occur only once

kamil: What are the advantages of using `Set` here? In the next lines, I iterate through message…
tomekUnsubmitted
Not Done
Inline Actions

Sounds ok. Overall, sets are beneficial if we want to deduplicate or when we check if the set contains something - none of these are the case here, so an array is fine.

tomek: Sounds ok. Overall, sets are beneficial if we want to deduplicate or when we check if the set…
for (const threadID in messageStore.threads) { for (const threadID in messageStore.threads) {
const threadMessageInfo = messageStore.threads[threadID]; const threadMessageInfo = messageStore.threads[threadID];
const newThreadID = pendingToRealizedThreadIDs.get(threadID); const newThreadID = pendingToRealizedThreadIDs.get(threadID);
if (!newThreadID) { if (!newThreadID) {
threads[threadID] = threadMessageInfo; threads[threadID] = threadMessageInfo;
continue; continue;
} }
const realizedThread = messageStore.threads[newThreadID]; const realizedThread = messageStore.threads[newThreadID];
if (!realizedThread) { if (!realizedThread) {
reassignedThreadIDs.push(newThreadID); reassignedThreadIDs.push(newThreadID);
threads[newThreadID] = threadMessageInfo; threads[newThreadID] = threadMessageInfo;
updatedThreads[newThreadID] = threadMessageInfo;
threadsToRemove.push(threadID);
continue; continue;
} }
threads[newThreadID] = mergeThreadMessageInfos( threads[newThreadID] = mergeThreadMessageInfos(
threadMessageInfo, threadMessageInfo,
realizedThread, realizedThread,
messages, messages,
); );
updatedThreads[newThreadID] = threads[newThreadID];
} }
messageStoreOperations.push({
type: 'remove_threads',
payload: {
ids: threadsToRemove,
},
});
messageStoreOperations.push({
type: 'replace_threads',
payload: {
threads: updatedThreads,
},
});
tomekUnsubmitted
Not Done
Inline Actions

We should add these only if they are meaningful. It seems like threadsToRemove can be empty and we don't need to create an op for it.

tomek: We should add these only if they are meaningful. It seems like `threadsToRemove` can be empty…
kamilAuthorUnsubmitted
Done
Inline Actions

right, good catch

kamil: right, good catch
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
...messageStore, ...messageStore,
threads, threads,
messages, messages,
}, },
reassignedThreadIDs, reassignedThreadIDs,
}; };
} }
type MergeNewMessagesResult = { type MergeNewMessagesResult = {
+messageStoreOperations: $ReadOnlyArray<MessageStoreOperation>, +messageStoreOperations: $ReadOnlyArray<MessageStoreOperation>,
+messageStore: MessageStore, +messageStore: MessageStore,
}; };
// oldMessageStore is from the old state // oldMessageStore is from the old state
// newMessageInfos, truncationStatus come from server // newMessageInfos, truncationStatus come from server
function mergeNewMessages( function mergeNewMessages(
oldMessageStore: MessageStore, oldMessageStore: MessageStore,
newMessageInfos: $ReadOnlyArray<RawMessageInfo>, newMessageInfos: $ReadOnlyArray<RawMessageInfo>,
truncationStatus: { [threadID: string]: MessageTruncationStatus }, truncationStatus: { [threadID: string]: MessageTruncationStatus },
threadInfos: { +[threadID: string]: RawThreadInfo }, threadInfos: { +[threadID: string]: RawThreadInfo },
actionType: string,
): MergeNewMessagesResult { ): MergeNewMessagesResult {
const { const {
messageStoreOperations: updateWithLatestThreadInfosOps, messageStoreOperations: updateWithLatestThreadInfosOps,
messageStore: messageStoreUpdatedWithLatestThreadInfos, messageStore: messageStoreUpdatedWithLatestThreadInfos,
reassignedThreadIDs, reassignedThreadIDs,
} = updateMessageStoreWithLatestThreadInfos(oldMessageStore, threadInfos); } = updateMessageStoreWithLatestThreadInfos(oldMessageStore, threadInfos);
const messageStoreAfterUpdateOps = processMessageStoreOperations( const messageStoreAfterUpdateOps = processMessageStoreOperations(
oldMessageStore, oldMessageStore,
updateWithLatestThreadInfosOps, updateWithLatestThreadInfosOps,
); );
assertMessageStoreThreadsAreEqual(
messageStoreAfterUpdateOps,
messageStoreUpdatedWithLatestThreadInfos,
`${actionType} | reassignment and filtering`,
);
const updatedMessageStore = { const updatedMessageStore = {
...messageStoreUpdatedWithLatestThreadInfos, ...messageStoreUpdatedWithLatestThreadInfos,
messages: messageStoreAfterUpdateOps.messages, messages: messageStoreAfterUpdateOps.messages,
}; };
const localIDsToServerIDs: Map<string, string> = new Map(); const localIDsToServerIDs: Map<string, string> = new Map();
const watchedThreadIDs = [ const watchedThreadIDs = [
▲ Show 20 Lines • Show All 121 Lines • ▼ Show 20 Lines): MergeNewMessagesResult {
const newMessageOps: MessageStoreOperation[] = []; const newMessageOps: MessageStoreOperation[] = [];
const threadsToMessageIDs = mapThreadsToMessageIDsFromOrderedMessageInfos( const threadsToMessageIDs = mapThreadsToMessageIDsFromOrderedMessageInfos(
orderedNewMessageInfos, orderedNewMessageInfos,
); );
const oldMessageInfosToCombine = []; const oldMessageInfosToCombine = [];
const threadsThatNeedMessageIDsResorted = []; const threadsThatNeedMessageIDsResorted = [];
const local = {}; const local = {};
const updatedThreads = {};
const threads = _flow( const threads = _flow(
_mapValuesWithKeys((messageIDs: string[], threadID: string) => { _mapValuesWithKeys((messageIDs: string[], threadID: string) => {
const oldThread = updatedMessageStore.threads[threadID]; const oldThread = updatedMessageStore.threads[threadID];
const truncate = truncationStatus[threadID]; const truncate = truncationStatus[threadID];
if (!oldThread) { if (!oldThread) {
return { updatedThreads[threadID] = {
...newThread(), ...newThread(),
messageIDs, messageIDs,
startReached: truncate === messageTruncationStatus.EXHAUSTIVE, startReached: truncate === messageTruncationStatus.EXHAUSTIVE,
}; };
return updatedThreads[threadID];
} }
let oldMessageIDsUnchanged = true; let oldMessageIDsUnchanged = true;
const oldMessageIDs = oldThread.messageIDs.map(oldID => { const oldMessageIDs = oldThread.messageIDs.map(oldID => {
const newID = localIDsToServerIDs.get(oldID); const newID = localIDsToServerIDs.get(oldID);
if (newID !== null && newID !== undefined) { if (newID !== null && newID !== undefined) {
oldMessageIDsUnchanged = false; oldMessageIDsUnchanged = false;
return newID; return newID;
} }
return oldID; return oldID;
}); });
if (truncate === messageTruncationStatus.TRUNCATED) { if (truncate === messageTruncationStatus.TRUNCATED) {
// If the result set in the payload isn't contiguous with what we have // 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 // 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 // it with the result set. We do this to achieve our two goals for the
// messageStore: currentness and contiguousness. // messageStore: currentness and contiguousness.
newMessageOps.push({ newMessageOps.push({
type: 'remove_messages_for_threads', type: 'remove_messages_for_threads',
payload: { threadIDs: [threadID] }, payload: { threadIDs: [threadID] },
}); });
return { updatedThreads[threadID] = {
messageIDs, messageIDs,
startReached: false, startReached: false,
lastNavigatedTo: oldThread.lastNavigatedTo, lastNavigatedTo: oldThread.lastNavigatedTo,
lastPruned: oldThread.lastPruned, lastPruned: oldThread.lastPruned,
}; };
return updatedThreads[threadID];
} }
const oldNotInNew = _difference(oldMessageIDs)(messageIDs); const oldNotInNew = _difference(oldMessageIDs)(messageIDs);
for (const id of oldNotInNew) { for (const id of oldNotInNew) {
const oldMessageInfo = updatedMessageStore.messages[id]; const oldMessageInfo = updatedMessageStore.messages[id];
invariant(oldMessageInfo, `could not find ${id} in messageStore`); invariant(oldMessageInfo, `could not find ${id} in messageStore`);
oldMessageInfosToCombine.push(oldMessageInfo); oldMessageInfosToCombine.push(oldMessageInfo);
const localInfo = updatedMessageStore.local[id]; const localInfo = updatedMessageStore.local[id];
if (localInfo) { if (localInfo) {
local[id] = localInfo; local[id] = localInfo;
} }
} }
const startReached = const startReached =
oldThread.startReached || oldThread.startReached ||
truncate === messageTruncationStatus.EXHAUSTIVE; truncate === messageTruncationStatus.EXHAUSTIVE;
if (_difference(messageIDs)(oldMessageIDs).length === 0) { if (_difference(messageIDs)(oldMessageIDs).length === 0) {
if (startReached === oldThread.startReached && oldMessageIDsUnchanged) { if (startReached === oldThread.startReached && oldMessageIDsUnchanged) {
return oldThread; return oldThread;
} }
return { updatedThreads[threadID] = {
messageIDs: oldMessageIDs, messageIDs: oldMessageIDs,
startReached, startReached,
lastNavigatedTo: oldThread.lastNavigatedTo, lastNavigatedTo: oldThread.lastNavigatedTo,
lastPruned: oldThread.lastPruned, lastPruned: oldThread.lastPruned,
}; };
return updatedThreads[threadID];
} }
const mergedMessageIDs = [...messageIDs, ...oldNotInNew]; const mergedMessageIDs = [...messageIDs, ...oldNotInNew];
threadsThatNeedMessageIDsResorted.push(threadID); threadsThatNeedMessageIDsResorted.push(threadID);
//NOTE: this will be updated after sorting
tomekUnsubmitted
Not Done
Inline Actions

This comment sounds confusing to me. Is it something you will do after "sorting" or "sorting" is a part of this code and the value is somehow updated after it? Please make this less confusing.

tomek: This comment sounds confusing to me. Is it something you will do after "sorting" or "sorting"…
kamilAuthorUnsubmitted
Done
Inline Actions

this comment was a note I made while developing and forgot to remove, I will simply get rid of this comment

kamil: this comment was a note I made while developing and forgot to remove, I will simply get rid of…
return { return {
messageIDs: mergedMessageIDs, messageIDs: mergedMessageIDs,
startReached, startReached,
lastNavigatedTo: oldThread.lastNavigatedTo, lastNavigatedTo: oldThread.lastNavigatedTo,
lastPruned: oldThread.lastPruned, lastPruned: oldThread.lastPruned,
}; };
}), }),
_pickBy(thread => !!thread), _pickBy(thread => !!thread),
)(threadsToMessageIDs); )(threadsToMessageIDs);
for (const threadID in updatedMessageStore.threads) { for (const threadID in updatedMessageStore.threads) {
if (threads[threadID]) { if (threads[threadID]) {
continue; continue;
} }
let thread = updatedMessageStore.threads[threadID]; let thread = updatedMessageStore.threads[threadID];
const truncate = truncationStatus[threadID]; const truncate = truncationStatus[threadID];
if (truncate === messageTruncationStatus.EXHAUSTIVE) { if (truncate === messageTruncationStatus.EXHAUSTIVE) {
thread = { thread = {
...thread, ...thread,
startReached: true, startReached: true,
}; };
} }
threads[threadID] = thread; threads[threadID] = thread;
updatedThreads[threadID] = thread;
for (const id of thread.messageIDs) { for (const id of thread.messageIDs) {
const messageInfo = updatedMessageStore.messages[id]; const messageInfo = updatedMessageStore.messages[id];
if (messageInfo) { if (messageInfo) {
oldMessageInfosToCombine.push(messageInfo); oldMessageInfosToCombine.push(messageInfo);
} }
const localInfo = updatedMessageStore.local[id]; const localInfo = updatedMessageStore.local[id];
if (localInfo) { if (localInfo) {
local[id] = localInfo; local[id] = localInfo;
Show All 20 Lines newMessageOps.push({
payload: { ids: [...localIDsToServerIDs.keys()] }, payload: { ids: [...localIDsToServerIDs.keys()] },
}); });
} }
for (const threadID of threadsThatNeedMessageIDsResorted) { for (const threadID of threadsThatNeedMessageIDsResorted) {
threads[threadID].messageIDs = sortMessageIDs(messages)( threads[threadID].messageIDs = sortMessageIDs(messages)(
threads[threadID].messageIDs, threads[threadID].messageIDs,
); );
updatedThreads[threadID] = threads[threadID];
} }
const currentAsOf = Math.max( const currentAsOf = Math.max(
orderedNewMessageInfos.length > 0 ? orderedNewMessageInfos[0].time : 0, orderedNewMessageInfos.length > 0 ? orderedNewMessageInfos[0].time : 0,
updatedMessageStore.currentAsOf, updatedMessageStore.currentAsOf,
); );
newMessageOps.push({
type: 'replace_threads',
payload: {
threads: updatedThreads,
},
});
const processedMessageStore = processMessageStoreOperations( const processedMessageStore = processMessageStoreOperations(
updatedMessageStore, updatedMessageStore,
newMessageOps, newMessageOps,
); );
const messageStore = {
messages: processedMessageStore.messages,
threads,
tomekUnsubmitted
Not Done
Inline Actions
messages: processedMessageStore.messages,
- threads,
+ threads: processedMessageStore.threads,
local,

Should we also take threads prop of processedMessageStore?

tomek: Should we also take `threads` prop of `processedMessageStore`?
kamilAuthorUnsubmitted
Done
Inline Actions

Lets assume that:
A - old state
B - state modified in reducer, the current logic
C - state modified by ops

previously it worked like this:

  1. We have an old state (l A)
  2. We modify the state (A -> B)
  3. We return the modified state (B)

this diff this changes to:

  1. We have an old state (A)
  2. We modify the state and generate ops (A ->B)
  3. We take the old state, apply ops, and check if this matches the modified state (A -> C, and then checking B === C)
  4. We return modified state (B)

In future, if devs will not encounter any issues:

  1. We have old state (A)
  2. We will remove modifying the state and only generate ops (A not changed)
  3. Apply ops to old state (A -> C)
  4. Return state modified by ops (C)
kamil: Lets assume that: A - old state B - state modified in reducer, the current logic C - state…
tomekUnsubmitted
Not Done
Inline Actions

Thanks for the explanation - it makes sense!

tomek: Thanks for the explanation - it makes sense!
local,
currentAsOf,
};
assertMessageStoreThreadsAreEqual(
processedMessageStore,
messageStore,
`${actionType} | processed`,
);
return { return {
messageStoreOperations: [ messageStoreOperations: [
...updateWithLatestThreadInfosOps, ...updateWithLatestThreadInfosOps,
...newMessageOps, ...newMessageOps,
], ],
messageStore: { messageStore,
messages: processedMessageStore.messages,
threads,
local,
currentAsOf,
},
}; };
} }
type UpdateMessageStoreWithLatestThreadInfosResult = { type UpdateMessageStoreWithLatestThreadInfosResult = {
+messageStoreOperations: MessageStoreOperation[], +messageStoreOperations: MessageStoreOperation[],
+messageStore: MessageStore, +messageStore: MessageStore,
+reassignedThreadIDs: string[], +reassignedThreadIDs: string[],
}; };
Show All 24 Lines if (watchedThreadInfos[threadID]) {
continue; continue;
} }
threadsToRemoveMessagesFrom.push(threadID); threadsToRemoveMessagesFrom.push(threadID);
for (const id of reassignedMessageStore.threads[threadID].messageIDs) { for (const id of reassignedMessageStore.threads[threadID].messageIDs) {
messageIDsToRemove.push(id); messageIDsToRemove.push(id);
} }
} }
const updatedThreads = {};
for (const threadID in threadInfos) { for (const threadID in threadInfos) {
const threadInfo = threadInfos[threadID]; const threadInfo = threadInfos[threadID];
if ( if (
isThreadWatched(threadID, threadInfo, watchedIDs) && isThreadWatched(threadID, threadInfo, watchedIDs) &&
!filteredThreads[threadID] !filteredThreads[threadID]
) { ) {
filteredThreads[threadID] = newThread(); filteredThreads[threadID] = newThread();
updatedThreads[threadID] = filteredThreads[threadID];
} }
} }
messageStoreOperations.push({ messageStoreOperations.push({
type: 'remove_threads',
payload: { ids: threadsToRemoveMessagesFrom },
});
tomekUnsubmitted
Not Done
Inline Actions

Why do we remove these threads?

tomek: Why do we remove these threads?
kamilAuthorUnsubmitted
Done
Inline Actions

in this function as threads we return filteredThreads which are threads from the message store whose ids are "watched".

  1. We can either take the diff of reassignedMessageStore.threads and filteredThreads and remove those
  2. Use threadsToRemoveMessagesFrom which is created in a loop that iterates through all reassignedMessageStore.threads (see line 667) and select those which are not present in watchedThreadInfos (current solution and better in terms of perf I think)
kamil: in this function as `threads` we return `filteredThreads` which are threads from the message…
messageStoreOperations.push({
type: 'replace_threads',
payload: {
threads: updatedThreads,
},
});
messageStoreOperations.push({
type: 'remove_messages_for_threads', type: 'remove_messages_for_threads',
payload: { threadIDs: threadsToRemoveMessagesFrom }, payload: { threadIDs: threadsToRemoveMessagesFrom },
}); });
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
messages: _omit(messageIDsToRemove)(reassignedMessageStore.messages), messages: _omit(messageIDsToRemove)(reassignedMessageStore.messages),
▲ Show 20 Lines • Show All 113 Lines • ▼ Show 20 Lines const messagesResult = mergeUpdatesWithMessageInfos(
action.payload.messagesResult.truncationStatuses, action.payload.messagesResult.truncationStatuses,
); );
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === processUpdatesActionType) { } else if (action.type === processUpdatesActionType) {
if (action.payload.updatesResult.newUpdates.length === 0) { if (action.payload.updatesResult.newUpdates.length === 0) {
return { messageStoreOperations: [], messageStore }; return { messageStoreOperations: [], messageStore };
} }
const messagesResult = mergeUpdatesWithMessageInfos( const messagesResult = mergeUpdatesWithMessageInfos(
[], [],
action.payload.updatesResult.newUpdates, action.payload.updatesResult.newUpdates,
); );
const { messageStoreOperations, messageStore: newMessageStore } = const { messageStoreOperations, messageStore: newMessageStore } =
mergeNewMessages( mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
messages: newMessageStore.messages, messages: newMessageStore.messages,
threads: newMessageStore.threads, threads: newMessageStore.threads,
local: newMessageStore.local, local: newMessageStore.local,
currentAsOf: messageStore.currentAsOf, currentAsOf: messageStore.currentAsOf,
}, },
}; };
} else if ( } else if (
action.type === fullStateSyncActionType || action.type === fullStateSyncActionType ||
action.type === processMessagesActionType action.type === processMessagesActionType
) { ) {
const { messagesResult } = action.payload; const { messagesResult } = action.payload;
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === fetchSingleMostRecentMessagesFromThreadsActionTypes.success action.type === fetchSingleMostRecentMessagesFromThreadsActionTypes.success
) { ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
action.payload.truncationStatuses, action.payload.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === fetchMessagesBeforeCursorActionTypes.success || action.type === fetchMessagesBeforeCursorActionTypes.success ||
action.type === fetchMostRecentMessagesActionTypes.success action.type === fetchMostRecentMessagesActionTypes.success
) { ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
{ [action.payload.threadID]: action.payload.truncationStatus }, { [action.payload.threadID]: action.payload.truncationStatus },
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === logOutActionTypes.success || action.type === logOutActionTypes.success ||
action.type === deleteAccountActionTypes.success || action.type === deleteAccountActionTypes.success ||
action.type === deleteThreadActionTypes.success || action.type === deleteThreadActionTypes.success ||
action.type === leaveThreadActionTypes.success || action.type === leaveThreadActionTypes.success ||
action.type === setNewSessionActionType action.type === setNewSessionActionType
) { ) {
Show All 17 Lines const messagesResult = mergeUpdatesWithMessageInfos(
action.payload.newMessageInfos, action.payload.newMessageInfos,
action.payload.updatesResult.newUpdates, action.payload.updatesResult.newUpdates,
); );
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === sendMessageReportActionTypes.success) { } else if (action.type === sendMessageReportActionTypes.success) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
[action.payload.messageInfo], [action.payload.messageInfo],
{ {
[action.payload.messageInfo.threadID]: [action.payload.messageInfo.threadID]:
messageTruncationStatus.UNCHANGED, messageTruncationStatus.UNCHANGED,
}, },
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === registerActionTypes.success) { } else if (action.type === registerActionTypes.success) {
const truncationStatuses = {}; const truncationStatuses = {};
for (const messageInfo of action.payload.rawMessageInfos) { for (const messageInfo of action.payload.rawMessageInfos) {
truncationStatuses[messageInfo.threadID] = truncationStatuses[messageInfo.threadID] =
messageTruncationStatus.EXHAUSTIVE; messageTruncationStatus.EXHAUSTIVE;
} }
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
truncationStatuses, truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === changeThreadSettingsActionTypes.success || action.type === changeThreadSettingsActionTypes.success ||
action.type === removeUsersFromThreadActionTypes.success || action.type === removeUsersFromThreadActionTypes.success ||
action.type === changeThreadMemberRolesActionTypes.success || action.type === changeThreadMemberRolesActionTypes.success ||
action.type === createEntryActionTypes.success || action.type === createEntryActionTypes.success ||
action.type === saveEntryActionTypes.success || action.type === saveEntryActionTypes.success ||
action.type === restoreEntryActionTypes.success || action.type === restoreEntryActionTypes.success ||
action.type === toggleMessagePinActionTypes.success action.type === toggleMessagePinActionTypes.success
) { ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.newMessageInfos, action.payload.newMessageInfos,
{ [action.payload.threadID]: messageTruncationStatus.UNCHANGED }, { [action.payload.threadID]: messageTruncationStatus.UNCHANGED },
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === deleteEntryActionTypes.success) { } else if (action.type === deleteEntryActionTypes.success) {
const payload = action.payload; const payload = action.payload;
if (payload) { if (payload) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
payload.newMessageInfos, payload.newMessageInfos,
{ [payload.threadID]: messageTruncationStatus.UNCHANGED }, { [payload.threadID]: messageTruncationStatus.UNCHANGED },
newThreadInfos, newThreadInfos,
action.type,
); );
} }
} else if (action.type === joinThreadActionTypes.success) { } else if (action.type === joinThreadActionTypes.success) {
const messagesResult = mergeUpdatesWithMessageInfos( const messagesResult = mergeUpdatesWithMessageInfos(
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
action.payload.updatesResult.newUpdates, action.payload.updatesResult.newUpdates,
); );
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === sendEditMessageActionTypes.success) { } else if (action.type === sendEditMessageActionTypes.success) {
const { newMessageInfos } = action.payload; const { newMessageInfos } = action.payload;
const truncationStatuses = {}; const truncationStatuses = {};
for (const messageInfo of newMessageInfos) { for (const messageInfo of newMessageInfos) {
truncationStatuses[messageInfo.threadID] = truncationStatuses[messageInfo.threadID] =
messageTruncationStatus.UNCHANGED; messageTruncationStatus.UNCHANGED;
} }
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
newMessageInfos, newMessageInfos,
truncationStatuses, truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === sendTextMessageActionTypes.started || action.type === sendTextMessageActionTypes.started ||
action.type === sendMultimediaMessageActionTypes.started || action.type === sendMultimediaMessageActionTypes.started ||
action.type === sendReactionMessageActionTypes.started action.type === sendReactionMessageActionTypes.started
) { ) {
const payload = ensureRealizedThreadIDIsUsedWhenPossible( const payload = ensureRealizedThreadIDIsUsedWhenPossible(
action.payload, action.payload,
▲ Show 20 Lines • Show All 207 Lines • ▼ Show 20 Lines for (const messageInfo of action.payload.rawMessageInfos) {
messageTruncationStatus.UNCHANGED; messageTruncationStatus.UNCHANGED;
} }
const { messageStoreOperations, messageStore: newMessageStore } = const { messageStoreOperations, messageStore: newMessageStore } =
mergeNewMessages( mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
truncationStatuses, truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
messages: newMessageStore.messages, messages: newMessageStore.messages,
threads: newMessageStore.threads, threads: newMessageStore.threads,
local: newMessageStore.local, local: newMessageStore.local,
// We avoid bumping currentAsOf because notifs may include a contracted // We avoid bumping currentAsOf because notifs may include a contracted
▲ Show 20 Lines • Show All 257 Lines • ▼ Show 20 Lines const {
messageStore: messageStoreAfterReassignment, messageStore: messageStoreAfterReassignment,
} = reassignMessagesToRealizedThreads(messageStore, newThreadInfos); } = reassignMessagesToRealizedThreads(messageStore, newThreadInfos);
const processedMessageStore = processMessageStoreOperations( const processedMessageStore = processMessageStoreOperations(
messageStore, messageStore,
messageStoreOperations, messageStoreOperations,
); );
assertMessageStoreThreadsAreEqual(
processedMessageStore,
messageStoreAfterReassignment,
action.type,
);
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
...messageStoreAfterReassignment, ...messageStoreAfterReassignment,
messages: processedMessageStore.messages, messages: processedMessageStore.messages,
}, },
}; };
} else if (action.type === setClientDBStoreActionType) { } else if (action.type === setClientDBStoreActionType) {
▲ Show 20 Lines • Show All 128 Lines • Show Last 20 Lines
View Options

services/tunnelbroker/src/cxx_bridge.rs

Show First 20 Lines • Show All 130 Lines • ▼ Show 20 Lines) {
return ( return (
threadIsPending(threadID) || threadIsPending(threadID) ||
(threadInfo && (threadInfo &&
threadHasPermission(threadInfo, threadPermissions.VISIBLE) && threadHasPermission(threadInfo, threadPermissions.VISIBLE) &&
(threadInChatList(threadInfo) || watchedIDs.includes(threadID))) (threadInChatList(threadInfo) || watchedIDs.includes(threadID)))
); );
} }
// eslint-disable-next-line no-unused-vars
function assertMessageStoreThreadsAreEqual( function assertMessageStoreThreadsAreEqual(
processedMessageStore: MessageStore, processedMessageStore: MessageStore,
expectedMessageStore: MessageStore, expectedMessageStore: MessageStore,
location: string, location: string,
) { ) {
if (PROCESSED_MSG_STORE_INVARIANTS_DISABLED) { if (PROCESSED_MSG_STORE_INVARIANTS_DISABLED) {
return; return;
} }
▲ Show 20 Lines • Show All 100 Lines • ▼ Show 20 Lines const updateMsgOperation: ReplaceMessageOperation = {
payload: { id: storeMessageID, messageInfo: messages[storeMessageID] }, payload: { id: storeMessageID, messageInfo: messages[storeMessageID] },
}; };
messageStoreOperations.push(updateMsgOperation); messageStoreOperations.push(updateMsgOperation);
} }
const threads = {}; const threads = {};
const reassignedThreadIDs = []; const reassignedThreadIDs = [];
const updatedThreads = {};
const threadsToRemove = [];
tomekUnsubmitted
Not Done
Inline Actions

I guess we can make this a Set

tomek: I guess we can make this a `Set`
kamilAuthorUnsubmitted
Done
Inline Actions

What are the advantages of using Set here?

In the next lines, I iterate through message store threads only once and if the condition is met I add value to the array, which means each value will occur only once

kamil: What are the advantages of using `Set` here? In the next lines, I iterate through message…
tomekUnsubmitted
Not Done
Inline Actions

Sounds ok. Overall, sets are beneficial if we want to deduplicate or when we check if the set contains something - none of these are the case here, so an array is fine.

tomek: Sounds ok. Overall, sets are beneficial if we want to deduplicate or when we check if the set…
for (const threadID in messageStore.threads) { for (const threadID in messageStore.threads) {
const threadMessageInfo = messageStore.threads[threadID]; const threadMessageInfo = messageStore.threads[threadID];
const newThreadID = pendingToRealizedThreadIDs.get(threadID); const newThreadID = pendingToRealizedThreadIDs.get(threadID);
if (!newThreadID) { if (!newThreadID) {
threads[threadID] = threadMessageInfo; threads[threadID] = threadMessageInfo;
continue; continue;
} }
const realizedThread = messageStore.threads[newThreadID]; const realizedThread = messageStore.threads[newThreadID];
if (!realizedThread) { if (!realizedThread) {
reassignedThreadIDs.push(newThreadID); reassignedThreadIDs.push(newThreadID);
threads[newThreadID] = threadMessageInfo; threads[newThreadID] = threadMessageInfo;
updatedThreads[newThreadID] = threadMessageInfo;
threadsToRemove.push(threadID);
continue; continue;
} }
threads[newThreadID] = mergeThreadMessageInfos( threads[newThreadID] = mergeThreadMessageInfos(
threadMessageInfo, threadMessageInfo,
realizedThread, realizedThread,
messages, messages,
); );
updatedThreads[newThreadID] = threads[newThreadID];
} }
messageStoreOperations.push({
type: 'remove_threads',
payload: {
ids: threadsToRemove,
},
});
messageStoreOperations.push({
type: 'replace_threads',
payload: {
threads: updatedThreads,
},
});
tomekUnsubmitted
Not Done
Inline Actions

We should add these only if they are meaningful. It seems like threadsToRemove can be empty and we don't need to create an op for it.

tomek: We should add these only if they are meaningful. It seems like `threadsToRemove` can be empty…
kamilAuthorUnsubmitted
Done
Inline Actions

right, good catch

kamil: right, good catch
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
...messageStore, ...messageStore,
threads, threads,
messages, messages,
}, },
reassignedThreadIDs, reassignedThreadIDs,
}; };
} }
type MergeNewMessagesResult = { type MergeNewMessagesResult = {
+messageStoreOperations: $ReadOnlyArray<MessageStoreOperation>, +messageStoreOperations: $ReadOnlyArray<MessageStoreOperation>,
+messageStore: MessageStore, +messageStore: MessageStore,
}; };
// oldMessageStore is from the old state // oldMessageStore is from the old state
// newMessageInfos, truncationStatus come from server // newMessageInfos, truncationStatus come from server
function mergeNewMessages( function mergeNewMessages(
oldMessageStore: MessageStore, oldMessageStore: MessageStore,
newMessageInfos: $ReadOnlyArray<RawMessageInfo>, newMessageInfos: $ReadOnlyArray<RawMessageInfo>,
truncationStatus: { [threadID: string]: MessageTruncationStatus }, truncationStatus: { [threadID: string]: MessageTruncationStatus },
threadInfos: { +[threadID: string]: RawThreadInfo }, threadInfos: { +[threadID: string]: RawThreadInfo },
actionType: string,
): MergeNewMessagesResult { ): MergeNewMessagesResult {
const { const {
messageStoreOperations: updateWithLatestThreadInfosOps, messageStoreOperations: updateWithLatestThreadInfosOps,
messageStore: messageStoreUpdatedWithLatestThreadInfos, messageStore: messageStoreUpdatedWithLatestThreadInfos,
reassignedThreadIDs, reassignedThreadIDs,
} = updateMessageStoreWithLatestThreadInfos(oldMessageStore, threadInfos); } = updateMessageStoreWithLatestThreadInfos(oldMessageStore, threadInfos);
const messageStoreAfterUpdateOps = processMessageStoreOperations( const messageStoreAfterUpdateOps = processMessageStoreOperations(
oldMessageStore, oldMessageStore,
updateWithLatestThreadInfosOps, updateWithLatestThreadInfosOps,
); );
assertMessageStoreThreadsAreEqual(
messageStoreAfterUpdateOps,
messageStoreUpdatedWithLatestThreadInfos,
`${actionType} | reassignment and filtering`,
);
const updatedMessageStore = { const updatedMessageStore = {
...messageStoreUpdatedWithLatestThreadInfos, ...messageStoreUpdatedWithLatestThreadInfos,
messages: messageStoreAfterUpdateOps.messages, messages: messageStoreAfterUpdateOps.messages,
}; };
const localIDsToServerIDs: Map<string, string> = new Map(); const localIDsToServerIDs: Map<string, string> = new Map();
const watchedThreadIDs = [ const watchedThreadIDs = [
▲ Show 20 Lines • Show All 121 Lines • ▼ Show 20 Lines): MergeNewMessagesResult {
const newMessageOps: MessageStoreOperation[] = []; const newMessageOps: MessageStoreOperation[] = [];
const threadsToMessageIDs = mapThreadsToMessageIDsFromOrderedMessageInfos( const threadsToMessageIDs = mapThreadsToMessageIDsFromOrderedMessageInfos(
orderedNewMessageInfos, orderedNewMessageInfos,
); );
const oldMessageInfosToCombine = []; const oldMessageInfosToCombine = [];
const threadsThatNeedMessageIDsResorted = []; const threadsThatNeedMessageIDsResorted = [];
const local = {}; const local = {};
const updatedThreads = {};
const threads = _flow( const threads = _flow(
_mapValuesWithKeys((messageIDs: string[], threadID: string) => { _mapValuesWithKeys((messageIDs: string[], threadID: string) => {
const oldThread = updatedMessageStore.threads[threadID]; const oldThread = updatedMessageStore.threads[threadID];
const truncate = truncationStatus[threadID]; const truncate = truncationStatus[threadID];
if (!oldThread) { if (!oldThread) {
return { updatedThreads[threadID] = {
...newThread(), ...newThread(),
messageIDs, messageIDs,
startReached: truncate === messageTruncationStatus.EXHAUSTIVE, startReached: truncate === messageTruncationStatus.EXHAUSTIVE,
}; };
return updatedThreads[threadID];
} }
let oldMessageIDsUnchanged = true; let oldMessageIDsUnchanged = true;
const oldMessageIDs = oldThread.messageIDs.map(oldID => { const oldMessageIDs = oldThread.messageIDs.map(oldID => {
const newID = localIDsToServerIDs.get(oldID); const newID = localIDsToServerIDs.get(oldID);
if (newID !== null && newID !== undefined) { if (newID !== null && newID !== undefined) {
oldMessageIDsUnchanged = false; oldMessageIDsUnchanged = false;
return newID; return newID;
} }
return oldID; return oldID;
}); });
if (truncate === messageTruncationStatus.TRUNCATED) { if (truncate === messageTruncationStatus.TRUNCATED) {
// If the result set in the payload isn't contiguous with what we have // 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 // 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 // it with the result set. We do this to achieve our two goals for the
// messageStore: currentness and contiguousness. // messageStore: currentness and contiguousness.
newMessageOps.push({ newMessageOps.push({
type: 'remove_messages_for_threads', type: 'remove_messages_for_threads',
payload: { threadIDs: [threadID] }, payload: { threadIDs: [threadID] },
}); });
return { updatedThreads[threadID] = {
messageIDs, messageIDs,
startReached: false, startReached: false,
lastNavigatedTo: oldThread.lastNavigatedTo, lastNavigatedTo: oldThread.lastNavigatedTo,
lastPruned: oldThread.lastPruned, lastPruned: oldThread.lastPruned,
}; };
return updatedThreads[threadID];
} }
const oldNotInNew = _difference(oldMessageIDs)(messageIDs); const oldNotInNew = _difference(oldMessageIDs)(messageIDs);
for (const id of oldNotInNew) { for (const id of oldNotInNew) {
const oldMessageInfo = updatedMessageStore.messages[id]; const oldMessageInfo = updatedMessageStore.messages[id];
invariant(oldMessageInfo, `could not find ${id} in messageStore`); invariant(oldMessageInfo, `could not find ${id} in messageStore`);
oldMessageInfosToCombine.push(oldMessageInfo); oldMessageInfosToCombine.push(oldMessageInfo);
const localInfo = updatedMessageStore.local[id]; const localInfo = updatedMessageStore.local[id];
if (localInfo) { if (localInfo) {
local[id] = localInfo; local[id] = localInfo;
} }
} }
const startReached = const startReached =
oldThread.startReached || oldThread.startReached ||
truncate === messageTruncationStatus.EXHAUSTIVE; truncate === messageTruncationStatus.EXHAUSTIVE;
if (_difference(messageIDs)(oldMessageIDs).length === 0) { if (_difference(messageIDs)(oldMessageIDs).length === 0) {
if (startReached === oldThread.startReached && oldMessageIDsUnchanged) { if (startReached === oldThread.startReached && oldMessageIDsUnchanged) {
return oldThread; return oldThread;
} }
return { updatedThreads[threadID] = {
messageIDs: oldMessageIDs, messageIDs: oldMessageIDs,
startReached, startReached,
lastNavigatedTo: oldThread.lastNavigatedTo, lastNavigatedTo: oldThread.lastNavigatedTo,
lastPruned: oldThread.lastPruned, lastPruned: oldThread.lastPruned,
}; };
return updatedThreads[threadID];
} }
const mergedMessageIDs = [...messageIDs, ...oldNotInNew]; const mergedMessageIDs = [...messageIDs, ...oldNotInNew];
threadsThatNeedMessageIDsResorted.push(threadID); threadsThatNeedMessageIDsResorted.push(threadID);
//NOTE: this will be updated after sorting
tomekUnsubmitted
Not Done
Inline Actions

This comment sounds confusing to me. Is it something you will do after "sorting" or "sorting" is a part of this code and the value is somehow updated after it? Please make this less confusing.

tomek: This comment sounds confusing to me. Is it something you will do after "sorting" or "sorting"…
kamilAuthorUnsubmitted
Done
Inline Actions

this comment was a note I made while developing and forgot to remove, I will simply get rid of this comment

kamil: this comment was a note I made while developing and forgot to remove, I will simply get rid of…
return { return {
messageIDs: mergedMessageIDs, messageIDs: mergedMessageIDs,
startReached, startReached,
lastNavigatedTo: oldThread.lastNavigatedTo, lastNavigatedTo: oldThread.lastNavigatedTo,
lastPruned: oldThread.lastPruned, lastPruned: oldThread.lastPruned,
}; };
}), }),
_pickBy(thread => !!thread), _pickBy(thread => !!thread),
)(threadsToMessageIDs); )(threadsToMessageIDs);
for (const threadID in updatedMessageStore.threads) { for (const threadID in updatedMessageStore.threads) {
if (threads[threadID]) { if (threads[threadID]) {
continue; continue;
} }
let thread = updatedMessageStore.threads[threadID]; let thread = updatedMessageStore.threads[threadID];
const truncate = truncationStatus[threadID]; const truncate = truncationStatus[threadID];
if (truncate === messageTruncationStatus.EXHAUSTIVE) { if (truncate === messageTruncationStatus.EXHAUSTIVE) {
thread = { thread = {
...thread, ...thread,
startReached: true, startReached: true,
}; };
} }
threads[threadID] = thread; threads[threadID] = thread;
updatedThreads[threadID] = thread;
for (const id of thread.messageIDs) { for (const id of thread.messageIDs) {
const messageInfo = updatedMessageStore.messages[id]; const messageInfo = updatedMessageStore.messages[id];
if (messageInfo) { if (messageInfo) {
oldMessageInfosToCombine.push(messageInfo); oldMessageInfosToCombine.push(messageInfo);
} }
const localInfo = updatedMessageStore.local[id]; const localInfo = updatedMessageStore.local[id];
if (localInfo) { if (localInfo) {
local[id] = localInfo; local[id] = localInfo;
Show All 20 Lines newMessageOps.push({
payload: { ids: [...localIDsToServerIDs.keys()] }, payload: { ids: [...localIDsToServerIDs.keys()] },
}); });
} }
for (const threadID of threadsThatNeedMessageIDsResorted) { for (const threadID of threadsThatNeedMessageIDsResorted) {
threads[threadID].messageIDs = sortMessageIDs(messages)( threads[threadID].messageIDs = sortMessageIDs(messages)(
threads[threadID].messageIDs, threads[threadID].messageIDs,
); );
updatedThreads[threadID] = threads[threadID];
} }
const currentAsOf = Math.max( const currentAsOf = Math.max(
orderedNewMessageInfos.length > 0 ? orderedNewMessageInfos[0].time : 0, orderedNewMessageInfos.length > 0 ? orderedNewMessageInfos[0].time : 0,
updatedMessageStore.currentAsOf, updatedMessageStore.currentAsOf,
); );
newMessageOps.push({
type: 'replace_threads',
payload: {
threads: updatedThreads,
},
});
const processedMessageStore = processMessageStoreOperations( const processedMessageStore = processMessageStoreOperations(
updatedMessageStore, updatedMessageStore,
newMessageOps, newMessageOps,
); );
const messageStore = {
messages: processedMessageStore.messages,
threads,
tomekUnsubmitted
Not Done
Inline Actions
messages: processedMessageStore.messages,
- threads,
+ threads: processedMessageStore.threads,
local,

Should we also take threads prop of processedMessageStore?

tomek: Should we also take `threads` prop of `processedMessageStore`?
kamilAuthorUnsubmitted
Done
Inline Actions

Lets assume that:
A - old state
B - state modified in reducer, the current logic
C - state modified by ops

previously it worked like this:

  1. We have an old state (l A)
  2. We modify the state (A -> B)
  3. We return the modified state (B)

this diff this changes to:

  1. We have an old state (A)
  2. We modify the state and generate ops (A ->B)
  3. We take the old state, apply ops, and check if this matches the modified state (A -> C, and then checking B === C)
  4. We return modified state (B)

In future, if devs will not encounter any issues:

  1. We have old state (A)
  2. We will remove modifying the state and only generate ops (A not changed)
  3. Apply ops to old state (A -> C)
  4. Return state modified by ops (C)
kamil: Lets assume that: A - old state B - state modified in reducer, the current logic C - state…
tomekUnsubmitted
Not Done
Inline Actions

Thanks for the explanation - it makes sense!

tomek: Thanks for the explanation - it makes sense!
local,
currentAsOf,
};
assertMessageStoreThreadsAreEqual(
processedMessageStore,
messageStore,
`${actionType} | processed`,
);
return { return {
messageStoreOperations: [ messageStoreOperations: [
...updateWithLatestThreadInfosOps, ...updateWithLatestThreadInfosOps,
...newMessageOps, ...newMessageOps,
], ],
messageStore: { messageStore,
messages: processedMessageStore.messages,
threads,
local,
currentAsOf,
},
}; };
} }
type UpdateMessageStoreWithLatestThreadInfosResult = { type UpdateMessageStoreWithLatestThreadInfosResult = {
+messageStoreOperations: MessageStoreOperation[], +messageStoreOperations: MessageStoreOperation[],
+messageStore: MessageStore, +messageStore: MessageStore,
+reassignedThreadIDs: string[], +reassignedThreadIDs: string[],
}; };
Show All 24 Lines if (watchedThreadInfos[threadID]) {
continue; continue;
} }
threadsToRemoveMessagesFrom.push(threadID); threadsToRemoveMessagesFrom.push(threadID);
for (const id of reassignedMessageStore.threads[threadID].messageIDs) { for (const id of reassignedMessageStore.threads[threadID].messageIDs) {
messageIDsToRemove.push(id); messageIDsToRemove.push(id);
} }
} }
const updatedThreads = {};
for (const threadID in threadInfos) { for (const threadID in threadInfos) {
const threadInfo = threadInfos[threadID]; const threadInfo = threadInfos[threadID];
if ( if (
isThreadWatched(threadID, threadInfo, watchedIDs) && isThreadWatched(threadID, threadInfo, watchedIDs) &&
!filteredThreads[threadID] !filteredThreads[threadID]
) { ) {
filteredThreads[threadID] = newThread(); filteredThreads[threadID] = newThread();
updatedThreads[threadID] = filteredThreads[threadID];
} }
} }
messageStoreOperations.push({ messageStoreOperations.push({
type: 'remove_threads',
payload: { ids: threadsToRemoveMessagesFrom },
});
tomekUnsubmitted
Not Done
Inline Actions

Why do we remove these threads?

tomek: Why do we remove these threads?
kamilAuthorUnsubmitted
Done
Inline Actions

in this function as threads we return filteredThreads which are threads from the message store whose ids are "watched".

  1. We can either take the diff of reassignedMessageStore.threads and filteredThreads and remove those
  2. Use threadsToRemoveMessagesFrom which is created in a loop that iterates through all reassignedMessageStore.threads (see line 667) and select those which are not present in watchedThreadInfos (current solution and better in terms of perf I think)
kamil: in this function as `threads` we return `filteredThreads` which are threads from the message…
messageStoreOperations.push({
type: 'replace_threads',
payload: {
threads: updatedThreads,
},
});
messageStoreOperations.push({
type: 'remove_messages_for_threads', type: 'remove_messages_for_threads',
payload: { threadIDs: threadsToRemoveMessagesFrom }, payload: { threadIDs: threadsToRemoveMessagesFrom },
}); });
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
messages: _omit(messageIDsToRemove)(reassignedMessageStore.messages), messages: _omit(messageIDsToRemove)(reassignedMessageStore.messages),
▲ Show 20 Lines • Show All 113 Lines • ▼ Show 20 Lines const messagesResult = mergeUpdatesWithMessageInfos(
action.payload.messagesResult.truncationStatuses, action.payload.messagesResult.truncationStatuses,
); );
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === processUpdatesActionType) { } else if (action.type === processUpdatesActionType) {
if (action.payload.updatesResult.newUpdates.length === 0) { if (action.payload.updatesResult.newUpdates.length === 0) {
return { messageStoreOperations: [], messageStore }; return { messageStoreOperations: [], messageStore };
} }
const messagesResult = mergeUpdatesWithMessageInfos( const messagesResult = mergeUpdatesWithMessageInfos(
[], [],
action.payload.updatesResult.newUpdates, action.payload.updatesResult.newUpdates,
); );
const { messageStoreOperations, messageStore: newMessageStore } = const { messageStoreOperations, messageStore: newMessageStore } =
mergeNewMessages( mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
messages: newMessageStore.messages, messages: newMessageStore.messages,
threads: newMessageStore.threads, threads: newMessageStore.threads,
local: newMessageStore.local, local: newMessageStore.local,
currentAsOf: messageStore.currentAsOf, currentAsOf: messageStore.currentAsOf,
}, },
}; };
} else if ( } else if (
action.type === fullStateSyncActionType || action.type === fullStateSyncActionType ||
action.type === processMessagesActionType action.type === processMessagesActionType
) { ) {
const { messagesResult } = action.payload; const { messagesResult } = action.payload;
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === fetchSingleMostRecentMessagesFromThreadsActionTypes.success action.type === fetchSingleMostRecentMessagesFromThreadsActionTypes.success
) { ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
action.payload.truncationStatuses, action.payload.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === fetchMessagesBeforeCursorActionTypes.success || action.type === fetchMessagesBeforeCursorActionTypes.success ||
action.type === fetchMostRecentMessagesActionTypes.success action.type === fetchMostRecentMessagesActionTypes.success
) { ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
{ [action.payload.threadID]: action.payload.truncationStatus }, { [action.payload.threadID]: action.payload.truncationStatus },
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === logOutActionTypes.success || action.type === logOutActionTypes.success ||
action.type === deleteAccountActionTypes.success || action.type === deleteAccountActionTypes.success ||
action.type === deleteThreadActionTypes.success || action.type === deleteThreadActionTypes.success ||
action.type === leaveThreadActionTypes.success || action.type === leaveThreadActionTypes.success ||
action.type === setNewSessionActionType action.type === setNewSessionActionType
) { ) {
Show All 17 Lines const messagesResult = mergeUpdatesWithMessageInfos(
action.payload.newMessageInfos, action.payload.newMessageInfos,
action.payload.updatesResult.newUpdates, action.payload.updatesResult.newUpdates,
); );
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === sendMessageReportActionTypes.success) { } else if (action.type === sendMessageReportActionTypes.success) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
[action.payload.messageInfo], [action.payload.messageInfo],
{ {
[action.payload.messageInfo.threadID]: [action.payload.messageInfo.threadID]:
messageTruncationStatus.UNCHANGED, messageTruncationStatus.UNCHANGED,
}, },
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === registerActionTypes.success) { } else if (action.type === registerActionTypes.success) {
const truncationStatuses = {}; const truncationStatuses = {};
for (const messageInfo of action.payload.rawMessageInfos) { for (const messageInfo of action.payload.rawMessageInfos) {
truncationStatuses[messageInfo.threadID] = truncationStatuses[messageInfo.threadID] =
messageTruncationStatus.EXHAUSTIVE; messageTruncationStatus.EXHAUSTIVE;
} }
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
truncationStatuses, truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === changeThreadSettingsActionTypes.success || action.type === changeThreadSettingsActionTypes.success ||
action.type === removeUsersFromThreadActionTypes.success || action.type === removeUsersFromThreadActionTypes.success ||
action.type === changeThreadMemberRolesActionTypes.success || action.type === changeThreadMemberRolesActionTypes.success ||
action.type === createEntryActionTypes.success || action.type === createEntryActionTypes.success ||
action.type === saveEntryActionTypes.success || action.type === saveEntryActionTypes.success ||
action.type === restoreEntryActionTypes.success || action.type === restoreEntryActionTypes.success ||
action.type === toggleMessagePinActionTypes.success action.type === toggleMessagePinActionTypes.success
) { ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.newMessageInfos, action.payload.newMessageInfos,
{ [action.payload.threadID]: messageTruncationStatus.UNCHANGED }, { [action.payload.threadID]: messageTruncationStatus.UNCHANGED },
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === deleteEntryActionTypes.success) { } else if (action.type === deleteEntryActionTypes.success) {
const payload = action.payload; const payload = action.payload;
if (payload) { if (payload) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
payload.newMessageInfos, payload.newMessageInfos,
{ [payload.threadID]: messageTruncationStatus.UNCHANGED }, { [payload.threadID]: messageTruncationStatus.UNCHANGED },
newThreadInfos, newThreadInfos,
action.type,
); );
} }
} else if (action.type === joinThreadActionTypes.success) { } else if (action.type === joinThreadActionTypes.success) {
const messagesResult = mergeUpdatesWithMessageInfos( const messagesResult = mergeUpdatesWithMessageInfos(
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
action.payload.updatesResult.newUpdates, action.payload.updatesResult.newUpdates,
); );
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === sendEditMessageActionTypes.success) { } else if (action.type === sendEditMessageActionTypes.success) {
const { newMessageInfos } = action.payload; const { newMessageInfos } = action.payload;
const truncationStatuses = {}; const truncationStatuses = {};
for (const messageInfo of newMessageInfos) { for (const messageInfo of newMessageInfos) {
truncationStatuses[messageInfo.threadID] = truncationStatuses[messageInfo.threadID] =
messageTruncationStatus.UNCHANGED; messageTruncationStatus.UNCHANGED;
} }
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
newMessageInfos, newMessageInfos,
truncationStatuses, truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === sendTextMessageActionTypes.started || action.type === sendTextMessageActionTypes.started ||
action.type === sendMultimediaMessageActionTypes.started || action.type === sendMultimediaMessageActionTypes.started ||
action.type === sendReactionMessageActionTypes.started action.type === sendReactionMessageActionTypes.started
) { ) {
const payload = ensureRealizedThreadIDIsUsedWhenPossible( const payload = ensureRealizedThreadIDIsUsedWhenPossible(
action.payload, action.payload,
▲ Show 20 Lines • Show All 207 Lines • ▼ Show 20 Lines for (const messageInfo of action.payload.rawMessageInfos) {
messageTruncationStatus.UNCHANGED; messageTruncationStatus.UNCHANGED;
} }
const { messageStoreOperations, messageStore: newMessageStore } = const { messageStoreOperations, messageStore: newMessageStore } =
mergeNewMessages( mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
truncationStatuses, truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
messages: newMessageStore.messages, messages: newMessageStore.messages,
threads: newMessageStore.threads, threads: newMessageStore.threads,
local: newMessageStore.local, local: newMessageStore.local,
// We avoid bumping currentAsOf because notifs may include a contracted // We avoid bumping currentAsOf because notifs may include a contracted
▲ Show 20 Lines • Show All 257 Lines • ▼ Show 20 Lines const {
messageStore: messageStoreAfterReassignment, messageStore: messageStoreAfterReassignment,
} = reassignMessagesToRealizedThreads(messageStore, newThreadInfos); } = reassignMessagesToRealizedThreads(messageStore, newThreadInfos);
const processedMessageStore = processMessageStoreOperations( const processedMessageStore = processMessageStoreOperations(
messageStore, messageStore,
messageStoreOperations, messageStoreOperations,
); );
assertMessageStoreThreadsAreEqual(
processedMessageStore,
messageStoreAfterReassignment,
action.type,
);
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
...messageStoreAfterReassignment, ...messageStoreAfterReassignment,
messages: processedMessageStore.messages, messages: processedMessageStore.messages,
}, },
}; };
} else if (action.type === setClientDBStoreActionType) { } else if (action.type === setClientDBStoreActionType) {
▲ Show 20 Lines • Show All 128 Lines • Show Last 20 Lines
View Options

services/tunnelbroker/src/main.rs

Show First 20 Lines • Show All 130 Lines • ▼ Show 20 Lines) {
return ( return (
threadIsPending(threadID) || threadIsPending(threadID) ||
(threadInfo && (threadInfo &&
threadHasPermission(threadInfo, threadPermissions.VISIBLE) && threadHasPermission(threadInfo, threadPermissions.VISIBLE) &&
(threadInChatList(threadInfo) || watchedIDs.includes(threadID))) (threadInChatList(threadInfo) || watchedIDs.includes(threadID)))
); );
} }
// eslint-disable-next-line no-unused-vars
function assertMessageStoreThreadsAreEqual( function assertMessageStoreThreadsAreEqual(
processedMessageStore: MessageStore, processedMessageStore: MessageStore,
expectedMessageStore: MessageStore, expectedMessageStore: MessageStore,
location: string, location: string,
) { ) {
if (PROCESSED_MSG_STORE_INVARIANTS_DISABLED) { if (PROCESSED_MSG_STORE_INVARIANTS_DISABLED) {
return; return;
} }
▲ Show 20 Lines • Show All 100 Lines • ▼ Show 20 Lines const updateMsgOperation: ReplaceMessageOperation = {
payload: { id: storeMessageID, messageInfo: messages[storeMessageID] }, payload: { id: storeMessageID, messageInfo: messages[storeMessageID] },
}; };
messageStoreOperations.push(updateMsgOperation); messageStoreOperations.push(updateMsgOperation);
} }
const threads = {}; const threads = {};
const reassignedThreadIDs = []; const reassignedThreadIDs = [];
const updatedThreads = {};
const threadsToRemove = [];
tomekUnsubmitted
Not Done
Inline Actions

I guess we can make this a Set

tomek: I guess we can make this a `Set`
kamilAuthorUnsubmitted
Done
Inline Actions

What are the advantages of using Set here?

In the next lines, I iterate through message store threads only once and if the condition is met I add value to the array, which means each value will occur only once

kamil: What are the advantages of using `Set` here? In the next lines, I iterate through message…
tomekUnsubmitted
Not Done
Inline Actions

Sounds ok. Overall, sets are beneficial if we want to deduplicate or when we check if the set contains something - none of these are the case here, so an array is fine.

tomek: Sounds ok. Overall, sets are beneficial if we want to deduplicate or when we check if the set…
for (const threadID in messageStore.threads) { for (const threadID in messageStore.threads) {
const threadMessageInfo = messageStore.threads[threadID]; const threadMessageInfo = messageStore.threads[threadID];
const newThreadID = pendingToRealizedThreadIDs.get(threadID); const newThreadID = pendingToRealizedThreadIDs.get(threadID);
if (!newThreadID) { if (!newThreadID) {
threads[threadID] = threadMessageInfo; threads[threadID] = threadMessageInfo;
continue; continue;
} }
const realizedThread = messageStore.threads[newThreadID]; const realizedThread = messageStore.threads[newThreadID];
if (!realizedThread) { if (!realizedThread) {
reassignedThreadIDs.push(newThreadID); reassignedThreadIDs.push(newThreadID);
threads[newThreadID] = threadMessageInfo; threads[newThreadID] = threadMessageInfo;
updatedThreads[newThreadID] = threadMessageInfo;
threadsToRemove.push(threadID);
continue; continue;
} }
threads[newThreadID] = mergeThreadMessageInfos( threads[newThreadID] = mergeThreadMessageInfos(
threadMessageInfo, threadMessageInfo,
realizedThread, realizedThread,
messages, messages,
); );
updatedThreads[newThreadID] = threads[newThreadID];
} }
messageStoreOperations.push({
type: 'remove_threads',
payload: {
ids: threadsToRemove,
},
});
messageStoreOperations.push({
type: 'replace_threads',
payload: {
threads: updatedThreads,
},
});
tomekUnsubmitted
Not Done
Inline Actions

We should add these only if they are meaningful. It seems like threadsToRemove can be empty and we don't need to create an op for it.

tomek: We should add these only if they are meaningful. It seems like `threadsToRemove` can be empty…
kamilAuthorUnsubmitted
Done
Inline Actions

right, good catch

kamil: right, good catch
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
...messageStore, ...messageStore,
threads, threads,
messages, messages,
}, },
reassignedThreadIDs, reassignedThreadIDs,
}; };
} }
type MergeNewMessagesResult = { type MergeNewMessagesResult = {
+messageStoreOperations: $ReadOnlyArray<MessageStoreOperation>, +messageStoreOperations: $ReadOnlyArray<MessageStoreOperation>,
+messageStore: MessageStore, +messageStore: MessageStore,
}; };
// oldMessageStore is from the old state // oldMessageStore is from the old state
// newMessageInfos, truncationStatus come from server // newMessageInfos, truncationStatus come from server
function mergeNewMessages( function mergeNewMessages(
oldMessageStore: MessageStore, oldMessageStore: MessageStore,
newMessageInfos: $ReadOnlyArray<RawMessageInfo>, newMessageInfos: $ReadOnlyArray<RawMessageInfo>,
truncationStatus: { [threadID: string]: MessageTruncationStatus }, truncationStatus: { [threadID: string]: MessageTruncationStatus },
threadInfos: { +[threadID: string]: RawThreadInfo }, threadInfos: { +[threadID: string]: RawThreadInfo },
actionType: string,
): MergeNewMessagesResult { ): MergeNewMessagesResult {
const { const {
messageStoreOperations: updateWithLatestThreadInfosOps, messageStoreOperations: updateWithLatestThreadInfosOps,
messageStore: messageStoreUpdatedWithLatestThreadInfos, messageStore: messageStoreUpdatedWithLatestThreadInfos,
reassignedThreadIDs, reassignedThreadIDs,
} = updateMessageStoreWithLatestThreadInfos(oldMessageStore, threadInfos); } = updateMessageStoreWithLatestThreadInfos(oldMessageStore, threadInfos);
const messageStoreAfterUpdateOps = processMessageStoreOperations( const messageStoreAfterUpdateOps = processMessageStoreOperations(
oldMessageStore, oldMessageStore,
updateWithLatestThreadInfosOps, updateWithLatestThreadInfosOps,
); );
assertMessageStoreThreadsAreEqual(
messageStoreAfterUpdateOps,
messageStoreUpdatedWithLatestThreadInfos,
`${actionType} | reassignment and filtering`,
);
const updatedMessageStore = { const updatedMessageStore = {
...messageStoreUpdatedWithLatestThreadInfos, ...messageStoreUpdatedWithLatestThreadInfos,
messages: messageStoreAfterUpdateOps.messages, messages: messageStoreAfterUpdateOps.messages,
}; };
const localIDsToServerIDs: Map<string, string> = new Map(); const localIDsToServerIDs: Map<string, string> = new Map();
const watchedThreadIDs = [ const watchedThreadIDs = [
▲ Show 20 Lines • Show All 121 Lines • ▼ Show 20 Lines): MergeNewMessagesResult {
const newMessageOps: MessageStoreOperation[] = []; const newMessageOps: MessageStoreOperation[] = [];
const threadsToMessageIDs = mapThreadsToMessageIDsFromOrderedMessageInfos( const threadsToMessageIDs = mapThreadsToMessageIDsFromOrderedMessageInfos(
orderedNewMessageInfos, orderedNewMessageInfos,
); );
const oldMessageInfosToCombine = []; const oldMessageInfosToCombine = [];
const threadsThatNeedMessageIDsResorted = []; const threadsThatNeedMessageIDsResorted = [];
const local = {}; const local = {};
const updatedThreads = {};
const threads = _flow( const threads = _flow(
_mapValuesWithKeys((messageIDs: string[], threadID: string) => { _mapValuesWithKeys((messageIDs: string[], threadID: string) => {
const oldThread = updatedMessageStore.threads[threadID]; const oldThread = updatedMessageStore.threads[threadID];
const truncate = truncationStatus[threadID]; const truncate = truncationStatus[threadID];
if (!oldThread) { if (!oldThread) {
return { updatedThreads[threadID] = {
...newThread(), ...newThread(),
messageIDs, messageIDs,
startReached: truncate === messageTruncationStatus.EXHAUSTIVE, startReached: truncate === messageTruncationStatus.EXHAUSTIVE,
}; };
return updatedThreads[threadID];
} }
let oldMessageIDsUnchanged = true; let oldMessageIDsUnchanged = true;
const oldMessageIDs = oldThread.messageIDs.map(oldID => { const oldMessageIDs = oldThread.messageIDs.map(oldID => {
const newID = localIDsToServerIDs.get(oldID); const newID = localIDsToServerIDs.get(oldID);
if (newID !== null && newID !== undefined) { if (newID !== null && newID !== undefined) {
oldMessageIDsUnchanged = false; oldMessageIDsUnchanged = false;
return newID; return newID;
} }
return oldID; return oldID;
}); });
if (truncate === messageTruncationStatus.TRUNCATED) { if (truncate === messageTruncationStatus.TRUNCATED) {
// If the result set in the payload isn't contiguous with what we have // 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 // 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 // it with the result set. We do this to achieve our two goals for the
// messageStore: currentness and contiguousness. // messageStore: currentness and contiguousness.
newMessageOps.push({ newMessageOps.push({
type: 'remove_messages_for_threads', type: 'remove_messages_for_threads',
payload: { threadIDs: [threadID] }, payload: { threadIDs: [threadID] },
}); });
return { updatedThreads[threadID] = {
messageIDs, messageIDs,
startReached: false, startReached: false,
lastNavigatedTo: oldThread.lastNavigatedTo, lastNavigatedTo: oldThread.lastNavigatedTo,
lastPruned: oldThread.lastPruned, lastPruned: oldThread.lastPruned,
}; };
return updatedThreads[threadID];
} }
const oldNotInNew = _difference(oldMessageIDs)(messageIDs); const oldNotInNew = _difference(oldMessageIDs)(messageIDs);
for (const id of oldNotInNew) { for (const id of oldNotInNew) {
const oldMessageInfo = updatedMessageStore.messages[id]; const oldMessageInfo = updatedMessageStore.messages[id];
invariant(oldMessageInfo, `could not find ${id} in messageStore`); invariant(oldMessageInfo, `could not find ${id} in messageStore`);
oldMessageInfosToCombine.push(oldMessageInfo); oldMessageInfosToCombine.push(oldMessageInfo);
const localInfo = updatedMessageStore.local[id]; const localInfo = updatedMessageStore.local[id];
if (localInfo) { if (localInfo) {
local[id] = localInfo; local[id] = localInfo;
} }
} }
const startReached = const startReached =
oldThread.startReached || oldThread.startReached ||
truncate === messageTruncationStatus.EXHAUSTIVE; truncate === messageTruncationStatus.EXHAUSTIVE;
if (_difference(messageIDs)(oldMessageIDs).length === 0) { if (_difference(messageIDs)(oldMessageIDs).length === 0) {
if (startReached === oldThread.startReached && oldMessageIDsUnchanged) { if (startReached === oldThread.startReached && oldMessageIDsUnchanged) {
return oldThread; return oldThread;
} }
return { updatedThreads[threadID] = {
messageIDs: oldMessageIDs, messageIDs: oldMessageIDs,
startReached, startReached,
lastNavigatedTo: oldThread.lastNavigatedTo, lastNavigatedTo: oldThread.lastNavigatedTo,
lastPruned: oldThread.lastPruned, lastPruned: oldThread.lastPruned,
}; };
return updatedThreads[threadID];
} }
const mergedMessageIDs = [...messageIDs, ...oldNotInNew]; const mergedMessageIDs = [...messageIDs, ...oldNotInNew];
threadsThatNeedMessageIDsResorted.push(threadID); threadsThatNeedMessageIDsResorted.push(threadID);
//NOTE: this will be updated after sorting
tomekUnsubmitted
Not Done
Inline Actions

This comment sounds confusing to me. Is it something you will do after "sorting" or "sorting" is a part of this code and the value is somehow updated after it? Please make this less confusing.

tomek: This comment sounds confusing to me. Is it something you will do after "sorting" or "sorting"…
kamilAuthorUnsubmitted
Done
Inline Actions

this comment was a note I made while developing and forgot to remove, I will simply get rid of this comment

kamil: this comment was a note I made while developing and forgot to remove, I will simply get rid of…
return { return {
messageIDs: mergedMessageIDs, messageIDs: mergedMessageIDs,
startReached, startReached,
lastNavigatedTo: oldThread.lastNavigatedTo, lastNavigatedTo: oldThread.lastNavigatedTo,
lastPruned: oldThread.lastPruned, lastPruned: oldThread.lastPruned,
}; };
}), }),
_pickBy(thread => !!thread), _pickBy(thread => !!thread),
)(threadsToMessageIDs); )(threadsToMessageIDs);
for (const threadID in updatedMessageStore.threads) { for (const threadID in updatedMessageStore.threads) {
if (threads[threadID]) { if (threads[threadID]) {
continue; continue;
} }
let thread = updatedMessageStore.threads[threadID]; let thread = updatedMessageStore.threads[threadID];
const truncate = truncationStatus[threadID]; const truncate = truncationStatus[threadID];
if (truncate === messageTruncationStatus.EXHAUSTIVE) { if (truncate === messageTruncationStatus.EXHAUSTIVE) {
thread = { thread = {
...thread, ...thread,
startReached: true, startReached: true,
}; };
} }
threads[threadID] = thread; threads[threadID] = thread;
updatedThreads[threadID] = thread;
for (const id of thread.messageIDs) { for (const id of thread.messageIDs) {
const messageInfo = updatedMessageStore.messages[id]; const messageInfo = updatedMessageStore.messages[id];
if (messageInfo) { if (messageInfo) {
oldMessageInfosToCombine.push(messageInfo); oldMessageInfosToCombine.push(messageInfo);
} }
const localInfo = updatedMessageStore.local[id]; const localInfo = updatedMessageStore.local[id];
if (localInfo) { if (localInfo) {
local[id] = localInfo; local[id] = localInfo;
Show All 20 Lines newMessageOps.push({
payload: { ids: [...localIDsToServerIDs.keys()] }, payload: { ids: [...localIDsToServerIDs.keys()] },
}); });
} }
for (const threadID of threadsThatNeedMessageIDsResorted) { for (const threadID of threadsThatNeedMessageIDsResorted) {
threads[threadID].messageIDs = sortMessageIDs(messages)( threads[threadID].messageIDs = sortMessageIDs(messages)(
threads[threadID].messageIDs, threads[threadID].messageIDs,
); );
updatedThreads[threadID] = threads[threadID];
} }
const currentAsOf = Math.max( const currentAsOf = Math.max(
orderedNewMessageInfos.length > 0 ? orderedNewMessageInfos[0].time : 0, orderedNewMessageInfos.length > 0 ? orderedNewMessageInfos[0].time : 0,
updatedMessageStore.currentAsOf, updatedMessageStore.currentAsOf,
); );
newMessageOps.push({
type: 'replace_threads',
payload: {
threads: updatedThreads,
},
});
const processedMessageStore = processMessageStoreOperations( const processedMessageStore = processMessageStoreOperations(
updatedMessageStore, updatedMessageStore,
newMessageOps, newMessageOps,
); );
const messageStore = {
messages: processedMessageStore.messages,
threads,
tomekUnsubmitted
Not Done
Inline Actions
messages: processedMessageStore.messages,
- threads,
+ threads: processedMessageStore.threads,
local,

Should we also take threads prop of processedMessageStore?

tomek: Should we also take `threads` prop of `processedMessageStore`?
kamilAuthorUnsubmitted
Done
Inline Actions

Lets assume that:
A - old state
B - state modified in reducer, the current logic
C - state modified by ops

previously it worked like this:

  1. We have an old state (l A)
  2. We modify the state (A -> B)
  3. We return the modified state (B)

this diff this changes to:

  1. We have an old state (A)
  2. We modify the state and generate ops (A ->B)
  3. We take the old state, apply ops, and check if this matches the modified state (A -> C, and then checking B === C)
  4. We return modified state (B)

In future, if devs will not encounter any issues:

  1. We have old state (A)
  2. We will remove modifying the state and only generate ops (A not changed)
  3. Apply ops to old state (A -> C)
  4. Return state modified by ops (C)
kamil: Lets assume that: A - old state B - state modified in reducer, the current logic C - state…
tomekUnsubmitted
Not Done
Inline Actions

Thanks for the explanation - it makes sense!

tomek: Thanks for the explanation - it makes sense!
local,
currentAsOf,
};
assertMessageStoreThreadsAreEqual(
processedMessageStore,
messageStore,
`${actionType} | processed`,
);
return { return {
messageStoreOperations: [ messageStoreOperations: [
...updateWithLatestThreadInfosOps, ...updateWithLatestThreadInfosOps,
...newMessageOps, ...newMessageOps,
], ],
messageStore: { messageStore,
messages: processedMessageStore.messages,
threads,
local,
currentAsOf,
},
}; };
} }
type UpdateMessageStoreWithLatestThreadInfosResult = { type UpdateMessageStoreWithLatestThreadInfosResult = {
+messageStoreOperations: MessageStoreOperation[], +messageStoreOperations: MessageStoreOperation[],
+messageStore: MessageStore, +messageStore: MessageStore,
+reassignedThreadIDs: string[], +reassignedThreadIDs: string[],
}; };
Show All 24 Lines if (watchedThreadInfos[threadID]) {
continue; continue;
} }
threadsToRemoveMessagesFrom.push(threadID); threadsToRemoveMessagesFrom.push(threadID);
for (const id of reassignedMessageStore.threads[threadID].messageIDs) { for (const id of reassignedMessageStore.threads[threadID].messageIDs) {
messageIDsToRemove.push(id); messageIDsToRemove.push(id);
} }
} }
const updatedThreads = {};
for (const threadID in threadInfos) { for (const threadID in threadInfos) {
const threadInfo = threadInfos[threadID]; const threadInfo = threadInfos[threadID];
if ( if (
isThreadWatched(threadID, threadInfo, watchedIDs) && isThreadWatched(threadID, threadInfo, watchedIDs) &&
!filteredThreads[threadID] !filteredThreads[threadID]
) { ) {
filteredThreads[threadID] = newThread(); filteredThreads[threadID] = newThread();
updatedThreads[threadID] = filteredThreads[threadID];
} }
} }
messageStoreOperations.push({ messageStoreOperations.push({
type: 'remove_threads',
payload: { ids: threadsToRemoveMessagesFrom },
});
tomekUnsubmitted
Not Done
Inline Actions

Why do we remove these threads?

tomek: Why do we remove these threads?
kamilAuthorUnsubmitted
Done
Inline Actions

in this function as threads we return filteredThreads which are threads from the message store whose ids are "watched".

  1. We can either take the diff of reassignedMessageStore.threads and filteredThreads and remove those
  2. Use threadsToRemoveMessagesFrom which is created in a loop that iterates through all reassignedMessageStore.threads (see line 667) and select those which are not present in watchedThreadInfos (current solution and better in terms of perf I think)
kamil: in this function as `threads` we return `filteredThreads` which are threads from the message…
messageStoreOperations.push({
type: 'replace_threads',
payload: {
threads: updatedThreads,
},
});
messageStoreOperations.push({
type: 'remove_messages_for_threads', type: 'remove_messages_for_threads',
payload: { threadIDs: threadsToRemoveMessagesFrom }, payload: { threadIDs: threadsToRemoveMessagesFrom },
}); });
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
messages: _omit(messageIDsToRemove)(reassignedMessageStore.messages), messages: _omit(messageIDsToRemove)(reassignedMessageStore.messages),
▲ Show 20 Lines • Show All 113 Lines • ▼ Show 20 Lines const messagesResult = mergeUpdatesWithMessageInfos(
action.payload.messagesResult.truncationStatuses, action.payload.messagesResult.truncationStatuses,
); );
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === processUpdatesActionType) { } else if (action.type === processUpdatesActionType) {
if (action.payload.updatesResult.newUpdates.length === 0) { if (action.payload.updatesResult.newUpdates.length === 0) {
return { messageStoreOperations: [], messageStore }; return { messageStoreOperations: [], messageStore };
} }
const messagesResult = mergeUpdatesWithMessageInfos( const messagesResult = mergeUpdatesWithMessageInfos(
[], [],
action.payload.updatesResult.newUpdates, action.payload.updatesResult.newUpdates,
); );
const { messageStoreOperations, messageStore: newMessageStore } = const { messageStoreOperations, messageStore: newMessageStore } =
mergeNewMessages( mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
messages: newMessageStore.messages, messages: newMessageStore.messages,
threads: newMessageStore.threads, threads: newMessageStore.threads,
local: newMessageStore.local, local: newMessageStore.local,
currentAsOf: messageStore.currentAsOf, currentAsOf: messageStore.currentAsOf,
}, },
}; };
} else if ( } else if (
action.type === fullStateSyncActionType || action.type === fullStateSyncActionType ||
action.type === processMessagesActionType action.type === processMessagesActionType
) { ) {
const { messagesResult } = action.payload; const { messagesResult } = action.payload;
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === fetchSingleMostRecentMessagesFromThreadsActionTypes.success action.type === fetchSingleMostRecentMessagesFromThreadsActionTypes.success
) { ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
action.payload.truncationStatuses, action.payload.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === fetchMessagesBeforeCursorActionTypes.success || action.type === fetchMessagesBeforeCursorActionTypes.success ||
action.type === fetchMostRecentMessagesActionTypes.success action.type === fetchMostRecentMessagesActionTypes.success
) { ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
{ [action.payload.threadID]: action.payload.truncationStatus }, { [action.payload.threadID]: action.payload.truncationStatus },
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === logOutActionTypes.success || action.type === logOutActionTypes.success ||
action.type === deleteAccountActionTypes.success || action.type === deleteAccountActionTypes.success ||
action.type === deleteThreadActionTypes.success || action.type === deleteThreadActionTypes.success ||
action.type === leaveThreadActionTypes.success || action.type === leaveThreadActionTypes.success ||
action.type === setNewSessionActionType action.type === setNewSessionActionType
) { ) {
Show All 17 Lines const messagesResult = mergeUpdatesWithMessageInfos(
action.payload.newMessageInfos, action.payload.newMessageInfos,
action.payload.updatesResult.newUpdates, action.payload.updatesResult.newUpdates,
); );
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === sendMessageReportActionTypes.success) { } else if (action.type === sendMessageReportActionTypes.success) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
[action.payload.messageInfo], [action.payload.messageInfo],
{ {
[action.payload.messageInfo.threadID]: [action.payload.messageInfo.threadID]:
messageTruncationStatus.UNCHANGED, messageTruncationStatus.UNCHANGED,
}, },
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === registerActionTypes.success) { } else if (action.type === registerActionTypes.success) {
const truncationStatuses = {}; const truncationStatuses = {};
for (const messageInfo of action.payload.rawMessageInfos) { for (const messageInfo of action.payload.rawMessageInfos) {
truncationStatuses[messageInfo.threadID] = truncationStatuses[messageInfo.threadID] =
messageTruncationStatus.EXHAUSTIVE; messageTruncationStatus.EXHAUSTIVE;
} }
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
truncationStatuses, truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === changeThreadSettingsActionTypes.success || action.type === changeThreadSettingsActionTypes.success ||
action.type === removeUsersFromThreadActionTypes.success || action.type === removeUsersFromThreadActionTypes.success ||
action.type === changeThreadMemberRolesActionTypes.success || action.type === changeThreadMemberRolesActionTypes.success ||
action.type === createEntryActionTypes.success || action.type === createEntryActionTypes.success ||
action.type === saveEntryActionTypes.success || action.type === saveEntryActionTypes.success ||
action.type === restoreEntryActionTypes.success || action.type === restoreEntryActionTypes.success ||
action.type === toggleMessagePinActionTypes.success action.type === toggleMessagePinActionTypes.success
) { ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.newMessageInfos, action.payload.newMessageInfos,
{ [action.payload.threadID]: messageTruncationStatus.UNCHANGED }, { [action.payload.threadID]: messageTruncationStatus.UNCHANGED },
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === deleteEntryActionTypes.success) { } else if (action.type === deleteEntryActionTypes.success) {
const payload = action.payload; const payload = action.payload;
if (payload) { if (payload) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
payload.newMessageInfos, payload.newMessageInfos,
{ [payload.threadID]: messageTruncationStatus.UNCHANGED }, { [payload.threadID]: messageTruncationStatus.UNCHANGED },
newThreadInfos, newThreadInfos,
action.type,
); );
} }
} else if (action.type === joinThreadActionTypes.success) { } else if (action.type === joinThreadActionTypes.success) {
const messagesResult = mergeUpdatesWithMessageInfos( const messagesResult = mergeUpdatesWithMessageInfos(
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
action.payload.updatesResult.newUpdates, action.payload.updatesResult.newUpdates,
); );
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if (action.type === sendEditMessageActionTypes.success) { } else if (action.type === sendEditMessageActionTypes.success) {
const { newMessageInfos } = action.payload; const { newMessageInfos } = action.payload;
const truncationStatuses = {}; const truncationStatuses = {};
for (const messageInfo of newMessageInfos) { for (const messageInfo of newMessageInfos) {
truncationStatuses[messageInfo.threadID] = truncationStatuses[messageInfo.threadID] =
messageTruncationStatus.UNCHANGED; messageTruncationStatus.UNCHANGED;
} }
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
newMessageInfos, newMessageInfos,
truncationStatuses, truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
} else if ( } else if (
action.type === sendTextMessageActionTypes.started || action.type === sendTextMessageActionTypes.started ||
action.type === sendMultimediaMessageActionTypes.started || action.type === sendMultimediaMessageActionTypes.started ||
action.type === sendReactionMessageActionTypes.started action.type === sendReactionMessageActionTypes.started
) { ) {
const payload = ensureRealizedThreadIDIsUsedWhenPossible( const payload = ensureRealizedThreadIDIsUsedWhenPossible(
action.payload, action.payload,
▲ Show 20 Lines • Show All 207 Lines • ▼ Show 20 Lines for (const messageInfo of action.payload.rawMessageInfos) {
messageTruncationStatus.UNCHANGED; messageTruncationStatus.UNCHANGED;
} }
const { messageStoreOperations, messageStore: newMessageStore } = const { messageStoreOperations, messageStore: newMessageStore } =
mergeNewMessages( mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
truncationStatuses, truncationStatuses,
newThreadInfos, newThreadInfos,
action.type,
); );
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
messages: newMessageStore.messages, messages: newMessageStore.messages,
threads: newMessageStore.threads, threads: newMessageStore.threads,
local: newMessageStore.local, local: newMessageStore.local,
// We avoid bumping currentAsOf because notifs may include a contracted // We avoid bumping currentAsOf because notifs may include a contracted
▲ Show 20 Lines • Show All 257 Lines • ▼ Show 20 Lines const {
messageStore: messageStoreAfterReassignment, messageStore: messageStoreAfterReassignment,
} = reassignMessagesToRealizedThreads(messageStore, newThreadInfos); } = reassignMessagesToRealizedThreads(messageStore, newThreadInfos);
const processedMessageStore = processMessageStoreOperations( const processedMessageStore = processMessageStoreOperations(
messageStore, messageStore,
messageStoreOperations, messageStoreOperations,
); );
assertMessageStoreThreadsAreEqual(
processedMessageStore,
messageStoreAfterReassignment,
action.type,
);
return { return {
messageStoreOperations, messageStoreOperations,
messageStore: { messageStore: {
...messageStoreAfterReassignment, ...messageStoreAfterReassignment,
messages: processedMessageStore.messages, messages: processedMessageStore.messages,
}, },
}; };
} else if (action.type === setClientDBStoreActionType) { } else if (action.type === setClientDBStoreActionType) {
▲ Show 20 Lines • Show All 128 Lines • Show Last 20 Lines