diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index c28372520b..e450d099b2 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/packages/stream_chat/CHANGELOG.md @@ -15,6 +15,7 @@ - Fixed `Event.aiState` reading `AI_STATE_EXTERNAL_SOURCES` as `idle` instead of `checkingSources`. - Fixed `Channel.disableSlowMode` and `StreamChatClient.disableSlowdown` always failing. - Fixed users still showing as typing after sending or editing a message, or after switching between a thread and the channel. +- 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 e007e28c62..b92cd97b4b 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) { + 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. @@ -841,6 +843,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( @@ -859,6 +864,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. 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..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 @@ -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 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/core/error/stream_chat_error_test.dart b/packages/stream_chat/test/src/core/error/stream_chat_error_test.dart index 20f302b107..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 @@ -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);