fix(llc): retry a message request the app was closed during (FLU-1136)

packages/stream_chat/CHANGELOG.md

@@ -7,6 +7,7 @@
 
 🐞 Fixed
 
+- Fixed a message stuck sending, updating or deleting forever when the app was closed before the request finished.
 - Fixed the connection getting stuck while connecting, and requests never completing, when the token provider throws while refreshing an expired token.
 - Fixed sending, updating and deleting a message not retrying when the server answers with a timeout (408), a rate limit (429) or a server error (5xx).
 - Fixed hard-deleting an unread message not lowering the channel's unread count.

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

@@ -28,6 +28,7 @@ class ChannelClientState {
     );
 
     final seededState = channelState.copyWith(
+      messages: _failInterruptedRequests(channelState.messages),
       pinnedMessages: MessageMerging.sortPinnedMessages(channelState.pinnedMessages),
     );
 
@@ -59,11 +60,37 @@ class ChannelClientState {
         ?.getChannelThreads(_channel.cid!)
         .then((threads) {
           // Load all the threads for the channel from the offline storage.
-          if (threads.isNotEmpty) _threads = threads;
+          if (threads.isEmpty) return;
+          _threads = threads.map(
+            (parentId, replies) => MapEntry(parentId, _failInterruptedRequests(replies)!),
+          );
         })
         .then((_) => retryFailedMessages());
   }
 
+  // A message that was still being sent, updated or deleted when this state
+  // was created belongs to a request that never finished, such as one cut off
+  // when the app was closed. A send or delete is marked failed so it gets
+  // retried. An update can't be retried as it was made, so it is settled and
+  // the server's copy replaces it.
+  static List<Message>? _failInterruptedRequests(List<Message>? messages) {
+    if (messages == null || !messages.any((it) => it.state.isOutgoing)) return messages;
+
+    return [
+      for (final message in messages)
+        switch (message.state) {
+          MessageOutgoing(state: Sending()) => message.copyWith(
+            state: MessageState.sendingFailed(skipPush: false, skipEnrichUrl: false),
+          ),
+          MessageOutgoing(state: Updating()) => message.copyWith(state: MessageState.updated),
+          MessageOutgoing(state: Deleting(:final scope)) => message.copyWith(
+            state: MessageState.deletingFailed(scope: scope),
+          ),
+          _ => message,
+        },
+    ];
+  }
+
   final Channel _channel;
   StreamChatClient get _client => _channel.client;
   final _subscriptions = CompositeSubscription();

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

@@ -0,0 +1,159 @@
+import 'package:mocktail/mocktail.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:test/test.dart';
+
+import '../../fakes.dart';
+import '../../mocks.dart';
+
+void main() {
+  setUpAll(() => registerFallbackValue(FakeMessage()));
+
+  group('Channel.fromState', () {
+    test('marks a message left sending as failed to send', () {
+      final channel = _channel(messages: [_message(state: MessageState.sending)]);
+
+      expect(
+        channel.state!.messages.single.state,
+        MessageState.sendingFailed(skipPush: false, skipEnrichUrl: false),
+      );
+    });
+
+    test('settles a message left updating, so it no longer shows as in progress', () {
+      final channel = _channel(messages: [_message(state: MessageState.updating)]);
+
+      expect(channel.state!.messages.single.state, MessageState.updated);
+    });
+
+    test('does not send a message left updating again', () async {
+      final client = MockStreamChatClientWithPersistence();
+      when(() => client.chatPersistenceClient.getChannelThreads(any())).thenAnswer((_) async => {});
+
+      _channel(
+        client: client,
+        messages: [_message(state: MessageState.updating)],
+      );
+      await pumpEventQueue();
+
+      verifyNever(
+        () => client.updateMessage(
+          any(),
+          skipPush: any(named: 'skipPush'),
+          skipEnrichUrl: any(named: 'skipEnrichUrl'),
+        ),
+      );
+      verifyNever(
+        () => client.partialUpdateMessage(
+          any(),
+          set: any(named: 'set'),
+          unset: any(named: 'unset'),
+          skipEnrichUrl: any(named: 'skipEnrichUrl'),
+        ),
+      );
+    });
+
+    test('marks a message left deleting as failed to delete for the same scope', () {
+      final channel = _channel(
+        messages: [_message(state: MessageState.deleting(scope: const MessageDeleteScope.deleteForMe()))],
+      );
+
+      expect(
+        channel.state!.messages.single.state,
+        MessageState.deletingFailed(scope: const MessageDeleteScope.deleteForMe()),
+      );
+    });
+
+    test('keeps the messages as they are when none was left in progress', () {
+      final messages = [_message(), _message(id: 'other-message')];
+
+      final channel = _channel(messages: messages);
+
+      expect(channel.state!.messages, same(messages));
+    });
+
+    test('sends a stored thread reply left sending again', () async {
+      final client = MockStreamChatClientWithPersistence();
+      final reply = _message(id: 'reply', parentId: 'parent', state: MessageState.sending);
+      when(() => client.chatPersistenceClient.getChannelThreads(any())).thenAnswer(
+        (_) async => {
+          'parent': [reply],
+        },
+      );
+      _stubSend(client);
+
+      final channel = _channel(client: client);
+      await pumpEventQueue();
+
+      verify(
+        () => client.sendMessage(any(that: isA<Message>().having((it) => it.id, 'id', 'reply')), any(), any()),
+      ).called(1);
+      expect(channel.state!.threads['parent']?.single.state.isCompleted, isTrue);
+    });
+
+    test('sends a message left sending again', () async {
+      final client = MockStreamChatClientWithPersistence();
+      when(() => client.chatPersistenceClient.getChannelThreads(any())).thenAnswer((_) async => {});
+      _stubSend(client);
+
+      final channel = _channel(
+        client: client,
+        messages: [_message(state: MessageState.sending)],
+      );
+      await pumpEventQueue();
+
+      verify(() => client.sendMessage(any(), channel.id!, channel.type)).called(1);
+    });
+  });
+}
+
+Message _message({
+  String id = 'message',
+  String? parentId,
+  MessageState state = MessageState.sent,
+}) {
+  return Message(
+    id: id,
+    text: 'Hello',
+    parentId: parentId,
+    user: User(id: 'current-user'),
+    createdAt: DateTime(2026, 10, 29, 12),
+    state: state,
+  );
+}
+
+void _stubSend(MockStreamChatClient client) {
+  when(
+    () => client.sendMessage(
+      any(),
+      any(),
+      any(),
+      skipPush: any(named: 'skipPush'),
+      skipEnrichUrl: any(named: 'skipEnrichUrl'),
+    ),
+  ).thenAnswer((invocation) async {
+    final message = invocation.positionalArguments.first as Message;
+    return SendMessageResponse()..message = message.copyWith(state: MessageState.sent);
+  });
+}
+
+Channel _channel({MockStreamChatClient? client, List<Message> messages = const []}) {
+  final chatClient = client ?? MockStreamChatClient();
+  when(
+    () => chatClient.detachedLogger(any()),
+  ).thenAnswer((invocation) => Logger.detached(invocation.positionalArguments.first as String));
+  when(() => chatClient.logger).thenReturn(Logger.detached('client'));
+  when(() => chatClient.state).thenReturn(FakeClientState());
+  when(() => chatClient.channelDeliveryReporter.submitForDelivery(any())).thenAnswer((_) async {});
+  when(() => chatClient.retryPolicy).thenReturn(
+    RetryPolicy(shouldRetry: (_, __, ___) => false, delayFactor: Duration.zero),
+  );
+
+  final channel = Channel.fromState(
+    chatClient,
+    ChannelState(
+      channel: ChannelModel(type: 'messaging', id: 'interrupted'),
+      messages: messages,
+    ),
+  );
+  addTearDown(channel.dispose);
+  return channel;
+}