diff --git a/packages/stream_chat/lib/src/client/channel.dart b/packages/stream_chat/lib/src/client/channel.dart index 54feb2ab..46c455cd 100644 --- a/packages/stream_chat/lib/src/client/channel.dart +++ b/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( - message, - id!, - type, - skipPush: skipPush, - skipEnrichUrl: skipEnrichUrl, + // Wait for the previous sendMessage call to finish. Otherwise, the order + // of messages will not be maintained. + final response = await _sendMessageLock.synchronized( + () => _client.sendMessage( + message, + 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( - message, - skipEnrichUrl: skipEnrichUrl, + // Wait for the previous update call to finish. Otherwise, the order of + // messages will not be maintained. + final response = await _updateMessageLock.synchronized( + () => _client.updateMessage( + message, + skipEnrichUrl: skipEnrichUrl, + ), ); - final m = response.message.copyWith( - ownReactions: message.ownReactions, - ); + final updatedMessage = response.message + .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( - message.id, - set: set, - unset: unset, - skipEnrichUrl: skipEnrichUrl, + // Wait for the previous update call to finish. Otherwise, the order of + // messages will not be maintained. + final response = await _updateMessageLock.synchronized( + () => _client.partialUpdateMessage( + message.id, + set: set, + unset: unset, + skipEnrichUrl: skipEnrichUrl, + ), ); - final updatedMessage = response.message.copyWith( - ownReactions: message.ownReactions, - ); + final updatedMessage = response.message + .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 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); - - state?.deleteMessage( - message.copyWith(status: MessageSendingStatus.sent), - hardDelete: hardDelete, + // Wait for the previous delete call to finish. Otherwise, the order of + // messages will not be maintained. + final response = await _deleteMessageLock.synchronized( + () => _client.deleteMessage(message.id, hard: hard), ); + 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 || - (message.parentId != null && message.showInChannel != true)) { + final showInChannel = + 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 _updatedStateMessages = updatedState.messages ?? []; + final _existingStateMessages = [...messages]; final newMessages = [ - ..._updatedStateMessages, - ..._existingStateMessages - .where((m) => - !_updatedStateMessages.any((newMessage) => newMessage.id == m.id)) - .toList(), - ]..sort(_sortByCreatedAt); + ..._existingStateMessages.merge(updatedState.messages), + ].sorted(_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 { + Iterable merge(Iterable? 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; + } +} diff --git a/packages/stream_chat/lib/src/client/client.dart b/packages/stream_chat/lib/src/client/client.dart index f131b12f..ec6443e9 100644 --- a/packages/stream_chat/lib/src/client/client.dart +++ b/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 sync({List? cids, DateTime? lastSyncAt}) { - return synchronized(() async { + return _syncLock.synchronized(() async { final channels = cids ?? await chatPersistenceClient?.getChannelCids(); if (channels == null || channels.isEmpty) { return; diff --git a/packages/stream_chat/lib/src/core/models/message.dart b/packages/stream_chat/lib/src/core/models/message.dart index f8631e1c..14d23d18 100644 --- a/packages/stream_chat/lib/src/core/models/message.dart +++ b/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, - _updatedAt = updatedAt, + remoteCreatedAt = createdAt, + 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; - - /// Reserved field indicating when the message was deleted. + /// Indicates when the message was created. + /// + /// 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, - updatedAt: updatedAt ?? _updatedAt, - deletedAt: deletedAt ?? this.deletedAt, + createdAt: createdAt ?? remoteCreatedAt, + localCreatedAt: localCreatedAt ?? this.localCreatedAt, + 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, - updatedAt: other.updatedAt, - deletedAt: other.deletedAt, + createdAt: other.remoteCreatedAt, + localCreatedAt: other.localCreatedAt, + 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 get props => [ id, @@ -407,8 +470,8 @@ class Message extends Equatable { shadowed, silent, command, - _createdAt, - _updatedAt, + createdAt, + updatedAt, deletedAt, user, pinned,