fix(llc): fix message order when sent quickly.

Signed-off-by: xsahil03x <[email protected]>

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

@@ -7,6 +7,7 @@ import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/src/client/retry_queue.dart';
 import 'package:stream_chat/src/core/util/utils.dart';
 import 'package:stream_chat/stream_chat.dart';
+import 'package:synchronized/synchronized.dart';
 
 /// The maximum time the incoming [Event.typingStart] event is valid before a
 /// [Event.typingStop] event is emitted automatically.
@@ -563,6 +564,8 @@ class Channel {
     });
   }
 
+  final _sendMessageLock = Lock();
+
   /// Send a [message] to this channel.
   ///
   /// If [skipPush] is true the message will not send a push notification.
@@ -586,7 +589,7 @@ class Channel {
     );
     // ignore: parameter_assignments
     message = message.copyWith(
-      createdAt: message.createdAt,
+      localCreatedAt: DateTime.now(),
       user: _client.state.currentUser,
       quotedMessage: quotedMessage,
       status: MessageSendingStatus.sending,
@@ -615,14 +618,21 @@ class Channel {
         message = await attachmentsUploadCompleter.future;
       }
 
-      final response = await _client.sendMessage(
+      // Wait for the previous sendMessage call to finish. Otherwise, the order
-        message,
+      // of messages will not be maintained.
-        id!,
+      final response = await _sendMessageLock.synchronized(
-        type,
+        () => _client.sendMessage(
-        skipPush: skipPush,
+          message,
-        skipEnrichUrl: skipEnrichUrl,
+          id!,
+          type,
+          skipPush: skipPush,
+          skipEnrichUrl: skipEnrichUrl,
+        ),
       );
-      state!.updateMessage(response.message);
+
+      final sentMessage = response.message.syncWith(message);
+
+      state!.updateMessage(sentMessage);
       if (cooldown > 0) cooldownStartedAt = DateTime.now();
       return response;
     } catch (e) {
@@ -633,6 +643,8 @@ class Channel {
     }
   }
 
+  final _updateMessageLock = Lock();
+
   /// Updates the [message] in this channel.
   ///
   /// Waits for a [_messageAttachmentsUploadCompleter] to complete
@@ -652,7 +664,7 @@ class Channel {
     // ignore: parameter_assignments
     message = message.copyWith(
       status: MessageSendingStatus.updating,
-      updatedAt: message.updatedAt,
+      localUpdatedAt: DateTime.now(),
       attachments: message.attachments.map(
         (it) {
           if (it.uploadState.isSuccess) return it;
@@ -678,16 +690,20 @@ class Channel {
         message = await attachmentsUploadCompleter.future;
       }
 
-      final response = await _client.updateMessage(
+      // Wait for the previous update call to finish. Otherwise, the order of
-        message,
+      // messages will not be maintained.
-        skipEnrichUrl: skipEnrichUrl,
+      final response = await _updateMessageLock.synchronized(
+        () => _client.updateMessage(
+          message,
+          skipEnrichUrl: skipEnrichUrl,
+        ),
       );
 
-      final m = response.message.copyWith(
+      final updatedMessage = response.message
-        ownReactions: message.ownReactions,
+          .syncWith(message)
-      );
+          .copyWith(ownReactions: message.ownReactions);
 
-      state?.updateMessage(m);
+      state?.updateMessage(updatedMessage);
 
       return response;
     } catch (e) {
@@ -714,16 +730,20 @@ class Channel {
     bool skipEnrichUrl = false,
   }) async {
     try {
-      final response = await _client.partialUpdateMessage(
+      // Wait for the previous update call to finish. Otherwise, the order of
-        message.id,
+      // messages will not be maintained.
-        set: set,
+      final response = await _updateMessageLock.synchronized(
-        unset: unset,
+        () => _client.partialUpdateMessage(
-        skipEnrichUrl: skipEnrichUrl,
+          message.id,
+          set: set,
+          unset: unset,
+          skipEnrichUrl: skipEnrichUrl,
+        ),
       );
 
-      final updatedMessage = response.message.copyWith(
+      final updatedMessage = response.message
-        ownReactions: message.ownReactions,
+          .syncWith(message)
-      );
+          .copyWith(ownReactions: message.ownReactions);
 
       state?.updateMessage(updatedMessage);
 
@@ -736,6 +756,8 @@ class Channel {
     }
   }
 
+  final _deleteMessageLock = Lock();
+
   /// Deletes the [message] from the channel.
   Future<EmptyResponse> deleteMessage(Message message, {bool? hard}) async {
     final hardDelete = hard ?? false;
@@ -746,7 +768,7 @@ class Channel {
       state!.deleteMessage(
         message.copyWith(
           type: 'deleted',
-          deletedAt: message.deletedAt ?? DateTime.now(),
+          localDeletedAt: DateTime.now(),
           status: MessageSendingStatus.sent,
         ),
         hardDelete: hardDelete,
@@ -770,13 +792,18 @@ class Channel {
 
       state?.deleteMessage(message, hardDelete: hardDelete);
 
-      final response = await _client.deleteMessage(message.id, hard: hard);
+      // Wait for the previous delete call to finish. Otherwise, the order of
-
+      // messages will not be maintained.
-      state?.deleteMessage(
+      final response = await _deleteMessageLock.synchronized(
-        message.copyWith(status: MessageSendingStatus.sent),
+        () => _client.deleteMessage(message.id, hard: hard),
-        hardDelete: hardDelete,
       );
 
+      final deletedMessage = message.copyWith(
+        status: MessageSendingStatus.sent,
+      );
+
+      state?.deleteMessage(deletedMessage, hardDelete: hardDelete);
+
       return response;
     } catch (e) {
       if (e is StreamChatNetworkError && e.isRetriable) {
@@ -1942,8 +1969,9 @@ class ChannelClientState {
     )
         .listen((event) {
       final message = event.message!;
-      if (isUpToDate ||
+      final showInChannel =
-          (message.parentId != null && message.showInChannel != true)) {
+          message.parentId != null && message.showInChannel != true;
+      if (isUpToDate || showInChannel) {
         updateMessage(message);
       }
 
@@ -1960,10 +1988,10 @@ class ChannelClientState {
       var newMessages = [...messages];
       final oldIndex = newMessages.indexWhere((m) => m.id == message.id);
       if (oldIndex != -1) {
-        var updatedMessage = message;
+        final oldMessage = newMessages[oldIndex];
+        var updatedMessage = message.syncWith(oldMessage);
         // Add quoted message to the message if it is not present.
         if (message.quotedMessageId != null && message.quotedMessage == null) {
-          final oldMessage = newMessages[oldIndex];
           updatedMessage = updatedMessage.copyWith(
             quotedMessage: oldMessage.quotedMessage,
           );
@@ -1980,7 +2008,7 @@ class ChannelClientState {
           return it.copyWith(
             quotedMessage: updatedMessage.copyWith(
               type: 'deleted',
-              deletedAt: updatedMessage.deletedAt ?? DateTime.now(),
+              deletedAt: DateTime.now(),
             ),
           );
         }).toList();
@@ -2004,7 +2032,7 @@ class ChannelClientState {
       }
 
       _channelState = _channelState.copyWith(
-        messages: newMessages..sort(_sortByCreatedAt),
+        messages: newMessages.sorted(_sortByCreatedAt),
         pinnedMessages: newPinnedMessages,
         channel: _channelState.channel?.copyWith(
           lastMessageAt: message.createdAt,
@@ -2226,7 +2254,7 @@ class ChannelClientState {
         ...newThreads[parentId]!.where(
           (newMessage) => !messages.any((m) => m.id == newMessage.id),
         ),
-      ]..sort(_sortByCreatedAt);
+      ].sorted(_sortByCreatedAt);
     } else {
       newThreads[parentId] = messages;
     }
@@ -2245,15 +2273,10 @@ class ChannelClientState {
 
   /// Update channelState with updated information.
   void updateChannelState(ChannelState updatedState) {
-    final _existingStateMessages = _channelState.messages ?? [];
+    final _existingStateMessages = [...messages];
-    final _updatedStateMessages = updatedState.messages ?? [];
     final newMessages = <Message>[
-      ..._updatedStateMessages,
+      ..._existingStateMessages.merge(updatedState.messages),
-      ..._existingStateMessages
+    ].sorted(_sortByCreatedAt);
-          .where((m) =>
-              !_updatedStateMessages.any((newMessage) => newMessage.id == m.id))
-          .toList(),
-    ]..sort(_sortByCreatedAt);
 
     final _existingStateWatchers = _channelState.watchers ?? [];
     final _updatedStateWatchers = updatedState.watchers ?? [];
@@ -2308,6 +2331,7 @@ class ChannelClientState {
   final Debounce _debouncedUpdatePersistenceChannelState;
 
   set _channelState(ChannelState v) {
+    print('State: ${StackTrace.current}');
     _channelStateController.add(v);
     _debouncedUpdatePersistenceChannelState.call([v]);
   }
@@ -2471,3 +2495,21 @@ bool _pinIsValid(Message message) {
   final now = DateTime.now();
   return message.pinExpires!.isAfter(now);
 }
+
+extension on Iterable<Message> {
+  Iterable<Message> merge(Iterable<Message>? other) {
+    if (other == null) return this;
+
+    final messageMap = {for (final message in this) message.id: message};
+
+    for (final message in other) {
+      messageMap.update(
+        message.id,
+        message.syncWith,
+        ifAbsent: () => message,
+      );
+    }
+
+    return messageMap.values;
+  }
+}

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

@@ -32,7 +32,7 @@ import 'package:stream_chat/src/event_type.dart';
 import 'package:stream_chat/src/ws/connection_status.dart';
 import 'package:stream_chat/src/ws/websocket.dart';
 import 'package:stream_chat/version.dart';
-import 'package:synchronized/extension.dart';
+import 'package:synchronized/synchronized.dart';
 
 /// Handler function used for logging records. Function requires a single
 /// [LogRecord] as the only parameter.
@@ -526,10 +526,13 @@ class StreamChatClient {
         event.type == eventType4);
   }
 
+  // Lock to make sure only one sync process is running at a time.
+  final _syncLock = Lock();
+
   /// Get the events missed while offline to sync the offline storage
   /// Will automatically fetch [cids] and [lastSyncedAt] if [persistenceEnabled]
   Future<void> sync({List<String>? cids, DateTime? lastSyncAt}) {
-    return synchronized(() async {
+    return _syncLock.synchronized(() async {
       final channels = cids ?? await chatPersistenceClient?.getChannelCids();
       if (channels == null || channels.isEmpty) {
         return;

packages/stream_chat/lib/src/core/models/message.dart

@@ -64,8 +64,11 @@ class Message extends Equatable {
     this.showInChannel,
     this.command,
     DateTime? createdAt,
+    this.localCreatedAt,
     DateTime? updatedAt,
-    this.deletedAt,
+    this.localUpdatedAt,
+    DateTime? deletedAt,
+    this.localDeletedAt,
     this.user,
     this.pinned = false,
     this.pinnedAt,
@@ -76,8 +79,9 @@ class Message extends Equatable {
     this.i18n,
   })  : id = id ?? const Uuid().v4(),
         pinExpires = pinExpires?.toUtc(),
-        _createdAt = createdAt,
+        remoteCreatedAt = createdAt,
-        _updatedAt = updatedAt,
+        remoteUpdatedAt = updatedAt,
+        remoteDeletedAt = deletedAt,
         _quotedMessageId = quotedMessageId;
 
   /// Create a new instance from JSON.
@@ -161,21 +165,49 @@ class Message extends Equatable {
   @JsonKey(includeToJson: false)
   final String? command;
 
-  final DateTime? _createdAt;
+  /// Indicates when the message was created.
-
+  ///
-  /// Reserved field indicating when the message was deleted.
+  /// Returns the latest between [localCreatedAt] and [remoteCreatedAt].
+  /// If both are null, returns [DateTime.now].
   @JsonKey(includeToJson: false)
-  final DateTime? deletedAt;
+  DateTime get createdAt => localCreatedAt ?? remoteCreatedAt ?? DateTime.now();
 
-  /// Reserved field indicating when the message was created.
+  /// Indicates when the message was created locally.
+  @JsonKey(includeToJson: false, includeFromJson: false)
+  final DateTime? localCreatedAt;
+
+  /// Indicates when the message was created on the server.
+  @JsonKey(includeToJson: false, includeFromJson: false)
+  final DateTime? remoteCreatedAt;
+
+  /// Indicates when the message was updated last time.
+  ///
+  /// Returns the latest between [localUpdatedAt] and [remoteUpdatedAt].
+  /// If both are null, returns [createdAt].
   @JsonKey(includeToJson: false)
-  DateTime get createdAt => _createdAt ?? DateTime.now();
+  DateTime get updatedAt => localUpdatedAt ?? remoteUpdatedAt ?? createdAt;
 
-  final DateTime? _updatedAt;
+  /// Indicates when the message was updated locally.
+  @JsonKey(includeToJson: false, includeFromJson: false)
+  final DateTime? localUpdatedAt;
 
-  /// Reserved field indicating when the message was updated last time.
+  /// Indicates when the message was updated on the server.
+  @JsonKey(includeToJson: false, includeFromJson: false)
+  final DateTime? remoteUpdatedAt;
+
+  /// Indicates when the message was deleted.
+  ///
+  /// Returns the latest between [localDeletedAt] and [remoteDeletedAt].
   @JsonKey(includeToJson: false)
-  DateTime get updatedAt => _updatedAt ?? DateTime.now();
+  DateTime? get deletedAt => localDeletedAt ?? remoteDeletedAt;
+
+  /// Indicates when the message was deleted locally.
+  @JsonKey(includeToJson: false, includeFromJson: false)
+  final DateTime? localDeletedAt;
+
+  /// Indicates when the message was deleted on the server.
+  @JsonKey(includeToJson: false, includeFromJson: false)
+  final DateTime? remoteDeletedAt;
 
   /// User who sent the message.
   @JsonKey(includeToJson: false)
@@ -273,8 +305,11 @@ class Message extends Equatable {
     bool? showInChannel,
     String? command,
     DateTime? createdAt,
+    DateTime? localCreatedAt,
     DateTime? updatedAt,
+    DateTime? localUpdatedAt,
     DateTime? deletedAt,
+    DateTime? localDeletedAt,
     User? user,
     bool? pinned,
     DateTime? pinnedAt,
@@ -338,9 +373,12 @@ class Message extends Equatable {
       threadParticipants: threadParticipants ?? this.threadParticipants,
       showInChannel: showInChannel ?? this.showInChannel,
       command: command ?? this.command,
-      createdAt: createdAt ?? _createdAt,
+      createdAt: createdAt ?? remoteCreatedAt,
-      updatedAt: updatedAt ?? _updatedAt,
+      localCreatedAt: localCreatedAt ?? this.localCreatedAt,
-      deletedAt: deletedAt ?? this.deletedAt,
+      updatedAt: updatedAt ?? remoteUpdatedAt,
+      localUpdatedAt: localUpdatedAt ?? this.localUpdatedAt,
+      deletedAt: deletedAt ?? remoteDeletedAt,
+      localDeletedAt: localDeletedAt ?? this.localDeletedAt,
       user: user ?? this.user,
       pinned: pinned ?? this.pinned,
       pinnedAt: pinnedAt ?? this.pinnedAt,
@@ -374,9 +412,12 @@ class Message extends Equatable {
         threadParticipants: other.threadParticipants,
         showInChannel: other.showInChannel,
         command: other.command,
-        createdAt: other.createdAt,
+        createdAt: other.remoteCreatedAt,
-        updatedAt: other.updatedAt,
+        localCreatedAt: other.localCreatedAt,
-        deletedAt: other.deletedAt,
+        updatedAt: other.remoteUpdatedAt,
+        localUpdatedAt: other.localUpdatedAt,
+        deletedAt: other.remoteDeletedAt,
+        localDeletedAt: other.localDeletedAt,
         user: other.user,
         pinned: other.pinned,
         pinnedAt: other.pinnedAt,
@@ -387,6 +428,28 @@ class Message extends Equatable {
         i18n: other.i18n,
       );
 
+  /// Returns a new [Message] that is [other] with local changes applied to it.
+  ///
+  /// This ensures that the local sync changes are not lost when the message is
+  /// updated on the server.
+  ///
+  /// For example, when a message is sent, it is immediately shown
+  /// optimistically in the UI. When the message is received from the server,
+  /// it will not contain the local changes. This method can be used to merge
+  /// the local changes back into the message.
+  ///
+  /// This also helps in maintaining the order of the messages in the channel
+  /// when the messages are sorted by the [createdAt] field.
+  Message syncWith(Message? other) {
+    if (other == null) return this;
+
+    return copyWith(
+      localCreatedAt: other.localCreatedAt,
+      localUpdatedAt: other.localUpdatedAt,
+      localDeletedAt: other.localDeletedAt,
+    );
+  }
+
   @override
   List<Object?> get props => [
         id,
@@ -407,8 +470,8 @@ class Message extends Equatable {
         shadowed,
         silent,
         command,
-        _createdAt,
+        createdAt,
-        _updatedAt,
+        updatedAt,
         deletedAt,
         user,
         pinned,