fix(llc): mark a message the server already stored as sent

packages/stream_chat/CHANGELOG.md

@@ -13,6 +13,7 @@
 
 - Fixed `Event.aiState` reading `AI_STATE_EXTERNAL_SOURCES` as `idle` instead of `checkingSources`.
 - Fixed `Channel.disableSlowMode` and `StreamChatClient.disableSlowdown` always failing.
+- Fixed a message the server had already stored being shown as failed after a lost response or a retried send.
 
 ## 10.5.0
 

packages/stream_chat/lib/src/client/channel/channel.dart

@@ -611,10 +611,7 @@ class Channel {
     String messageId,
     Iterable<String> attachmentIds,
   ) {
-    var message = [
-      ...state!.messages,
-      ...state!.threads.values.expand((messages) => messages),
-    ].firstWhereOrNull((it) => it.id == messageId);
+    var message = _findMessageInLocalState(messageId);
 
     if (message == null) {
       throw const StreamChatError('Error, Message not found');
@@ -745,6 +742,11 @@ class Channel {
     });
   }
 
+  Message? _findMessageInLocalState(String messageId) {
+    final threadMessages = state!.threads.values.expand((messages) => messages);
+    return state!.messages.followedBy(threadMessages).firstWhereOrNull((it) => it.id == messageId);
+  }
+
   final _sendMessageLock = Lock();
 
   /// Send a [message] to this channel.
@@ -837,6 +839,9 @@ class Channel {
 
       return response;
     } catch (e) {
+      if (e is StreamChatNetworkError && e.isMessageAlreadyExistsError) {
+        return SendMessageResponse()..message = _markMessageAsSent(message);
+      }
       final failedMessage = message.copyWith(
         // Update the message state to failed.
         state: MessageState.sendingFailed(
@@ -855,6 +860,19 @@ class Channel {
     }
   }
 
+  Message _markMessageAsSent(Message message) {
+    if (state == null) return message.copyWith(state: MessageState.sent);
+
+    final localStateMessage = _findMessageInLocalState(message.id) ?? message;
+    final sentMessage = localStateMessage.copyWith(
+      createdAt: localStateMessage.remoteCreatedAt ?? localStateMessage.localCreatedAt,
+      state: MessageState.sent,
+    );
+
+    state!.updateMessage(sentMessage);
+    return sentMessage;
+  }
+
   final _updateMessageLock = Lock();
 
   /// Updates the [message] in this channel.

packages/stream_chat/lib/src/core/error/stream_chat_error.dart

@@ -1,6 +1,7 @@
 import 'dart:convert';
 
 import 'package:equatable/equatable.dart';
+import 'package:meta/meta.dart';
 import 'package:web_socket_channel/web_socket_channel.dart';
 
 import '../../../stream_chat.dart';
@@ -163,6 +164,15 @@ class StreamChatNetworkError extends StreamChatError {
   /// Whether the operation can be retried.
   bool get isRetriable => data == null;
 
+  /// Whether the server rejected a send because it already stored a message
+  /// with the same id.
+  @internal
+  bool get isMessageAlreadyExistsError {
+    if (errorCode != ChatErrorCode.inputError) return false;
+    if (!message.contains('a message with ID')) return false;
+    return message.contains('already exists');
+  }
+
   @override
   List<Object?> get props => [...super.props, code, statusCode, type];
 

packages/stream_chat/test/src/client/channel/channel_send_test.dart

@@ -0,0 +1,161 @@
+import 'package:mocktail/mocktail.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:test/test.dart';
+
+import '../../fakes.dart';
+import '../../matchers.dart';
+import '../../mocks.dart';
+
+void main() {
+  setUpAll(() {
+    registerFallbackValue(FakeMessage());
+    registerFallbackValue(FakeEvent());
+  });
+
+  group('when the message id already exists', () {
+    test('marks the message as sent', () async {
+      final client = MockStreamChatClient();
+      final channel = _initializedChannel(client);
+      final message = Message(id: 'stored-message-id', text: 'Hello world!');
+
+      _whenSendingMessage(client, channel, message).thenThrow(_messageAlreadyExistsError(message.id));
+      await channel.sendMessage(message);
+
+      expect(channel.state!.messages.single.state, MessageState.sent);
+    });
+
+    test('returns the message marked as sent', () async {
+      final client = MockStreamChatClient();
+      final channel = _initializedChannel(client);
+      final message = Message(id: 'stored-message-id', text: 'Hello world!');
+
+      _whenSendingMessage(client, channel, message).thenThrow(_messageAlreadyExistsError(message.id));
+      final response = await channel.sendMessage(message);
+
+      expect(response.message.state, MessageState.sent);
+    });
+
+    test('stamps the local creation date as the server creation date when no server copy arrived', () async {
+      final client = MockStreamChatClient();
+      final channel = _initializedChannel(client);
+      final message = Message(id: 'stored-message-id', text: 'Hello world!');
+
+      _whenSendingMessage(client, channel, message).thenThrow(_messageAlreadyExistsError(message.id));
+      await channel.sendMessage(message);
+
+      final sentMessage = channel.state!.messages.single;
+      expect(sentMessage.remoteCreatedAt, sentMessage.localCreatedAt);
+    });
+
+    test('keeps the original creation date when a failed message is resent', () async {
+      final client = MockStreamChatClient();
+      final channel = _initializedChannel(client);
+      final originalCreatedAt = DateTime.now().subtract(const Duration(hours: 1));
+      final failedMessage = Message(
+        id: 'stored-message-id',
+        text: 'Hello world!',
+        localCreatedAt: originalCreatedAt,
+        state: MessageState.sendingFailed(skipPush: false, skipEnrichUrl: false),
+      );
+      channel.state!.updateMessage(failedMessage);
+
+      _whenSendingMessage(client, channel, failedMessage).thenThrow(_messageAlreadyExistsError(failedMessage.id));
+      await channel.retryMessage(failedMessage);
+
+      expect(channel.state!.messages.single.createdAt, originalCreatedAt);
+    });
+
+    test('keeps the server creation date when message.new arrived first', () async {
+      final client = MockStreamChatClient();
+      final channel = _initializedChannel(client);
+      final message = Message(id: 'stored-message-id', text: 'Hello world!');
+      final serverCreatedAt = DateTime.utc(2026, 10, 8, 7, 47, 56);
+
+      _whenSendingMessage(client, channel, message).thenAnswer((_) async {
+        // The server's copy arrives over the WebSocket before the send fails.
+        client.addEvent(
+          Event(
+            cid: channel.cid,
+            type: EventType.messageNew,
+            message: message.copyWith(createdAt: serverCreatedAt),
+          ),
+        );
+        await pumpEventQueue();
+        throw _messageAlreadyExistsError(message.id);
+      });
+      await channel.sendMessage(message);
+
+      expect(channel.state!.messages.single.remoteCreatedAt, serverCreatedAt);
+    });
+
+    test('marks a timed-out send as sent once its retry is rejected as a duplicate', () async {
+      final client = MockStreamChatClient();
+      final channel = _initializedChannel(client);
+      final message = Message(id: 'stored-message-id', text: 'Hello world!');
+
+      // The first attempt times out, so the retry queue re-sends the message.
+      final receiveTimeout = StreamChatNetworkError.raw(
+        code: -1,
+        message: 'The request took longer than the receive timeout',
+        type: StreamChatNetworkErrorType.receiveTimeout,
+      );
+      final errors = [receiveTimeout, _messageAlreadyExistsError(message.id)];
+      var attempt = 0;
+      _whenSendingMessage(client, channel, message).thenAnswer((_) async {
+        final error = errors[attempt];
+        attempt += 1;
+        throw error;
+      });
+      await expectLater(channel.sendMessage(message), throwsA(isA<StreamChatNetworkError>()));
+      await pumpEventQueue();
+
+      expect(channel.state!.messages.single.state, MessageState.sent);
+    });
+  });
+}
+
+Channel _initializedChannel(MockStreamChatClient client) {
+  when(() => client.detachedLogger(any())).thenAnswer((it) => Logger.detached(it.positionalArguments.first as String));
+  when(() => client.logger).thenReturn(Logger.detached('test-client'));
+  when(() => client.state).thenReturn(FakeClientState());
+  when(() => client.retryPolicy).thenReturn(
+    RetryPolicy(shouldRetry: (_, __, ___) => false, delayFactor: Duration.zero),
+  );
+  when(() => client.channelDeliveryReporter.submitForDelivery(any())).thenAnswer((_) async {});
+  addTearDown(client.dispose);
+
+  final channel = Channel.fromState(
+    client,
+    ChannelState(
+      channel: ChannelModel(id: 'test-channel-id', type: 'test-channel-type'),
+    ),
+  );
+  addTearDown(channel.dispose);
+
+  return channel;
+}
+
+When<Future<SendMessageResponse>> _whenSendingMessage(
+  StreamChatClient client,
+  Channel channel,
+  Message message,
+) {
+  return when(() => client.sendMessage(any(that: isSameMessageAs(message)), channel.id!, channel.type));
+}
+
+StreamChatNetworkError _messageAlreadyExistsError(String messageId) {
+  return _inputError('SendMessage failed with error: "a message with ID $messageId already exists"');
+}
+
+StreamChatNetworkError _inputError(String message) {
+  return StreamChatNetworkError.raw(
+    code: ChatErrorCode.inputError.code,
+    message: message,
+    statusCode: 400,
+    data: ErrorResponse()
+      ..code = ChatErrorCode.inputError.code
+      ..message = message
+      ..statusCode = 400,
+    type: StreamChatNetworkErrorType.badResponse,
+  );
+}

packages/stream_chat/test/src/core/error/stream_chat_error_test.dart

@@ -215,6 +215,51 @@ void main() {
       expect(error.isRetriable, isTrue);
     });
 
+    test('`.isMessageAlreadyExistsError` should return true for a duplicate message id rejection', () {
+      const message = 'SendMessage failed with error: "a message with ID abc already exists"';
+      final error = StreamChatNetworkError.raw(
+        code: ChatErrorCode.inputError.code,
+        statusCode: 400,
+        message: message,
+        data: ErrorResponse()
+          ..code = ChatErrorCode.inputError.code
+          ..message = message
+          ..statusCode = 400,
+      );
+
+      expect(error.isMessageAlreadyExistsError, isTrue);
+    });
+
+    test('`.isMessageAlreadyExistsError` should return false for a poll option that already exists', () {
+      const message = 'poll option with text A already exists';
+      final error = StreamChatNetworkError.raw(
+        code: ChatErrorCode.inputError.code,
+        statusCode: 400,
+        message: message,
+        data: ErrorResponse()
+          ..code = ChatErrorCode.inputError.code
+          ..message = message
+          ..statusCode = 400,
+      );
+
+      expect(error.isMessageAlreadyExistsError, isFalse);
+    });
+
+    test('`.isMessageAlreadyExistsError` should return false for the same text with another error code', () {
+      const message = 'SendMessage failed with error: "a message with ID abc already exists"';
+      final error = StreamChatNetworkError.raw(
+        code: ChatErrorCode.notAllowed.code,
+        statusCode: 403,
+        message: message,
+        data: ErrorResponse()
+          ..code = ChatErrorCode.notAllowed.code
+          ..message = message
+          ..statusCode = 403,
+      );
+
+      expect(error.isMessageAlreadyExistsError, isFalse);
+    });
+
     test('`.toString`', () {
       const errorCode = ChatErrorCode.tokenExpired;
       final error = StreamChatNetworkError(errorCode);