fix: send-message pending task race condition

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]]);