From 097ed43e79e5f0596de7147d20e0d64b2a3d08a1 Mon Sep 17 00:00:00 2001 From: VelikovPetar Date: Thu, 8 Oct 2026 11:31:22 +0200 Subject: [PATCH 1/2] fix(llc): mark a message the server already stored as sent Co-Authored-By: Claude Opus 5.5 --- packages/stream_chat/CHANGELOG.md | 1 + .../lib/src/client/channel/channel.dart | 24 +- .../lib/src/client/retry_queue.dart | 7 +- .../lib/src/core/error/stream_chat_error.dart | 9 + .../test/src/client/channel/channel_test.dart | 255 ++++++++++-------- .../core/error/stream_chat_error_test.dart | 30 +++ 6 files changed, 209 insertions(+), 117 deletions(-) diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 86113d9591..9f1e7ee0c8 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/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 diff --git a/packages/stream_chat/lib/src/client/channel/channel.dart b/packages/stream_chat/lib/src/client/channel/channel.dart index 8a0b26cc80..dd245b88a1 100644 --- a/packages/stream_chat/lib/src/client/channel/channel.dart +++ b/packages/stream_chat/lib/src/client/channel/channel.dart @@ -611,10 +611,7 @@ class Channel { String messageId, Iterable 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) => [ + ...state!.messages, + ...state!.threads.values.expand((messages) => messages), + ].firstWhereOrNull((it) => it.id == messageId); + final _sendMessageLock = Lock(); /// Send a [message] to this channel. @@ -837,6 +839,10 @@ class Channel { return response; } catch (e) { + if (e is StreamChatNetworkError && e.isMessageAlreadyExistsError) { + _markMessageAsSent(message); + rethrow; + } final failedMessage = message.copyWith( // Update the message state to failed. state: MessageState.sendingFailed( @@ -855,6 +861,16 @@ class Channel { } } + void _markMessageAsSent(Message message) { + if (state == null) return; + final localStateMessage = _findMessageInLocalState(message.id) ?? message; + final sentMessage = localStateMessage.copyWith( + createdAt: localStateMessage.remoteCreatedAt ?? localStateMessage.localCreatedAt, + state: MessageState.sent, + ); + state?.updateMessage(sentMessage); + } + final _updateMessageLock = Lock(); /// Updates the [message] in this channel. diff --git a/packages/stream_chat/lib/src/client/retry_queue.dart b/packages/stream_chat/lib/src/client/retry_queue.dart index 8b05181f3c..d65c450668 100644 --- a/packages/stream_chat/lib/src/client/retry_queue.dart +++ b/packages/stream_chat/lib/src/client/retry_queue.dart @@ -89,8 +89,9 @@ class RetryQueue { } catch (error) { logger?.severe('Error while retrying message ${message.id}', error); // If we are unable to successfully retry the message, update the state - // with the failed state. - channel.state?.updateMessage(message); + // with the failed state, unless an earlier attempt was stored by the + // server and the message is already marked as sent. + if (!_isAlreadySent(error)) channel.state?.updateMessage(message); } finally { // remove the message from the queue after it's handled. _messageQueue.removeMessage(message); @@ -100,6 +101,8 @@ class RetryQueue { _isProcessing = false; } + static bool _isAlreadySent(Object error) => error is StreamChatNetworkError && error.isMessageAlreadyExistsError; + /// Whether our [_messageQueue] has messages or not. bool get hasMessages => _messageQueue.isNotEmpty; diff --git a/packages/stream_chat/lib/src/core/error/stream_chat_error.dart b/packages/stream_chat/lib/src/core/error/stream_chat_error.dart index 4f5c6ff257..9de04f1e94 100644 --- a/packages/stream_chat/lib/src/core/error/stream_chat_error.dart +++ b/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,14 @@ 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 => + errorCode == ChatErrorCode.inputError && + message.contains('a message with ID') && + message.contains('already exists'); + @override List get props => [...super.props, code, statusCode, type]; diff --git a/packages/stream_chat/test/src/client/channel/channel_test.dart b/packages/stream_chat/test/src/client/channel/channel_test.dart index 21aa70487c..c230e6e499 100644 --- a/packages/stream_chat/test/src/client/channel/channel_test.dart +++ b/packages/stream_chat/test/src/client/channel/channel_test.dart @@ -39,6 +39,42 @@ void main() { return logger; } + When> _whenSendingMessage( + StreamChatClient client, + Channel channel, + Message message, { + bool skipPush = false, + bool skipEnrichUrl = false, + }) { + return when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channel.id!, + channel.type, + skipPush: skipPush, + skipEnrichUrl: skipEnrichUrl, + ), + ); + } + + 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, + ); + } + + // The error returned when sending a message whose id already exists. + StreamChatNetworkError _messageAlreadyExistsError(String messageId) { + return _inputError('SendMessage failed with error: "a message with ID $messageId already exists"'); + } + group('Non-Initialized Channel', () { late final client = MockStreamChatClient(); const channelId = 'test-channel-id'; @@ -278,13 +314,7 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - ), - ).thenAnswer((_) async => sendMessageResponse); + _whenSendingMessage(client, channel, message).thenAnswer((_) async => sendMessageResponse); expectLater( // skipping first seed message list -> [] messages @@ -328,13 +358,11 @@ void main() { user: client.state.currentUser, ); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - skipPush: true, - ), + _whenSendingMessage( + client, + channel, + message, + skipPush: true, ).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); expectLater( @@ -382,14 +410,12 @@ void main() { user: client.state.currentUser, ); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - skipPush: true, - skipEnrichUrl: true, - ), + _whenSendingMessage( + client, + channel, + message, + skipPush: true, + skipEnrichUrl: true, ).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); expectLater( @@ -438,13 +464,11 @@ void main() { user: client.state.currentUser, ); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - skipEnrichUrl: true, - ), + _whenSendingMessage( + client, + channel, + message, + skipEnrichUrl: true, ).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); expectLater( @@ -492,13 +516,7 @@ void main() { user: client.state.currentUser, ); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - ), - ).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); + _whenSendingMessage(client, channel, message).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); expectLater( // skipping first seed message list -> [] messages @@ -542,22 +560,7 @@ void main() { user: client.state.currentUser, ); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - ), - ).thenThrow( - StreamChatNetworkError.raw( - code: ChatErrorCode.inputError.code, - message: 'Input error', - data: ErrorResponse() - ..code = ChatErrorCode.inputError.code - ..message = 'Input error' - ..statusCode = 400, - ), - ); + _whenSendingMessage(client, channel, message).thenThrow(_inputError('Input error')); expectLater( // skipping first seed message list -> [] messages @@ -589,6 +592,82 @@ void main() { ); }); + group('when the message id already exists', () { + test('marks the message as sent', () async { + final message = Message(id: 'stored-message-id', text: 'Hello world!'); + + _whenSendingMessage(client, channel, message).thenThrow(_messageAlreadyExistsError(message.id)); + await expectLater(channel.sendMessage(message), throwsA(isA())); + + expect(channel.state!.messages.single.state, MessageState.sent); + }); + + test('stamps the local creation date as the server creation date when no server copy arrived', () async { + final message = Message(id: 'stored-message-id', text: 'Hello world!'); + + _whenSendingMessage(client, channel, message).thenThrow(_messageAlreadyExistsError(message.id)); + await expectLater(channel.sendMessage(message), throwsA(isA())); + + 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 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 expectLater(channel.retryMessage(failedMessage), throwsA(isA())); + + expect(channel.state!.messages.single.createdAt, originalCreatedAt); + }); + + test('keeps the server creation date when message.new arrived first', () async { + 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 expectLater(channel.sendMessage(message), throwsA(isA())); + + 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 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 => throw errors[attempt++]); + await expectLater(channel.sendMessage(message), throwsA(isA())); + await pumpEventQueue(); + + expect(channel.state!.messages.single.state, MessageState.sent); + }); + }); + test('with attachments should work just fine', () async { final attachments = List.generate( 3, @@ -629,13 +708,7 @@ void main() { ), ).thenAnswer((_) async => sendFileResponse); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - ), - ).thenAnswer( + _whenSendingMessage(client, channel, message).thenAnswer( (_) async => SendMessageResponse() ..message = message.copyWith( attachments: attachments @@ -863,13 +936,7 @@ void main() { ), ); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - ), - ).thenAnswer( + _whenSendingMessage(client, channel, message).thenAnswer( (_) async => SendMessageResponse() ..message = message.copyWith( attachments: [], @@ -940,13 +1007,7 @@ void main() { ), ); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - ), - ).thenAnswer( + _whenSendingMessage(client, channel, message).thenAnswer( (_) async => SendMessageResponse() ..message = message.copyWith( attachments: [], @@ -1012,13 +1073,7 @@ void main() { ), ); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - ), - ).thenAnswer( + _whenSendingMessage(client, channel, message).thenAnswer( (_) async => SendMessageResponse() ..message = message.copyWith( attachments: [], @@ -6370,14 +6425,12 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - skipPush: true, - skipEnrichUrl: true, - ), + _whenSendingMessage( + client, + channel, + message, + skipPush: true, + skipEnrichUrl: true, ).thenAnswer((_) async => sendMessageResponse); final result = await channel.retryMessage(message); @@ -6408,14 +6461,7 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - skipPush: true, - ), - ).thenAnswer((_) async => sendMessageResponse); + _whenSendingMessage(client, channel, message, skipPush: true).thenAnswer((_) async => sendMessageResponse); final result = await channel.retryMessage(message); @@ -6444,14 +6490,7 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - skipEnrichUrl: true, - ), - ).thenAnswer((_) async => sendMessageResponse); + _whenSendingMessage(client, channel, message, skipEnrichUrl: true).thenAnswer((_) async => sendMessageResponse); final result = await channel.retryMessage(message); @@ -6480,13 +6519,7 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channelId, - channelType, - ), - ).thenAnswer((_) async => sendMessageResponse); + _whenSendingMessage(client, channel, message).thenAnswer((_) async => sendMessageResponse); final result = await channel.retryMessage(message); diff --git a/packages/stream_chat/test/src/core/error/stream_chat_error_test.dart b/packages/stream_chat/test/src/core/error/stream_chat_error_test.dart index 20f302b107..924d7ff09e 100644 --- a/packages/stream_chat/test/src/core/error/stream_chat_error_test.dart +++ b/packages/stream_chat/test/src/core/error/stream_chat_error_test.dart @@ -215,6 +215,36 @@ void main() { expect(error.isRetriable, isTrue); }); + test('`.isMessageAlreadyExistsError` should return true for a duplicate message id rejection', () { + final error = StreamChatNetworkError.raw( + code: ChatErrorCode.inputError.code, + statusCode: 400, + message: 'SendMessage failed with error: "a message with ID abc already exists"', + ); + + expect(error.isMessageAlreadyExistsError, isTrue); + }); + + test('`.isMessageAlreadyExistsError` should return false for a poll option that already exists', () { + final error = StreamChatNetworkError.raw( + code: ChatErrorCode.inputError.code, + statusCode: 400, + message: 'poll option with text A already exists', + ); + + expect(error.isMessageAlreadyExistsError, isFalse); + }); + + test('`.isMessageAlreadyExistsError` should return false for the same text with another error code', () { + final error = StreamChatNetworkError.raw( + code: ChatErrorCode.notAllowed.code, + statusCode: 403, + message: 'SendMessage failed with error: "a message with ID abc already exists"', + ); + + expect(error.isMessageAlreadyExistsError, isFalse); + }); + test('`.toString`', () { const errorCode = ChatErrorCode.tokenExpired; final error = StreamChatNetworkError(errorCode); From 0cdff122e44f617742f2c6bce533e9e843f3fecc Mon Sep 17 00:00:00 2001 From: VelikovPetar Date: Tue, 13 Oct 2026 09:33:05 +0200 Subject: [PATCH 2/2] fix(llc): return the sent message instead of rethrowing on a duplicate id Co-Authored-By: Claude Opus 5.5 --- .../lib/src/client/channel/channel.dart | 20 +- .../lib/src/client/retry_queue.dart | 7 +- .../lib/src/core/error/stream_chat_error.dart | 9 +- .../src/client/channel/channel_send_test.dart | 161 +++++++++++ .../test/src/client/channel/channel_test.dart | 255 ++++++++---------- .../core/error/stream_chat_error_test.dart | 21 +- 6 files changed, 308 insertions(+), 165 deletions(-) create mode 100644 packages/stream_chat/test/src/client/channel/channel_send_test.dart diff --git a/packages/stream_chat/lib/src/client/channel/channel.dart b/packages/stream_chat/lib/src/client/channel/channel.dart index dd245b88a1..375e8e4fb0 100644 --- a/packages/stream_chat/lib/src/client/channel/channel.dart +++ b/packages/stream_chat/lib/src/client/channel/channel.dart @@ -742,10 +742,10 @@ class Channel { }); } - Message? _findMessageInLocalState(String messageId) => [ - ...state!.messages, - ...state!.threads.values.expand((messages) => messages), - ].firstWhereOrNull((it) => it.id == messageId); + 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(); @@ -840,8 +840,7 @@ class Channel { return response; } catch (e) { if (e is StreamChatNetworkError && e.isMessageAlreadyExistsError) { - _markMessageAsSent(message); - rethrow; + return SendMessageResponse()..message = _markMessageAsSent(message); } final failedMessage = message.copyWith( // Update the message state to failed. @@ -861,14 +860,17 @@ class Channel { } } - void _markMessageAsSent(Message message) { - if (state == null) return; + 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); + + state!.updateMessage(sentMessage); + return sentMessage; } final _updateMessageLock = Lock(); diff --git a/packages/stream_chat/lib/src/client/retry_queue.dart b/packages/stream_chat/lib/src/client/retry_queue.dart index d65c450668..8b05181f3c 100644 --- a/packages/stream_chat/lib/src/client/retry_queue.dart +++ b/packages/stream_chat/lib/src/client/retry_queue.dart @@ -89,9 +89,8 @@ class RetryQueue { } catch (error) { logger?.severe('Error while retrying message ${message.id}', error); // If we are unable to successfully retry the message, update the state - // with the failed state, unless an earlier attempt was stored by the - // server and the message is already marked as sent. - if (!_isAlreadySent(error)) channel.state?.updateMessage(message); + // with the failed state. + channel.state?.updateMessage(message); } finally { // remove the message from the queue after it's handled. _messageQueue.removeMessage(message); @@ -101,8 +100,6 @@ class RetryQueue { _isProcessing = false; } - static bool _isAlreadySent(Object error) => error is StreamChatNetworkError && error.isMessageAlreadyExistsError; - /// Whether our [_messageQueue] has messages or not. bool get hasMessages => _messageQueue.isNotEmpty; diff --git a/packages/stream_chat/lib/src/core/error/stream_chat_error.dart b/packages/stream_chat/lib/src/core/error/stream_chat_error.dart index 9de04f1e94..b62c467e59 100644 --- a/packages/stream_chat/lib/src/core/error/stream_chat_error.dart +++ b/packages/stream_chat/lib/src/core/error/stream_chat_error.dart @@ -167,10 +167,11 @@ class StreamChatNetworkError extends StreamChatError { /// Whether the server rejected a send because it already stored a message /// with the same id. @internal - bool get isMessageAlreadyExistsError => - errorCode == ChatErrorCode.inputError && - message.contains('a message with ID') && - message.contains('already exists'); + 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 get props => [...super.props, code, statusCode, type]; diff --git a/packages/stream_chat/test/src/client/channel/channel_send_test.dart b/packages/stream_chat/test/src/client/channel/channel_send_test.dart new file mode 100644 index 0000000000..a12e38682a --- /dev/null +++ b/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())); + 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> _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, + ); +} diff --git a/packages/stream_chat/test/src/client/channel/channel_test.dart b/packages/stream_chat/test/src/client/channel/channel_test.dart index c230e6e499..21aa70487c 100644 --- a/packages/stream_chat/test/src/client/channel/channel_test.dart +++ b/packages/stream_chat/test/src/client/channel/channel_test.dart @@ -39,42 +39,6 @@ void main() { return logger; } - When> _whenSendingMessage( - StreamChatClient client, - Channel channel, - Message message, { - bool skipPush = false, - bool skipEnrichUrl = false, - }) { - return when( - () => client.sendMessage( - any(that: isSameMessageAs(message)), - channel.id!, - channel.type, - skipPush: skipPush, - skipEnrichUrl: skipEnrichUrl, - ), - ); - } - - 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, - ); - } - - // The error returned when sending a message whose id already exists. - StreamChatNetworkError _messageAlreadyExistsError(String messageId) { - return _inputError('SendMessage failed with error: "a message with ID $messageId already exists"'); - } - group('Non-Initialized Channel', () { late final client = MockStreamChatClient(); const channelId = 'test-channel-id'; @@ -314,7 +278,13 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - _whenSendingMessage(client, channel, message).thenAnswer((_) async => sendMessageResponse); + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + ), + ).thenAnswer((_) async => sendMessageResponse); expectLater( // skipping first seed message list -> [] messages @@ -358,11 +328,13 @@ void main() { user: client.state.currentUser, ); - _whenSendingMessage( - client, - channel, - message, - skipPush: true, + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + skipPush: true, + ), ).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); expectLater( @@ -410,12 +382,14 @@ void main() { user: client.state.currentUser, ); - _whenSendingMessage( - client, - channel, - message, - skipPush: true, - skipEnrichUrl: true, + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + skipPush: true, + skipEnrichUrl: true, + ), ).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); expectLater( @@ -464,11 +438,13 @@ void main() { user: client.state.currentUser, ); - _whenSendingMessage( - client, - channel, - message, - skipEnrichUrl: true, + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + skipEnrichUrl: true, + ), ).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); expectLater( @@ -516,7 +492,13 @@ void main() { user: client.state.currentUser, ); - _whenSendingMessage(client, channel, message).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + ), + ).thenThrow(StreamChatNetworkError(ChatErrorCode.notAllowed)); expectLater( // skipping first seed message list -> [] messages @@ -560,7 +542,22 @@ void main() { user: client.state.currentUser, ); - _whenSendingMessage(client, channel, message).thenThrow(_inputError('Input error')); + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + ), + ).thenThrow( + StreamChatNetworkError.raw( + code: ChatErrorCode.inputError.code, + message: 'Input error', + data: ErrorResponse() + ..code = ChatErrorCode.inputError.code + ..message = 'Input error' + ..statusCode = 400, + ), + ); expectLater( // skipping first seed message list -> [] messages @@ -592,82 +589,6 @@ void main() { ); }); - group('when the message id already exists', () { - test('marks the message as sent', () async { - final message = Message(id: 'stored-message-id', text: 'Hello world!'); - - _whenSendingMessage(client, channel, message).thenThrow(_messageAlreadyExistsError(message.id)); - await expectLater(channel.sendMessage(message), throwsA(isA())); - - expect(channel.state!.messages.single.state, MessageState.sent); - }); - - test('stamps the local creation date as the server creation date when no server copy arrived', () async { - final message = Message(id: 'stored-message-id', text: 'Hello world!'); - - _whenSendingMessage(client, channel, message).thenThrow(_messageAlreadyExistsError(message.id)); - await expectLater(channel.sendMessage(message), throwsA(isA())); - - 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 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 expectLater(channel.retryMessage(failedMessage), throwsA(isA())); - - expect(channel.state!.messages.single.createdAt, originalCreatedAt); - }); - - test('keeps the server creation date when message.new arrived first', () async { - 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 expectLater(channel.sendMessage(message), throwsA(isA())); - - 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 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 => throw errors[attempt++]); - await expectLater(channel.sendMessage(message), throwsA(isA())); - await pumpEventQueue(); - - expect(channel.state!.messages.single.state, MessageState.sent); - }); - }); - test('with attachments should work just fine', () async { final attachments = List.generate( 3, @@ -708,7 +629,13 @@ void main() { ), ).thenAnswer((_) async => sendFileResponse); - _whenSendingMessage(client, channel, message).thenAnswer( + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + ), + ).thenAnswer( (_) async => SendMessageResponse() ..message = message.copyWith( attachments: attachments @@ -936,7 +863,13 @@ void main() { ), ); - _whenSendingMessage(client, channel, message).thenAnswer( + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + ), + ).thenAnswer( (_) async => SendMessageResponse() ..message = message.copyWith( attachments: [], @@ -1007,7 +940,13 @@ void main() { ), ); - _whenSendingMessage(client, channel, message).thenAnswer( + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + ), + ).thenAnswer( (_) async => SendMessageResponse() ..message = message.copyWith( attachments: [], @@ -1073,7 +1012,13 @@ void main() { ), ); - _whenSendingMessage(client, channel, message).thenAnswer( + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + ), + ).thenAnswer( (_) async => SendMessageResponse() ..message = message.copyWith( attachments: [], @@ -6425,12 +6370,14 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - _whenSendingMessage( - client, - channel, - message, - skipPush: true, - skipEnrichUrl: true, + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + skipPush: true, + skipEnrichUrl: true, + ), ).thenAnswer((_) async => sendMessageResponse); final result = await channel.retryMessage(message); @@ -6461,7 +6408,14 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - _whenSendingMessage(client, channel, message, skipPush: true).thenAnswer((_) async => sendMessageResponse); + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + skipPush: true, + ), + ).thenAnswer((_) async => sendMessageResponse); final result = await channel.retryMessage(message); @@ -6490,7 +6444,14 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - _whenSendingMessage(client, channel, message, skipEnrichUrl: true).thenAnswer((_) async => sendMessageResponse); + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + skipEnrichUrl: true, + ), + ).thenAnswer((_) async => sendMessageResponse); final result = await channel.retryMessage(message); @@ -6519,7 +6480,13 @@ void main() { final sendMessageResponse = SendMessageResponse()..message = message.copyWith(state: MessageState.sent); - _whenSendingMessage(client, channel, message).thenAnswer((_) async => sendMessageResponse); + when( + () => client.sendMessage( + any(that: isSameMessageAs(message)), + channelId, + channelType, + ), + ).thenAnswer((_) async => sendMessageResponse); final result = await channel.retryMessage(message); diff --git a/packages/stream_chat/test/src/core/error/stream_chat_error_test.dart b/packages/stream_chat/test/src/core/error/stream_chat_error_test.dart index 924d7ff09e..7befb4aeb7 100644 --- a/packages/stream_chat/test/src/core/error/stream_chat_error_test.dart +++ b/packages/stream_chat/test/src/core/error/stream_chat_error_test.dart @@ -216,30 +216,45 @@ void main() { }); 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: 'SendMessage failed with error: "a message with ID abc already exists"', + 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: 'poll option with text A already exists', + 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: 'SendMessage failed with error: "a message with ID abc already exists"', + message: message, + data: ErrorResponse() + ..code = ChatErrorCode.notAllowed.code + ..message = message + ..statusCode = 403, ); expect(error.isMessageAlreadyExistsError, isFalse);