fix: send-message pending task race condition (#3538)

## 🎯 Goal

This PR resolves a very long standing race condition that has evaded me
for quite a while.

Whenever we queue up send message requests while we're offline, they are
to be executed when we regain connection. However, the UI SDK takes care
of failed messages in the sense of re-adding them in case they weren't
resolved or saved in the LLC. This is done so that when a resync happens
of the channel, the state is preserved and the failed messages do not
disappear.

However, since our state layer is not LLC only fully, we have 2 vectors
of resyncing the SDK:

- One that comes from the LLC (i.e `channel.watch()` being called)
- One that comes from the SDK itself (re-adding those failed messages)

When `send-message` pending tasks are executed, the LLC takes care of
upserting the state immediately so that the UI SDK can simply consume
it. This is fine, except for the fact that for the UI SDK to also
resolve its own state, we rely on WS events (which might be a bit late,
especially for the last 1-2 messages, which is when this bug would
mostly happen).

So the flow would look something like this:

- We regain connection
- Pending tasks are executed, `channel.state.messages` are updated (from
the pending task execution)
- After this, the `onSyncStatusChange` callback is invoked
- The `channel` is actually resynced
- `channel.watch()` is called
- We also upsert all failed messages once again
- For any `message` for which we did not receive a WS event of the
successful pending task execution on time, both the failed `message` and
the actual (serverside) one appear in the list

To fix this, we deem messages eligible for recovery if they:

- are `FAILED`
- are not already present in `channel.state` 

This makes sure that the race condition never happens when the state is
already up to date. If it's not, we anyway have to do it.

## 🛠 Implementation details

<!-- Provide a description of the implementation -->

## 🎨 UI Changes

<!-- Add relevant screenshots -->

<details>
<summary>iOS</summary>


<table>
    <thead>
        <tr>
            <td>Before</td>
            <td>After</td>
        </tr>
    </thead>
    <tbody>
        <tr>
            <td>
                <!--<img src="" /> -->
            </td>
            <td>
                <!--<img src="" /> -->
            </td>
        </tr>
    </tbody>
</table>
</details>


<details>
<summary>Android</summary>

<table>
    <thead>
        <tr>
            <td>Before</td>
            <td>After</td>
        </tr>
    </thead>
    <tbody>
        <tr>
            <td>
                <!--<img src="" /> -->
            </td>
            <td>
                <!--<img src="" /> -->
            </td>
        </tr>
    </tbody>
</table>
</details>

## 🧪 Testing

<!-- Explain how this change can be tested (or why it can't be tested)
-->

## ☑️ Checklist

- [ ] I have signed the [Stream
CLA](https://docs.google.com/forms/d/e/1FAIpQLScFKsKkAJI7mhCr7K9rEIOpqIDThrWxuvxnwUq2XkHyG154vQ/viewform)
(required)
- [ ] PR targets the `develop` branch
- [ ] Documentation is updated
- [ ] New code is tested in main example apps, including all possible
scenarios
  - [ ] SampleApp iOS and Android
  - [ ] Expo iOS and Android

Author: isekovanic

package/src/__tests__/offline-support/optimistic-update.js

@@ -720,6 +720,131 @@ export const OptimisticUpdates = () => {
           expect(sendMessageSpy).toHaveBeenCalled();
         });
       });
+
+      it('should not re-add a failed local message after reconnect when its pending send task was resolved', async () => {
+        const localMessage = generateMessage({
+          status: MessageStatusTypes.SENDING,
+          text: 'offline resend',
+          user: chatClient.user,
+          userId: chatClient.userID,
+        });
+        const serverMessage = generateMessage({
+          id: localMessage.id,
+          text: localMessage.text,
+          user: chatClient.user,
+          userId: chatClient.userID,
+        });
+
+        jest.spyOn(channel.messageComposer, 'compose').mockResolvedValue({
+          localMessage,
+          message: localMessage,
+          options: {},
+        });
+
+        render(
+          <Chat client={chatClient} enableOfflineSupport>
+            <Channel channel={channel} initialValue={localMessage.text}>
+              <CallbackEffectWithContext
+                callback={async ({ sendMessage }) => {
+                  useMockedApis(chatClient, [erroredPostApi()]);
+                  await sendMessage();
+                }}
+                context={MessageInputContext}
+              >
+                <View testID='children' />
+              </CallbackEffectWithContext>
+            </Channel>
+          </Chat>,
+        );
+        await waitFor(() => expect(screen.getByTestId('children')).toBeTruthy());
+
+        let pendingTask;
+        await waitFor(async () => {
+          const pendingTasks = await chatClient.offlineDb.getPendingTasks();
+          expect(pendingTasks).toHaveLength(1);
+          pendingTask = pendingTasks[0];
+        });
+
+        expect(channel.state.messages.some((message) => message.id === localMessage.id)).toBe(true);
+
+        jest.spyOn(channel, 'watch').mockResolvedValue({});
+
+        channel.state.removeMessage(localMessage);
+        channel.state.addMessageSorted(serverMessage, true);
+        await chatClient.offlineDb.deletePendingTask({ id: pendingTask.id });
+
+        await act(async () => {
+          await chatClient.offlineDb.syncManager.invokeSyncStatusListeners(true);
+        });
+
+        await waitFor(() => {
+          const matchingMessages = channel.state.messages.filter(
+            (message) => message.text === localMessage.text,
+          );
+
+          expect(matchingMessages).toHaveLength(1);
+          expect(matchingMessages[0].id).toBe(serverMessage.id);
+          expect(matchingMessages[0].status).not.toBe(MessageStatusTypes.FAILED);
+        });
+      });
+
+      it('should re-add a failed local message after reconnect when fresh state still does not contain it', async () => {
+        const localMessage = generateMessage({
+          status: MessageStatusTypes.SENDING,
+          text: 'offline resend unresolved',
+          user: chatClient.user,
+          userId: chatClient.userID,
+        });
+
+        jest.spyOn(channel.messageComposer, 'compose').mockResolvedValue({
+          localMessage,
+          message: localMessage,
+          options: {},
+        });
+
+        render(
+          <Chat client={chatClient} enableOfflineSupport>
+            <Channel channel={channel} initialValue={localMessage.text}>
+              <CallbackEffectWithContext
+                callback={async ({ sendMessage }) => {
+                  useMockedApis(chatClient, [erroredPostApi()]);
+                  await sendMessage();
+                }}
+                context={MessageInputContext}
+              >
+                <View testID='children' />
+              </CallbackEffectWithContext>
+            </Channel>
+          </Chat>,
+        );
+        await waitFor(() => expect(screen.getByTestId('children')).toBeTruthy());
+
+        let pendingTask;
+        await waitFor(async () => {
+          const pendingTasks = await chatClient.offlineDb.getPendingTasks();
+          expect(pendingTasks).toHaveLength(1);
+          pendingTask = pendingTasks[0];
+        });
+
+        jest.spyOn(channel, 'watch').mockResolvedValue({});
+
+        channel.state.removeMessage(localMessage);
+        await chatClient.offlineDb.deletePendingTask({ id: pendingTask.id });
+
+        await act(async () => {
+          await chatClient.offlineDb.syncManager.invokeSyncStatusListeners(true);
+        });
+
+        await waitFor(() => {
+          const matchingMessages = channel.state.messages.filter(
+            (message) => message.id === localMessage.id,
+          );
+
+          expect(matchingMessages).toHaveLength(1);
+          expect(matchingMessages[0].status).toBe(MessageStatusTypes.FAILED);
+          expect(matchingMessages[0].text).toBe(localMessage.text);
+        });
+      });
     });
   });
 };

package/src/components/Channel/Channel.tsx

@@ -1167,6 +1167,15 @@ const ChannelWithContext = (props: PropsWithChildren<ChannelPropsWithContext>) =
         updated_at: message.updated_at?.toString(),
       }) as unknown as MessageResponse;
 
+    const getRecoverableFailedMessages = (messages: LocalMessage[] = []) =>
+      messages
+        .filter(
+          (message) =>
+            message.status === MessageStatusTypes.FAILED &&
+            !channel.state.findMessage(message.id, message.parent_id),
+        )
+        .map(parseMessage);
+
     try {
       if (channelMessagesState?.messages) {
         await channel?.watch({
@@ -1181,9 +1190,7 @@ const ChannelWithContext = (props: PropsWithChildren<ChannelPropsWithContext>) =
       if (!thread) {
         copyChannelState();
 
-        const failedMessages = channelMessagesState.messages
-          ?.filter((message) => message.status === MessageStatusTypes.FAILED)
-          .map(parseMessage);
+        const failedMessages = getRecoverableFailedMessages(channelMessagesState.messages);
         if (failedMessages?.length) {
           channel.state.addMessagesSorted(failedMessages);
         }
@@ -1192,11 +1199,7 @@ const ChannelWithContext = (props: PropsWithChildren<ChannelPropsWithContext>) =
       } else {
         await reloadThread();
 
-        const failedThreadMessages = thread
-          ? threadMessages
-              .filter((message) => message.status === MessageStatusTypes.FAILED)
-              .map(parseMessage)
-          : [];
+        const failedThreadMessages = thread ? getRecoverableFailedMessages(threadMessages) : [];
         if (failedThreadMessages.length) {
           channel.state.addMessagesSorted(failedThreadMessages);
           setThreadMessages([...channel.state.threads[thread.id]]);