channel.dart

// ignore_for_file: avoid_redundant_argument_values

import 'dart:async';
import 'dart:math' as math;

import 'package:collection/collection.dart';
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.
const incomingTypingStartEventTimeout = 7;

/// Class that manages a specific channel.
///
/// #### Channel name
///
/// {@template name}
/// If an optional [name] argument is provided in the constructor then it
/// will be set on [extraData] with a key of 'name'.
///
/// ```dart
/// final channel = Channel(client, type, id, name: 'Channel name');
/// print(channel.name == channel.extraData['name']); // true
/// ```
///
/// Before the channel is initialized the name can be set directly:
/// ```dart
/// channel.name = 'New channel name';
/// ```
///
/// To update the name after the channel has been initialized, call:
/// ```dart
/// channel.updateName('Updated channel name');
/// ```
///
/// This will do a partial update to update the name.
/// {@endtemplate}
///
/// #### Channel image
///
/// {@template image}
/// If an optional [image] argument is provided in the constructor then it
/// will be set on [extraData] with a key of 'image'.
///
/// ```dart
/// final channel = Channel(client, type, id, image: 'https://getstream.io/image.png');
/// print(channel.image == channel.extraData['image']); // true
/// ```
///
/// Before the channel is initialized the image can be set directly:
/// ```dart
/// channel.image = 'https://getstream.io/new-image';
/// ```
///
/// To update the image after the channel has been initialized, call:
/// ```dart
/// channel.updateImage('https://getstream.io/new-image');
/// ```
///
/// This will do a partial update to update the image.
/// {@endtemplate}
class Channel {
  /// Class that manages a specific channel.
  ///
  /// Optional [extraData] and [image] properties can be provided. The [image]
  /// is exposed to easily set a key of 'image' on [extraData].
  Channel(
    this._client,
    this._type,
    this._id, {
    String? name,
    String? image,
    Map<String, Object?>? extraData,
  }) : _cid = _id != null ? '$_type:$_id' : null,
       _extraData = {
         ...?extraData,
         if (name != null) 'name': name,
         if (image != null) 'image': image,
       } {
    _client.logger.info('New Channel instance created, not yet initialized');
  }

  /// Create a channel client instance from a [ChannelState] object.
  Channel.fromState(this._client, ChannelState channelState)
    : assert(
        channelState.channel != null,
        'No channel found inside channel state',
      ),
      _id = channelState.channel!.id,
      _type = channelState.channel!.type,
      _cid = channelState.channel!.cid,
      _extraData = channelState.channel!.extraData {
    _initState(channelState); // Initialize the state immediately.
  }

  /// This client state
  ChannelClientState? state;

  /// The channel type
  final String _type;

  String? _id;
  String? _cid;
  final Map<String, Object?> _extraData;

  /// Shortcut to set channel name.
  ///
  /// {@macro name}
  set name(String? name) {
    if (_isInitialized) {
      throw StateError(
        'Once the channel is initialized you should use `channel.updateName` '
        'to update the channel name',
      );
    }
    _extraData.addAll({'name': name});
  }

  /// Shortcut to set channel image.
  ///
  /// {@macro image}
  set image(String? image) {
    if (_isInitialized) {
      throw StateError(
        'Once the channel is initialized you should use `channel.updateImage` '
        'to update the channel image',
      );
    }
    _extraData.addAll({'image': image});
  }

  set extraData(Map<String, Object?> extraData) {
    if (_isInitialized) {
      throw StateError(
        'Once the channel is initialized you should use `channel.update` '
        'to update channel data',
      );
    }
    _extraData.addAll(extraData);
  }

  /// Whether this channel is identified by its member set rather than an
  /// explicit id.
  ///
  /// Stream auto-generates ids of the form `!members-<hash>` for channels
  /// created with members but no id, so the same set of users always
  /// references the same channel.
  ///
  /// Distinct channels can lose members but can't gain them after creation.
  ///
  /// See [isOneToOne] for the typical 1-to-1 predicate built on this.
  bool get isDistinct => id?.startsWith('!members') == true;

  /// Whether this is a group channel.
  ///
  /// True when the channel has more than two members, or isn't [isDistinct].
  /// Custom-id channels are treated as groups regardless of current member
  /// count because they aren't bounded — they can grow back into
  /// multi-person conversations.
  ///
  /// Near-inverse of [isOneToOne].
  bool get isGroup => (memberCount ?? 0) > 2 || !isDistinct;

  /// Whether this is a 1-to-1 conversation.
  ///
  /// True when the channel is [isDistinct] and has exactly two members.
  /// Distinct channels can't gain members, so a 2-member distinct channel
  /// is permanently bounded to two participants — including channels that
  /// shrunk down from a larger group DM.
  ///
  /// This is a structural predicate without a current-user check. Combine
  /// with capability / permission checks at the call site if you need
  /// perspective gating.
  ///
  /// Near-inverse of [isGroup].
  bool get isOneToOne => isDistinct && memberCount == 2;

  /// Returns true if the channel is muted.
  bool get isMuted {
    final channelMutes = _client.state.currentUser?.channelMutes;
    if (channelMutes == null) return false;

    return channelMutes.any((it) => it.channel.cid == cid);
  }

  /// Returns true if the channel is muted, as a stream.
  Stream<bool> get isMutedStream => _client.state.currentUserStream.map((user) {
    final channelMutes = user?.channelMutes;
    if (channelMutes == null) return false;

    return channelMutes.any((it) => it.channel.cid == cid);
  }).distinct();

  /// Channel configuration.
  ChannelConfig? get config {
    _checkInitialized();
    return state!._channelState.channel?.config;
  }

  /// Channel configuration as a stream.
  Stream<ChannelConfig?> get configStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.config);
  }

  /// Relationship of the current user to this channel.
  Member? get membership {
    _checkInitialized();
    return state!._channelState.membership;
  }

  /// Relationship of the current user to this channel as a stream.
  Stream<Member?> get membershipStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.membership);
  }

  /// Channel user creator.
  User? get createdBy {
    _checkInitialized();
    return state!._channelState.channel?.createdBy;
  }

  /// Channel user creator as a stream.
  Stream<User?> get createdByStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.createdBy);
  }

  /// Channel frozen status.
  bool get frozen {
    _checkInitialized();
    return state!._channelState.channel?.frozen == true;
  }

  /// Channel frozen status as a stream.
  Stream<bool> get frozenStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.frozen == true);
  }

  /// Channel disabled status.
  bool get disabled {
    _checkInitialized();
    return state!._channelState.channel?.disabled == true;
  }

  /// Channel disabled status as a stream.
  Stream<bool> get disabledStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.disabled == true);
  }

  /// Channel hidden status.
  bool get hidden {
    _checkInitialized();
    return state!._channelState.channel?.hidden == true;
  }

  /// Channel hidden status as a stream.
  Stream<bool> get hiddenStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.hidden == true);
  }

  /// Channel pinned status.
  /// Status is specific to the current user.
  bool get isPinned {
    _checkInitialized();
    return membership?.pinnedAt != null;
  }

  /// Channel pinned status as a stream.
  /// Status is specific to the current user.
  Stream<bool> get isPinnedStream {
    return membershipStream.map((m) => m?.pinnedAt != null);
  }

  /// Channel archived status.
  /// Status is specific to the current user.
  bool get isArchived {
    _checkInitialized();
    return membership?.archivedAt != null;
  }

  /// Channel archived status as a stream.
  /// Status is specific to the current user.
  Stream<bool> get isArchivedStream {
    return membershipStream.map((m) => m?.archivedAt != null);
  }

  /// The last date at which the channel got truncated.
  DateTime? get truncatedAt {
    _checkInitialized();
    return state!._channelState.channel?.truncatedAt;
  }

  /// The last date at which the channel got truncated as a stream.
  Stream<DateTime?> get truncatedAtStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.truncatedAt);
  }

  /// Cooldown count
  int get cooldown {
    _checkInitialized();
    return state!._channelState.channel?.cooldown ?? 0;
  }

  /// Cooldown count as a stream
  Stream<int> get cooldownStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.cooldown ?? 0);
  }

  /// Remaining cooldown duration in seconds for the channel.
  ///
  /// Returns 0 if there is no cooldown active.
  ///
  /// Optionally, provide [lastMessageAt] to calculate the remaining cooldown based on a specific message timestamp
  /// instead of the last message sent by the current user in this channel.
  int getRemainingCooldown({DateTime? lastMessageAt}) {
    _checkInitialized();

    final cooldownDuration = cooldown;
    if (cooldownDuration <= 0) return 0;

    final userLastMessageAt = lastMessageAt ?? currentUserLastMessageAt;
    if (userLastMessageAt == null) return 0;

    if (canSkipSlowMode) return 0;

    final currentTime = DateTime.timestamp();
    final elapsedTime = currentTime.difference(userLastMessageAt).inSeconds;

    return math.max(0, cooldownDuration - elapsedTime);
  }

  /// Channel creation date.
  DateTime? get createdAt {
    _checkInitialized();
    return state!._channelState.channel?.createdAt;
  }

  /// Channel creation date as a stream.
  Stream<DateTime?> get createdAtStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.createdAt);
  }

  /// Channel last message date.
  DateTime? get lastMessageAt {
    _checkInitialized();
    return state!._channelState.channel?.lastMessageAt;
  }

  /// Channel last message date as a stream.
  Stream<DateTime?> get lastMessageAtStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.lastMessageAt);
  }

  DateTime? _currentUserLastMessageAt({
    required List<Message>? messages,
    required Map<String, List<Message>> threads,
  }) {
    final currentUserId = client.state.currentUser?.id;
    if (currentUserId == null) return null;

    bool ours(Message m) => !m.isEphemeral && m.user?.id == currentUserId;

    DateTime? max;

    if (messages != null) {
      final idx = messages.lastIndexWhere(ours);
      if (idx != -1) max = messages[idx].createdAt;
    }

    for (final replies in threads.values) {
      final idx = replies.lastIndexWhere(ours);
      if (idx == -1) continue;
      final createdAt = replies[idx].createdAt;
      if (max == null || createdAt.isAfter(max)) max = createdAt;
    }

    return max;
  }

  /// The date of the last message sent by the current user.
  ///
  /// Returns null if the channel is not up to date or
  /// if the current user has not sent any messages in this channel.
  ///
  /// Note: This includes both regular messages and thread messages.
  DateTime? get currentUserLastMessageAt {
    _checkInitialized();

    // If the channel is not up to date, we can't rely on the last message
    // from the current user.
    if (!state!.isUpToDate) return null;

    final threads = state!.threads;
    final messages = state!.channelState.messages;

    return _currentUserLastMessageAt(messages: messages, threads: threads);
  }

  /// The date of the last message sent by the current user as a stream.
  ///
  /// Returns null if the channel is not up to date or
  /// if the current user has not sent any messages in this channel.
  ///
  /// Note: This includes both regular messages and thread messages.
  Stream<DateTime?> get currentUserLastMessageAtStream {
    _checkInitialized();

    return CombineLatestStream.combine3(
      state!.isUpToDateStream,
      state!.channelStateStream.map((s) => s.messages).distinct(identical),
      state!.threadsStream,
      (isUpToDate, messages, threads) {
        // If the channel is not up to date, we can't rely on the last message
        // from the current user.
        if (!isUpToDate) return null;

        return _currentUserLastMessageAt(messages: messages, threads: threads);
      },
    );
  }

  /// Channel updated date.
  DateTime? get updatedAt {
    _checkInitialized();
    return state!._channelState.channel?.updatedAt;
  }

  /// Channel updated date as a stream.
  Stream<DateTime?> get updatedAtStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.updatedAt);
  }

  /// Channel deletion date.
  DateTime? get deletedAt {
    _checkInitialized();
    return state!._channelState.channel?.deletedAt;
  }

  /// Channel deletion date as a stream.
  Stream<DateTime?> get deletedAtStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.deletedAt);
  }

  /// Channel member count.
  int? get memberCount {
    _checkInitialized();
    return state!._channelState.channel?.memberCount;
  }

  /// Channel member count as a stream.
  Stream<int?> get memberCountStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.memberCount);
  }

  /// Channel message count.
  ///
  /// Note: This field is only populated if the `count_messages` option is
  /// enabled for your app.
  int? get messageCount {
    _checkInitialized();
    return state!._channelState.channel?.messageCount;
  }

  /// Channel message count as a stream.
  ///
  /// Note: This field is only populated if the `count_messages` option is
  /// enabled for your app.
  Stream<int?> get messageCountStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.messageCount);
  }

  /// List of filter tags applied to this channel.
  ///
  /// Generally used for filtering channels while querying.
  List<String>? get filterTags {
    _checkInitialized();
    return state!._channelState.channel?.filterTags;
  }

  /// Channel id.
  String? get id => state?._channelState.channel?.id ?? _id;

  /// Channel type.
  String get type => state?._channelState.channel?.type ?? _type;

  /// Channel cid.
  String? get cid => state?._channelState.channel?.cid ?? _cid;

  /// Channel team.
  String? get team {
    _checkInitialized();
    return state!._channelState.channel?.team;
  }

  /// Channel extra data.
  Map<String, Object?> get extraData {
    var data = state?._channelState.channel?.extraData;
    if (data == null || data.isEmpty) {
      data = _extraData;
    }
    return data;
  }

  /// List of user permissions on this channel
  List<ChannelCapability> get ownCapabilities => state?._channelState.channel?.ownCapabilities ?? [];

  /// List of user permissions on this channel
  Stream<List<ChannelCapability>> get ownCapabilitiesStream {
    _checkInitialized();
    return state!.channelStateStream.map((cs) => cs.channel?.ownCapabilities ?? []).distinct();
  }

  /// Channel extra data as a stream.
  Stream<Map<String, Object?>> get extraDataStream {
    _checkInitialized();
    return state!.channelStateStream.map(
      (cs) => cs.channel?.extraData ?? _extraData,
    );
  }

  /// Shortcut to get channel name.
  ///
  /// {@macro name}
  String? get name => extraData['name'] as String?;

  /// Channel [name] as a stream.
  ///
  /// The channel needs to be initialized.
  ///
  /// {@macro name}
  Stream<String?> get nameStream {
    _checkInitialized();
    return extraDataStream.map((it) => it['name'] as String?);
  }

  /// Shortcut to get channel image.
  ///
  /// {@macro image}
  String? get image => extraData['image'] as String?;

  /// Channel [image] as a stream.
  ///
  /// The channel needs to be initialized.
  ///
  /// {@macro image}
  Stream<String?> get imageStream {
    _checkInitialized();
    return extraDataStream.map((it) => it['image'] as String?);
  }

  /// The main Stream chat client.
  StreamChatClient get client => _client;
  final StreamChatClient _client;

  Completer<bool> _initializedCompleter = Completer();

  /// True if this is initialized.
  ///
  /// Call [watch] to initialize the client or instantiate it using
  /// [Channel.fromState].
  Future<bool> get initialized => _initializedCompleter.future;

  // Whether the channel is successfully initialized and not disposed.
  bool get _isInitialized => _initializedCompleter.isCompleted && state != null;

  final _cancelableAttachmentUploadRequest = <String, CancelToken>{};
  final _messageAttachmentsUploadCompleter = <String, Completer<Message>>{};

  /// Cancels [attachmentId] upload request. Throws exception if the request
  /// hasn't even started yet, Already completed or Already cancelled.
  ///
  /// Optionally, provide a [reason] for the cancellation.
  void cancelAttachmentUpload(
    String attachmentId, {
    String? reason,
  }) {
    final cancelToken = _cancelableAttachmentUploadRequest[attachmentId];
    if (cancelToken == null) {
      throw const StreamChatError(
        "Upload request for this Attachment hasn't started yet or maybe "
        'Already completed',
      );
    }
    if (cancelToken.isCancelled) {
      throw const StreamChatError('Upload request already cancelled');
    }
    cancelToken.cancel(reason);
  }

  /// Retries the failed [attachmentId] upload request.
  Future<void> retryAttachmentUpload(String messageId, String attachmentId) =>
      _uploadAttachments(messageId, [attachmentId]);

  Future<void> _uploadAttachments(
    String messageId,
    Iterable<String> attachmentIds,
  ) {
    var message = [
      ...state!.messages,
      ...state!.threads.values.expand((messages) => messages),
    ].firstWhereOrNull((it) => it.id == messageId);

    if (message == null) {
      throw const StreamChatError('Error, Message not found');
    }

    final attachments = message.attachments.where((it) {
      if (it.uploadState.isSuccess) return false;
      return attachmentIds.contains(it.id);
    });

    if (attachments.isEmpty) {
      client.logger.info('No attachments available to upload');
      if (message.attachments.every((it) => it.uploadState.isSuccess)) {
        _messageAttachmentsUploadCompleter.remove(messageId)?.complete(message);
      }
      return Future.value();
    }

    client.logger.info('Found ${attachments.length} attachments');

    void updateAttachment(Attachment attachment, {bool remove = false}) {
      final index = message!.attachments.indexWhere(
        (it) => it.id == attachment.id,
      );
      if (index != -1) {
        // update or remove attachment from message.
        final List<Attachment> newAttachments;
        if (remove) {
          newAttachments = [...message!.attachments]..removeAt(index);
        } else {
          newAttachments = [...message!.attachments]..[index] = attachment;
        }

        final updatedMessage = message!.copyWith(attachments: newAttachments);
        state?.updateMessage(updatedMessage);
        // updating original message for next iteration
        message = message!.merge(updatedMessage);
      }
    }

    return Future.wait(
      attachments.map((it) {
        client.logger.info('Uploading ${it.id} attachment...');

        final throttledUpdateAttachment = updateAttachment.throttled(
          const Duration(milliseconds: 500),
        );

        void onSendProgress(int sent, int total) {
          throttledUpdateAttachment([
            it.copyWith(
              uploadState: UploadState.inProgress(uploaded: sent, total: total),
            ),
          ]);
        }

        final isImage = it.type == AttachmentType.image;
        final cancelToken = CancelToken();
        Future<SendAttachmentResponse> future;
        if (isImage) {
          future = sendImage(
            it.file!,
            onSendProgress: onSendProgress,
            cancelToken: cancelToken,
            extraData: it.extraData,
          );
        } else {
          future = sendFile(
            it.file!,
            onSendProgress: onSendProgress,
            cancelToken: cancelToken,
            extraData: it.extraData,
          );
        }
        _cancelableAttachmentUploadRequest[it.id] = cancelToken;
        return future
            .then((response) {
              client.logger.info('Attachment ${it.id} uploaded successfully...');

              // If the response is SendFileResponse, then we might also be getting
              // thumbUrl in case of video. So we need to update the attachment with
              // both the assetUrl and thumbUrl.
              if (response is SendFileResponse) {
                updateAttachment(
                  it.copyWith(
                    assetUrl: response.file,
                    thumbUrl: response.thumbUrl,
                    uploadState: const UploadState.success(),
                  ),
                );
              } else {
                updateAttachment(
                  it.copyWith(
                    imageUrl: response.file,
                    uploadState: const UploadState.success(),
                  ),
                );
              }
            })
            .catchError((e, stk) {
              if (e is StreamChatNetworkError && e.type == .cancel) {
                client.logger.info('Attachment ${it.id} upload cancelled');

                // remove attachment from message if cancelled.
                updateAttachment(it, remove: true);
                return;
              }

              client.logger.severe('error uploading the attachment', e, stk);
              updateAttachment(
                it.copyWith(uploadState: UploadState.failed(error: e.toString())),
              );
            })
            .whenComplete(() {
              throttledUpdateAttachment.cancel();
              _cancelableAttachmentUploadRequest.remove(it.id);
            });
      }),
    ).whenComplete(() {
      final completer = _messageAttachmentsUploadCompleter.remove(messageId);
      if (completer == null || completer.isCompleted) return;

      // Always complete with the latest message view so callers can decide
      // success vs. partial failure by inspecting per-attachment upload
      // states. Cancellation is still surfaced via `completeError` from the
      // sendMessage/updateMessage/deleteMessage entry points.
      completer.complete(message);
    });
  }

  final _sendMessageLock = Lock();

  /// Send a [message] to this channel.
  ///
  /// If [skipPush] is true the message will not send a push notification.
  ///
  /// Waits for a [_messageAttachmentsUploadCompleter] to complete
  /// before actually sending the message.
  Future<SendMessageResponse> sendMessage(
    Message message, {
    bool skipPush = false,
    bool skipEnrichUrl = false,
  }) async {
    _checkInitialized();

    // Clean up stale error messages before sending a new message.
    state?.cleanUpStaleErrorMessages();

    // Cancelling previous completer in case it's called again in the process
    // Eg. Updating the message while the previous call is in progress.
    _messageAttachmentsUploadCompleter.remove(message.id)?.completeError(const StreamChatError('Message cancelled'));

    final quotedMessage = state!.messages.firstWhereOrNull(
      (m) => m.id == message.quotedMessageId,
    );
    // ignore: parameter_assignments
    message = message.copyWith(
      localCreatedAt: DateTime.now(),
      user: _client.state.currentUser,
      quotedMessage: quotedMessage,
      state: MessageState.sending,
      attachments: message.attachments.map(
        (it) {
          if (it.uploadState.isSuccess) return it;
          return it.copyWith(uploadState: const UploadState.preparing());
        },
      ).toList(),
    );

    state?.updateMessage(message);

    try {
      if (message.attachments.any((it) => !it.uploadState.isSuccess)) {
        final attachmentsUploadCompleter = Completer<Message>();
        _messageAttachmentsUploadCompleter[message.id] = attachmentsUploadCompleter;

        _uploadAttachments(
          message.id,
          message.attachments.map((it) => it.id),
        );

        // ignore: parameter_assignments
        message = await attachmentsUploadCompleter.future;

        // Fail the whole message if any attachment failed to upload
        if (message.attachments.any((it) => it.uploadState.isFailed)) {
          throw const StreamChatError('Failed to upload one or more attachments');
        }
      }

      // Validate the final message before sending it to the server.
      if (MessageRules.canUpload(message) != true) {
        client.logger.warning('Message is not valid for sending, removing it');

        // Remove the message from state as it is invalid.
        state!.deleteMessage(message, hardDelete: true);
        throw const StreamChatError('Message is not valid for sending');
      }

      // 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,
        ),
      );

      final sentMessage = message
          .updateWith(response.message)
          .copyWith(
            // Update the message state to sent.
            state: MessageState.sent,
          );

      state?.updateMessage(sentMessage);

      return response;
    } catch (e) {
      final failedMessage = message.copyWith(
        // Update the message state to failed.
        state: MessageState.sendingFailed(
          skipPush: skipPush,
          skipEnrichUrl: skipEnrichUrl,
        ),
      );

      state?.updateMessage(failedMessage);
      // If the error is retriable, add it to the retry queue.
      if (e is StreamChatNetworkError && e.isRetriable) {
        state?._retryQueue.add([failedMessage]);
      }

      rethrow;
    }
  }

  final _updateMessageLock = Lock();

  /// Updates the [message] in this channel.
  ///
  /// Waits for a [_messageAttachmentsUploadCompleter] to complete
  /// before actually updating the message.
  Future<UpdateMessageResponse> updateMessage(
    Message message, {
    bool skipPush = false,
    bool skipEnrichUrl = false,
  }) async {
    _checkInitialized();

    // Cancelling previous completer in case it's called again in the process
    // Eg. Updating the message while the previous call is in progress.
    _messageAttachmentsUploadCompleter.remove(message.id)?.completeError(const StreamChatError('Message cancelled'));

    // ignore: parameter_assignments
    message = message.copyWith(
      state: MessageState.updating,
      localUpdatedAt: DateTime.now(),
      attachments: message.attachments.map(
        (it) {
          if (it.uploadState.isSuccess) return it;
          return it.copyWith(uploadState: const UploadState.preparing());
        },
      ).toList(),
    );

    state?.updateMessage(message);

    try {
      if (message.attachments.any((it) => !it.uploadState.isSuccess)) {
        final attachmentsUploadCompleter = Completer<Message>();
        _messageAttachmentsUploadCompleter[message.id] = attachmentsUploadCompleter;

        _uploadAttachments(
          message.id,
          message.attachments.map((it) => it.id),
        );

        // ignore: parameter_assignments
        message = await attachmentsUploadCompleter.future;

        // Fail the whole message if any attachment failed to upload
        if (message.attachments.any((it) => it.uploadState.isFailed)) {
          throw const StreamChatError('Failed to upload one or more attachments');
        }
      }

      // 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,
          skipPush: skipPush,
          skipEnrichUrl: skipEnrichUrl,
        ),
      );

      final updateMessage = message
          .updateWith(response.message)
          .copyWith(
            // Update the message state to updated.
            state: MessageState.updated,
          );

      state?.updateMessage(updateMessage);

      return response;
    } catch (e) {
      final failedMessage = message.copyWith(
        // Update the message state to failed.
        state: MessageState.updatingFailed(
          skipPush: skipPush,
          skipEnrichUrl: skipEnrichUrl,
        ),
      );

      state?.updateMessage(failedMessage);
      // If the error is retriable, add it to the retry queue.
      if (e is StreamChatNetworkError && e.isRetriable) {
        state?._retryQueue.add([failedMessage]);
      }

      rethrow;
    }
  }

  /// Partially updates the [message] in this channel.
  ///
  /// Use [set] to define values to be set.
  ///
  /// Use [unset] to define values to be unset.
  Future<UpdateMessageResponse> partialUpdateMessage(
    Message message, {
    Map<String, Object?>? set,
    List<String>? unset,
    bool skipEnrichUrl = false,
  }) async {
    _checkInitialized();

    // Cancelling previous completer in case it's called again in the process
    // Eg. Updating the message while the previous call is in progress.
    _messageAttachmentsUploadCompleter.remove(message.id)?.completeError(const StreamChatError('Message cancelled'));

    // ignore: parameter_assignments
    message = message.copyWith(
      state: MessageState.updating,
      localUpdatedAt: DateTime.now(),
    );

    state?.updateMessage(message);

    try {
      // 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 = message
          .updateWith(response.message)
          .copyWith(
            // Update the message state to updated.
            state: MessageState.updated,
          );

      state?.updateMessage(updatedMessage);

      return response;
    } catch (e) {
      final failedMessage = message.copyWith(
        // Update the message state to failed.
        state: MessageState.partialUpdatingFailed(
          set: set,
          unset: unset,
          skipEnrichUrl: skipEnrichUrl,
        ),
      );

      state?.updateMessage(failedMessage);
      // If the error is retriable, add it to the retry queue.
      if (e is StreamChatNetworkError && e.isRetriable) {
        state?._retryQueue.add([failedMessage]);
      }

      rethrow;
    }
  }

  final _deleteMessageLock = Lock();

  /// Deletes the [message] for everyone.
  ///
  /// If [hard] is true, the message is permanently deleted from the server
  /// and cannot be recovered. In this case, any attachments associated with the
  /// message are also deleted from the server.
  Future<EmptyResponse> deleteMessage(Message message, {bool hard = false}) {
    final deletionScope = MessageDeleteScope.deleteForAll(hard: hard);

    return _deleteMessage(message, scope: deletionScope);
  }

  /// Deletes the [message] only for the current user.
  ///
  /// Note: This does not delete the message for other channel members and
  /// they can still see the message.
  Future<EmptyResponse> deleteMessageForMe(Message message) {
    const deletionScope = MessageDeleteScope.deleteForMe();

    return _deleteMessage(message, scope: deletionScope);
  }

  // Deletes the [message] from the channel.
  //
  // The [scope] defines whether to delete the message for everyone or just
  // for the current user.
  //
  // If the message is a local message (not yet sent to the server) or a bounced
  // error message, it is deleted locally without making an API call.
  //
  // If the message is deleted for everyone and [scope.hard] is true, the
  // message is permanently deleted from the server and cannot be recovered.
  // In this case, any attachments associated with the message are also deleted
  // from the server.
  Future<EmptyResponse> _deleteMessage(
    Message message, {
    required MessageDeleteScope scope,
  }) async {
    _checkInitialized();

    // Directly deleting the local messages and bounced error messages as they
    // are not available on the server.
    if (message.remoteCreatedAt == null || message.isBouncedWithError) {
      _deleteLocalMessage(message);
      // Returning empty response to mark the api call as success.
      return EmptyResponse();
    }

    // ignore: parameter_assignments
    message = message.copyWith(
      type: MessageType.deleted,
      deletedAt: DateTime.now(),
      deletedForMe: scope is DeleteForMe,
      state: MessageState.deleting(scope: scope),
    );

    state?.deleteMessage(message, hardDelete: scope.hard);

    try {
      // Wait for the previous delete call to finish. Otherwise, the order of
      // messages will not be maintained.
      final response = await _deleteMessageLock.synchronized(
        () => switch (scope) {
          DeleteForMe() => _client.deleteMessageForMe(message.id),
          DeleteForAll() => _client.deleteMessage(message.id, hard: scope.hard),
        },
      );

      final deletedMessage = message.copyWith(
        deletedForMe: scope is DeleteForMe,
        state: MessageState.deleted(scope: scope),
      );

      state?.deleteMessage(deletedMessage, hardDelete: scope.hard);
      // If hard delete, also delete the attachments from the server.
      if (scope.hard) _deleteMessageAttachments(deletedMessage);

      return response;
    } catch (e) {
      final failedMessage = message.copyWith(
        // Update the message state to failed.
        state: MessageState.deletingFailed(scope: scope),
      );

      state?.deleteMessage(failedMessage, hardDelete: scope.hard);
      // If the error is retriable, add it to the retry queue.
      if (e is StreamChatNetworkError && e.isRetriable) {
        state?._retryQueue.add([failedMessage]);
      }

      rethrow;
    }
  }

  // Deletes a local [message] that is not yet sent to the server.
  //
  // This is typically called when a user wants to delete a message that they
  // have composed but not yet sent, or if a message failed to send and the user
  // wants to remove it from their local view.
  void _deleteLocalMessage(Message message) {
    state?.deleteMessage(
      hardDelete: true, // Local messages are always hard deleted.
      message.copyWith(
        type: MessageType.deleted,
        localDeletedAt: DateTime.now(),
        state: MessageState.hardDeleted,
      ),
    );

    // Removing the attachments upload completer to stop the `sendMessage`
    // waiting for attachments to complete.
    final completer = _messageAttachmentsUploadCompleter.remove(message.id);
    completer?.completeError(const StreamChatError('Message deleted'));
  }

  // Deletes all the attachments associated with the given [message]
  // from the server. This is typically called when a message is hard deleted.
  Future<void> _deleteMessageAttachments(Message message) async {
    final attachments = message.attachments;
    final deleteFutures = attachments.map((it) async {
      if (it.imageUrl case final url?) return deleteImage(url);
      if (it.assetUrl case final url?) return deleteFile(url);
    });

    try {
      await Future.wait(deleteFutures);
    } catch (e, stk) {
      _client.logger.warning('Error deleting message attachments', e, stk);
    }
  }

  /// Retries operations on a message based on its failed state.
  ///
  /// This method examines the message's state and performs the appropriate
  /// retry action:
  /// - For [MessageState.sendingFailed], it attempts to send the message.
  /// - For [MessageState.updatingFailed], it attempts to update the message.
  /// - For [MessageState.partialUpdatingFailed], it attempts to partially
  ///   update the message with the same 'set' and 'unset' parameters that were
  ///   used in the original request.
  /// - For [MessageState.deletingFailed], it attempts to delete the message
  ///   again, using the same scope (for me or for all) as the original request.
  /// - For messages with [isBouncedWithError], it attempts to send the message.
  ///
  /// Throws a [StateError] if the message is not in a failed state or
  /// bounced with an error.
  Future<Object> retryMessage(Message message) async {
    assert(
      message.state.isFailed || message.isBouncedWithError,
      'Only failed or bounced messages can be retried',
    );

    return message.state.maybeWhen(
      failed: (state, _) => state.when(
        sendingFailed: (skipPush, skipEnrichUrl) => sendMessage(
          message,
          skipPush: skipPush,
          skipEnrichUrl: skipEnrichUrl,
        ),
        updatingFailed: (skipPush, skipEnrichUrl) => updateMessage(
          message,
          skipPush: skipPush,
          skipEnrichUrl: skipEnrichUrl,
        ),
        partialUpdatingFailed: (set, unset, skipEnrichUrl) {
          return partialUpdateMessage(
            message,
            set: set,
            unset: unset,
            skipEnrichUrl: skipEnrichUrl,
          );
        },
        deletingFailed: (scope) => switch (scope) {
          DeleteForMe() => deleteMessageForMe(message),
          DeleteForAll(hard: final hard) => deleteMessage(message, hard: hard),
        },
      ),
      orElse: () {
        // Check if the message is bounced with error.
        if (message.isBouncedWithError) return sendMessage(message);

        throw StateError(
          'Only failed or bounced messages can be retried',
        );
      },
    );
  }

  /// Pins provided message
  Future<UpdateMessageResponse> pinMessage(
    Message message, {
    Object? /*num|DateTime*/ timeoutOrExpirationDate,
  }) {
    assert(() {
      if (timeoutOrExpirationDate is! DateTime && timeoutOrExpirationDate != null && timeoutOrExpirationDate is! num) {
        throw ArgumentError('Invalid timeout or Expiration date');
      }
      return true;
    }(), 'Check for invalid timeout or expiration date');

    DateTime? pinExpires;
    if (timeoutOrExpirationDate is DateTime) {
      pinExpires = timeoutOrExpirationDate;
    } else if (timeoutOrExpirationDate is num) {
      pinExpires = DateTime.now().add(
        Duration(seconds: timeoutOrExpirationDate.toInt()),
      );
    }
    return partialUpdateMessage(
      message,
      set: {
        'pinned': true,
        'pin_expires': pinExpires?.toUtc().toIso8601String(),
      },
    );
  }

  /// Unpins provided message.
  Future<UpdateMessageResponse> unpinMessage(Message message) => partialUpdateMessage(
    message,
    set: {
      'pinned': false,
    },
  );

  /// Creates or updates a new [draft] for this channel.
  Future<CreateDraftResponse> createDraft(
    DraftMessage draft,
  ) {
    _checkInitialized();
    return _client.createDraft(draft, id!, type);
  }

  /// Retrieves the draft for this channel.
  ///
  /// Optionally, provide a [parentId] to get the draft for a specific thread.
  Future<GetDraftResponse> getDraft({
    String? parentId,
  }) {
    _checkInitialized();
    return _client.getDraft(id!, type, parentId: parentId);
  }

  /// Deletes the draft for this channel.
  ///
  /// Optionally, provide a [parentId] to delete the draft for a specific
  /// thread.
  Future<EmptyResponse> deleteDraft({
    String? parentId,
  }) {
    _checkInitialized();
    return _client.deleteDraft(id!, type, parentId: parentId);
  }

  /// Sends a static location to this channel.
  ///
  /// Optionally, provide a [messageText] and [extraData] to send along with
  /// the location.
  Future<SendMessageResponse> sendStaticLocation({
    String? id,
    String? messageText,
    String? createdByDeviceId,
    required LocationCoordinates location,
    Map<String, Object?> extraData = const {},
  }) {
    final message = Message(
      id: id,
      text: messageText,
      extraData: extraData,
    );

    final currentUserId = _client.state.currentUser?.id;
    final locationMessage = message.copyWith(
      sharedLocation: Location(
        channelCid: cid,
        userId: currentUserId,
        messageId: message.id,
        latitude: location.latitude,
        longitude: location.longitude,
        createdByDeviceId: createdByDeviceId,
      ),
    );

    return sendMessage(locationMessage);
  }

  /// Sends a live location sharing message to this channel.
  ///
  /// Optionally, provide a [messageText] and [extraData] to send along with
  /// the location.
  Future<SendMessageResponse> startLiveLocationSharing({
    String? id,
    String? messageText,
    String? createdByDeviceId,
    required DateTime endSharingAt,
    required LocationCoordinates location,
    Map<String, Object?> extraData = const {},
  }) {
    final message = Message(
      id: id,
      text: messageText,
      extraData: extraData,
    );

    final currentUserId = _client.state.currentUser?.id;
    final locationMessage = message.copyWith(
      sharedLocation: Location(
        channelCid: cid,
        userId: currentUserId,
        messageId: message.id,
        endAt: endSharingAt,
        latitude: location.latitude,
        longitude: location.longitude,
        createdByDeviceId: createdByDeviceId,
      ),
    );

    return sendMessage(locationMessage);
  }

  /// Send a file to this channel.
  Future<SendFileResponse> sendFile(
    AttachmentFile file, {
    ProgressCallback? onSendProgress,
    CancelToken? cancelToken,
    Map<String, Object?>? extraData,
  }) {
    _checkInitialized();
    return _client.sendFile(
      file,
      id!,
      type,
      onSendProgress: onSendProgress,
      cancelToken: cancelToken,
      extraData: extraData,
    );
  }

  /// Send an image to this channel.
  Future<SendImageResponse> sendImage(
    AttachmentFile file, {
    ProgressCallback? onSendProgress,
    CancelToken? cancelToken,
    Map<String, Object?>? extraData,
  }) {
    _checkInitialized();
    return _client.sendImage(
      file,
      id!,
      type,
      onSendProgress: onSendProgress,
      cancelToken: cancelToken,
      extraData: extraData,
    );
  }

  /// Search for a message with the given options.
  Future<SearchMessagesResponse> search({
    String? query,
    Filter? messageFilters,
    List<SortOption>? sort,
    PaginationParams? paginationParams,
  }) {
    _checkInitialized();
    return _client.search(
      Filter.in_('cid', [cid!]),
      sort: sort,
      query: query,
      paginationParams: paginationParams,
      messageFilters: messageFilters,
    );
  }

  /// Delete a file from this channel.
  Future<EmptyResponse> deleteFile(
    String url, {
    CancelToken? cancelToken,
    Map<String, Object?>? extraData,
  }) {
    _checkInitialized();
    return _client.deleteFile(
      url,
      id!,
      type,
      cancelToken: cancelToken,
      extraData: extraData,
    );
  }

  /// Delete an image from this channel.
  Future<EmptyResponse> deleteImage(
    String url, {
    CancelToken? cancelToken,
    Map<String, Object?>? extraData,
  }) {
    _checkInitialized();
    return _client.deleteImage(
      url,
      id!,
      type,
      cancelToken: cancelToken,
      extraData: extraData,
    );
  }

  /// Send an event on this channel.
  Future<EmptyResponse> sendEvent(Event event) {
    _checkInitialized();
    return _client.sendEvent(id!, type, event);
  }

  final _pollLock = Lock();

  /// Send a message with a poll to this channel.
  ///
  /// Optionally provide a [messageText] to send a message along with the poll.
  Future<SendMessageResponse> sendPoll(
    Poll poll, {
    String? messageText,
  }) async {
    _checkInitialized();
    final res = await _pollLock.synchronized(() => _client.createPoll(poll));
    return sendMessage(
      Message(
        text: messageText,
        poll: res.poll,
        pollId: res.poll.id,
      ),
    );
  }

  /// Updates the [poll] in this channel.
  Future<UpdatePollResponse> updatePoll(Poll poll) {
    _checkInitialized();
    return _pollLock.synchronized(() => _client.updatePoll(poll));
  }

  /// Deletes the given [poll] from this channel.
  Future<EmptyResponse> deletePoll(Poll poll) {
    _checkInitialized();
    return _pollLock.synchronized(() => _client.deletePoll(poll.id));
  }

  /// Close the given [poll].
  Future<UpdatePollResponse> closePoll(Poll poll) {
    _checkInitialized();
    return _pollLock.synchronized(() => _client.closePoll(poll.id));
  }

  /// Create a new poll option for the given [poll].
  Future<CreatePollOptionResponse> createPollOption(
    Poll poll,
    PollOption option,
  ) {
    _checkInitialized();
    return _pollLock.synchronized(
      () => _client.createPollOption(poll.id, option),
    );
  }

  final _pollVoteLock = Lock();

  /// Cast a vote on the given [poll] with the given [option].
  Future<CastPollVoteResponse> castPollVote(
    Message message,
    Poll poll,
    PollOption option,
  ) async {
    _checkInitialized();

    final optionId = option.id;
    if (optionId == null) {
      throw ArgumentError('Option id cannot be null');
    }

    return _pollVoteLock.synchronized(
      () => _client.castPollVote(
        message.id,
        poll.id,
        optionId: optionId,
      ),
    );
  }

  /// Add a new answer to the given [poll].
  Future<CastPollVoteResponse> addPollAnswer(
    Message message,
    Poll poll, {
    required String answerText,
  }) {
    _checkInitialized();
    return _pollVoteLock.synchronized(
      () => _client.addPollAnswer(
        message.id,
        poll.id,
        answerText: answerText,
      ),
    );
  }

  /// Remove a vote on the given [poll] with the given [vote].
  Future<RemovePollVoteResponse> removePollVote(
    Message message,
    Poll poll,
    PollVote vote,
  ) {
    _checkInitialized();

    final voteId = vote.id;
    if (voteId == null) {
      throw ArgumentError('Vote id cannot be null');
    }

    return _pollVoteLock.synchronized(
      () => _client.removePollVote(
        message.id,
        poll.id,
        voteId,
      ),
    );
  }

  /// Query the poll votes for the given [pollId] with the given [filter] and
  /// [sort] options.
  Future<QueryPollVotesResponse> queryPollVotes(
    String pollId, {
    Filter? filter,
    SortOrder<PollVote>? sort,
    PaginationParams pagination = const PaginationParams(),
  }) {
    _checkInitialized();
    return _client.queryPollVotes(
      pollId,
      filter: filter,
      sort: sort,
      pagination: pagination,
    );
  }

  /// Create a reminder for the given [messageId].
  ///
  /// Optionally, provide a [remindAt] date to set when the reminder should
  /// be triggered. If not provided, the reminder will be created as a
  /// bookmark type instead.
  Future<CreateReminderResponse> createReminder(
    String messageId, {
    DateTime? remindAt,
  }) {
    _checkInitialized();
    return _client.createReminder(
      messageId,
      remindAt: remindAt,
    );
  }

  /// Update an existing reminder with the given [reminderId].
  ///
  /// Optionally, provide a [remindAt] date to set when the reminder should
  /// be triggered. If not provided, the reminder will be updated as a
  /// bookmark type instead.
  Future<UpdateReminderResponse> updateReminder(
    String messageId, {
    DateTime? remindAt,
  }) {
    _checkInitialized();
    return _client.updateReminder(
      messageId,
      remindAt: remindAt,
    );
  }

  /// Remove the reminder for the given [messageId].
  Future<EmptyResponse> deleteReminder(String messageId) {
    _checkInitialized();
    return _client.deleteReminder(messageId);
  }

  /// Send a reaction to this channel.
  ///
  /// Set [enforceUnique] to true to remove the existing user reaction.
  Future<SendReactionResponse> sendReaction(
    Message message,
    Reaction reaction, {
    bool skipPush = false,
    bool enforceUnique = false,
  }) async {
    _checkInitialized();

    final messageId = message.id;
    // ignore: parameter_assignments
    reaction = reaction.copyWith(
      messageId: messageId,
      user: _client.state.currentUser,
    );

    final updatedMessage = message.addMyReaction(
      reaction,
      enforceUnique: enforceUnique,
    );

    state?.updateMessage(updatedMessage);

    try {
      final reactionResp = await _client.sendReaction(
        messageId,
        reaction,
        skipPush: skipPush,
        enforceUnique: enforceUnique,
      );
      return reactionResp;
    } catch (_) {
      // Reset the message if the update fails. Use replace (not merge)
      // so the rollback wins over the optimistic local state — otherwise
      // `Message.updateWith`'s enrichment preservation would keep the
      // optimistic `ownReactions` for messages that previously had none.
      state?.replaceMessage(message);
      rethrow;
    }
  }

  /// Delete a reaction from this channel.
  Future<EmptyResponse> deleteReaction(
    Message message,
    Reaction reaction,
  ) async {
    _checkInitialized();

    final updatedMessage = message.deleteMyReaction(
      reactionType: reaction.type,
    );

    state?.updateMessage(updatedMessage);

    try {
      final deleteResponse = await _client.deleteReaction(
        message.id,
        reaction.type,
      );
      return deleteResponse;
    } catch (_) {
      // Reset the message if the update fails. Use replace (not merge)
      // for symmetry with `sendReaction` — see that method for context.
      state?.replaceMessage(message);
      rethrow;
    }
  }

  /// Sends an event to stop AI response generation, leaving the message in
  /// its current state.
  Future<EmptyResponse> stopAIResponse() async {
    return sendEvent(
      Event(
        type: EventType.aiIndicatorStop,
      ),
    );
  }

  /// Update the channel's [name].
  ///
  /// This is the same as calling [updatePartial] and providing a map with a
  /// 'name' key:
  ///
  /// ```dart
  /// channel.updatePartial(
  ///   set: {'name': 'Updated channel name'}
  /// );
  /// ```
  ///
  /// Instead do:
  /// ```dart
  /// channel.updateName('Updated channel name');
  /// ```
  Future<PartialUpdateChannelResponse> updateName(String name) => updatePartial(set: {'name': name});

  /// Update the channel's [image].
  ///
  /// This is the same as calling [updatePartial] and providing a map with an
  /// 'image' key:
  ///
  /// ```dart
  /// channel.updatePartial(
  ///   set: {'image': 'https://getstream.io/new-image'}
  /// );
  /// ```
  ///
  /// Instead do:
  /// ```dart
  /// channel.updateImage('https://getstream.io/new-image');
  /// ```
  Future<PartialUpdateChannelResponse> updateImage(String image) => updatePartial(set: {'image': image});

  /// Update the channel custom data. This replaces all of the channel data
  /// with the given [channelData].
  ///
  /// If you instead want to do a partial update, use [updatePartial].
  ///
  /// See, https://getstream.io/chat/docs/other-rest/channel_update/?language=dart
  /// for more information.
  Future<UpdateChannelResponse> update(
    Map<String, Object?> channelData, {
    Message? updateMessage,
  }) async {
    _checkInitialized();
    return _client.updateChannel(
      id!,
      type,
      channelData,
      message: updateMessage,
    );
  }

  /// A partial update can be used to set and unset specific custom data fields
  /// when it is necessary to retain additional custom data fields on the
  /// object.
  ///
  /// - [set] will add, or update existing attributes.
  /// - [unset] will remove the attributes with the provided list of
  /// values (keys).
  ///
  /// If you want to do a full update/replacement, use [update] instead.
  ///
  /// See, https://getstream.io/chat/docs/other-rest/channel_update/?language=dart
  /// for more information.
  Future<PartialUpdateChannelResponse> updatePartial({
    Map<String, Object?>? set,
    List<String>? unset,
  }) async {
    _checkInitialized();
    return _client.updateChannelPartial(id!, type, set: set, unset: unset);
  }

  /// Enable slow mode
  Future<PartialUpdateChannelResponse> enableSlowMode({
    required int cooldownInterval,
  }) async {
    _checkInitialized();
    return _client.enableSlowdown(id!, type, cooldownInterval);
  }

  /// Disable slow mode
  Future<PartialUpdateChannelResponse> disableSlowMode() async {
    _checkInitialized();
    return _client.disableSlowdown(id!, type);
  }

  /// Delete this channel. Messages are permanently removed.
  Future<EmptyResponse> delete() async {
    _checkInitialized();
    return _client.deleteChannel(id!, type);
  }

  /// Removes all messages from the channel up to [truncatedAt] or now if
  /// [truncatedAt] is not provided.
  /// If [skipPush] is true, no push notification will be sent.
  /// [Message] is the system message that will be sent to the channel.
  Future<EmptyResponse> truncate({
    Message? message,
    bool? skipPush,
    DateTime? truncatedAt,
  }) async {
    _checkInitialized();
    return _client.truncateChannel(
      id!,
      type,
      message: message,
      skipPush: skipPush,
      truncatedAt: truncatedAt,
    );
  }

  /// Accept invitation to the channel.
  Future<AcceptInviteResponse> acceptInvite([Message? message]) async {
    _checkInitialized();
    return _client.acceptChannelInvite(id!, type, message: message);
  }

  /// Reject invitation to the channel.
  Future<RejectInviteResponse> rejectInvite([Message? message]) async {
    _checkInitialized();
    return _client.rejectChannelInvite(id!, type, message: message);
  }

  /// Add members to the channel.
  Future<AddMembersResponse> addMembers(
    List<String> memberIds, {
    Message? message,
    bool hideHistory = false,
    DateTime? hideHistoryBefore,
  }) async {
    _checkInitialized();
    return _client.addChannelMembers(
      id!,
      type,
      memberIds,
      message: message,
      hideHistory: hideHistory,
      hideHistoryBefore: hideHistoryBefore,
    );
  }

  /// Invite members to the channel.
  Future<InviteMembersResponse> inviteMembers(
    List<String> memberIds, {
    Message? message,
  }) async {
    _checkInitialized();
    return _client.inviteChannelMembers(id!, type, memberIds, message: message);
  }

  /// Remove members from the channel.
  Future<RemoveMembersResponse> removeMembers(
    List<String> memberIds, {
    Message? message,
  }) async {
    _checkInitialized();
    return _client.removeChannelMembers(id!, type, memberIds, message: message);
  }

  /// Send action for a specific message of this channel.
  Future<SendActionResponse> sendAction(
    Message message,
    Map<String, dynamic> formData,
  ) async {
    _checkInitialized();
    final messageId = message.id;
    final res = await _client.sendAction(id!, type, messageId, formData);

    // update the passed message with response message
    if (res.message != null) {
      state!.updateMessage(res.message!);
    } else {
      // remove the passed message if response does
      // not contain message
      state!.removeMessage(message);
    }
    return res;
  }

  /// Mark all messages as read.
  ///
  /// Optionally provide a [messageId] if you want to mark channel as
  /// read from a particular message onwards.
  ///
  /// If [usesLocalUnreadCount] is `true` for this channel, this updates the
  /// unread count locally, on-device, without making a network request. In
  /// that case [messageId] is recorded as the read boundary but does **not**
  /// narrow the count: the channel is always treated as fully read and the
  /// count drops to zero. See [ChannelClientState.markReadLocally].
  Future<EmptyResponse> markRead({String? messageId}) async {
    _checkInitialized();

    if (usesLocalUnreadCount) {
      state!.markReadLocally(messageId: messageId);
      return EmptyResponse();
    }

    if (!canUseReadReceipts) {
      throw const StreamChatError(
        'Cannot mark as read: Channel does not support read events. '
        'Enable read_events in your channel type configuration.',
      );
    }

    return _client.markChannelRead(id!, type, messageId: messageId);
  }

  /// Marks the channel as unread by a given [messageId].
  ///
  /// All messages from the provided message onwards will be marked as unread,
  /// **including** the message itself. Contrast with
  /// [markUnreadByTimestamp], which is exclusive: a message created at
  /// exactly the given timestamp stays read.
  ///
  /// If [usesLocalUnreadCount] is `true` for this channel, this updates the
  /// unread count locally, on-device, without making a network request. The
  /// message must be part of the locally-known messages ([Channel.messages])
  /// for the count to be recomputed.
  Future<EmptyResponse> markUnread(String messageId) async {
    _checkInitialized();

    if (usesLocalUnreadCount) {
      // [ChannelClientState.messages] is sorted ascending by `createdAt`, so
      // the entry before the anchor is the newest message that stays read.
      final messages = state!.messages;
      final anchorIndex = messages.indexWhere((it) => it.id == messageId);
      if (anchorIndex < 0) {
        throw StreamChatError(
          'Cannot mark as unread: Message "$messageId" was not found in the '
          'locally-known messages for this channel.',
        );
      }

      final anchor = messages[anchorIndex];
      final previous = anchorIndex > 0 ? messages[anchorIndex - 1] : null;

      // Subtract a microsecond so the anchor message itself is treated as
      // "after" the new read boundary, matching the "from the provided
      // message onwards" semantics described above. Preferred over using
      // `previous.createdAt` as the boundary, which would leak the anchor
      // back into the read set if the two share an identical `createdAt`.
      final lastRead = anchor.createdAt.subtract(const Duration(microseconds: 1));
      state!.markUnreadLocally(lastRead: lastRead, lastReadMessageId: previous?.id);
      return EmptyResponse();
    }

    if (!canUseReadReceipts) {
      throw const StreamChatError(
        'Cannot mark as unread: Channel does not support read events. '
        'Enable read_events in your channel type configuration.',
      );
    }

    return _client.markChannelUnread(id!, type, messageId);
  }

  /// Marks the channel as unread by a given [timestamp].
  ///
  /// All messages after the provided timestamp will be marked as unread. This
  /// boundary is **exclusive**: a message created at exactly [timestamp] stays
  /// read. Contrast with [markUnread], which is inclusive of the message it is
  /// given — `markUnread(m.id)` is equivalent to
  /// `markUnreadByTimestamp(m.createdAt - 1µs)`, not to
  /// `markUnreadByTimestamp(m.createdAt)`.
  ///
  /// If [usesLocalUnreadCount] is `true` for this channel, this updates the
  /// unread count locally, on-device, without making a network request.
  Future<EmptyResponse> markUnreadByTimestamp(DateTime timestamp) async {
    _checkInitialized();

    if (usesLocalUnreadCount) {
      // The newest locally-known message at or before the boundary is the last
      // one that stays read.
      final lastReadMessage = state!.messages.lastWhereOrNull(
        (it) => !it.createdAt.isAfter(timestamp),
      );

      state!.markUnreadLocally(
        lastRead: timestamp,
        lastReadMessageId: lastReadMessage?.id,
      );
      return EmptyResponse();
    }

    if (!canUseReadReceipts) {
      throw const StreamChatError(
        'Cannot mark as unread: Channel does not support read events. '
        'Enable read_events in your channel type configuration.',
      );
    }

    return _client.markChannelUnreadByTimestamp(id!, type, timestamp);
  }

  /// Mark the thread with [threadId] in the channel as read.
  Future<EmptyResponse> markThreadRead(String threadId) async {
    _checkInitialized();

    if (!canUseReadReceipts) {
      throw const StreamChatError(
        'Cannot mark thread as read: Channel does not support read events. '
        'Enable read_events in your channel type configuration.',
      );
    }

    return _client.markThreadRead(id!, type, threadId);
  }

  /// Mark the thread with [threadId] in the channel as unread.
  Future<EmptyResponse> markThreadUnread(String threadId) async {
    _checkInitialized();

    if (!canUseReadReceipts) {
      throw const StreamChatError(
        'Cannot mark thread as unread: Channel does not support read events. '
        'Enable read_events in your channel type configuration.',
      );
    }

    return _client.markThreadUnread(id!, type, threadId);
  }

  void _initState(ChannelState channelState) {
    state = ChannelClientState(this, channelState);
    _initializedCompleter.safeComplete(true);

    if (cid case final cid?) client.state.addChannels({cid: this});
    _client.logger.info('Channel ${channelState.channel?.cid} initialized');
  }

  /// Loads the initial channel state and watches for changes.
  Future<ChannelState> watch({
    bool presence = false,
    PaginationParams? messagesPagination,
    PaginationParams? membersPagination,
    PaginationParams? watchersPagination,
  }) {
    return query(
      watch: true,
      presence: presence,
      messagesPagination: messagesPagination,
      membersPagination: membersPagination,
      watchersPagination: watchersPagination,
    );
  }

  /// Stop watching the channel.
  Future<EmptyResponse> stopWatching() async {
    _checkInitialized();
    return _client.stopChannelWatching(id!, type);
  }

  /// List the message replies for a parent message.
  ///
  /// Set [preferOffline] to true to avoid the api call if the data is already
  /// in the offline storage.
  Future<QueryRepliesResponse> getReplies(
    String parentId, {
    PaginationParams? options,
    bool preferOffline = false,
  }) async {
    QueryRepliesResponse? response;

    // If we prefer offline, we first try to get the replies from the
    // offline storage.
    if (preferOffline) {
      if (_client.chatPersistenceClient case final persistenceClient?) {
        final cachedReplies = await persistenceClient.getReplies(
          parentId,
          options: options,
        );

        // If the cached replies are not empty, we can use them.
        if (cachedReplies.isNotEmpty) {
          response = QueryRepliesResponse()..messages = cachedReplies;
        }
      }
    }

    // If we still don't have the replies, we try to get them from the API.
    response ??= await _client.getReplies(parentId, options: options);

    // Before updating the state, we check if we are querying around a
    // reply, If we are, we have to clear the state to avoid potential
    // gaps in the message sequence.
    final isQueryingAround = switch (options) {
      PaginationParams(idAround: _?) => true,
      PaginationParams(createdAtAround: _?) => true,
      _ => false,
    };

    if (isQueryingAround) state?.clearThread(parentId);
    state?.updateThreadInfo(parentId, response.messages);

    return response;
  }

  /// List the reactions for a message in the channel.
  Future<QueryReactionsResponse> getReactions(
    String messageId, {
    PaginationParams? pagination,
  }) => _client.getReactions(
    messageId,
    pagination: pagination,
  );

  /// Retrieves a list of messages by given [messageIDs].
  Future<GetMessagesByIdResponse> getMessagesById(
    List<String> messageIDs,
  ) async {
    _checkInitialized();
    return _client.getMessagesById(id!, type, messageIDs);
  }

  /// Translate a message by given [messageId] and [language].
  Future<TranslateMessageResponse> translateMessage(
    String messageId,
    String language,
  ) => _client.translateMessage(
    messageId,
    language,
  );

  /// Creates a new channel.
  Future<ChannelState> create() => query(state: false);

  /// Query the API, get messages, members or other channel fields.
  ///
  /// Set [preferOffline] to true to avoid the API call if the data is already
  /// in the offline storage.
  Future<ChannelState> query({
    bool state = true,
    bool watch = false,
    bool presence = false,
    PaginationParams? messagesPagination,
    PaginationParams? membersPagination,
    PaginationParams? watchersPagination,
    bool preferOffline = false,
  }) async {
    // A prior failed init left the completer errored; reset it so this attempt
    // owns the `initialized` result. Must stay before the first `await` so a
    // caller reading `initialized` right after `query()`/`watch()` begins sees
    // the fresh completer.
    if (_initializedCompleter.isCompleted && this.state == null) {
      _initializedCompleter = Completer<bool>();
    }

    ChannelState? channelState;

    try {
      // If we prefer offline, we first try to get the channel state from the
      // offline storage.
      if (preferOffline && !watch && cid != null) {
        final persistenceClient = _client.chatPersistenceClient;
        if (persistenceClient != null) {
          final cachedState = await persistenceClient.getChannelStateByCid(
            cid!,
            messagePagination: messagesPagination,
          );

          // If the cached state contains messages, we can use it.
          if (cachedState.messages?.isNotEmpty == true) {
            channelState = cachedState;
          }
        }
      }

      // If we still don't have the channelState, we try to get it from the API.
      channelState ??= await _client.queryChannel(
        type,
        channelId: id,
        channelData: _extraData,
        state: state,
        watch: watch,
        presence: presence,
        messagesPagination: messagesPagination,
        membersPagination: membersPagination,
        watchersPagination: watchersPagination,
      );

      if (_id == null) {
        _id = channelState.channel!.id;
        _cid = channelState.channel!.cid;
      }

      // Initialize the channel state if it's not initialized yet.
      if (this.state == null) {
        _initState(channelState);
      } else {
        // Otherwise, we update the existing state with the new channel state.
        //
        // But, before updating the state, we check if we are querying around a
        // message, If we are, we have to truncate the state to avoid potential
        // gaps in the message sequence.
        final isQueryingAround = switch (messagesPagination) {
          PaginationParams(idAround: _?) => true,
          PaginationParams(createdAtAround: _?) => true,
          _ => false,
        };

        if (isQueryingAround) this.state?.truncate();
        this.state?.updateChannelStateFromServer(channelState);
      }

      // Submit for delivery reporting only when fetching the latest messages.
      // This happens when no pagination params are provided (initial query).
      if (messagesPagination == null) {
        _client.channelDeliveryReporter.submitForDelivery([this]);
      }

      return channelState;
    } catch (e, stk) {
      // If we failed to get the channel state from the API and we were not
      // supposed to watch the channel, we will try to get the channel state
      // from the offline storage.
      if (watch == false) {
        if (_client.persistenceEnabled) {
          return _client.chatPersistenceClient!.getChannelStateByCid(
            cid!,
            messagePagination: messagesPagination,
          );
        }
      }

      // Otherwise, we will just rethrow the error.
      _initializedCompleter.safeCompleteError(e, stk);

      rethrow;
    }
  }

  /// Query channel members.
  Future<QueryMembersResponse> queryMembers({
    Filter? filter,
    SortOrder<Member>? sort,
    PaginationParams? pagination,
  }) => _client.queryMembers(
    type,
    channelId: id,
    filter: filter,
    members: state?.members,
    sort: sort,
    pagination: pagination,
  );

  /// Query channel banned users.
  Future<QueryBannedUsersResponse> queryBannedUsers({
    Filter? filter,
    SortOrder<BannedUser>? sort,
    PaginationParams? pagination,
  }) {
    _checkInitialized();
    filter ??= Filter.equal('channel_cid', cid!);
    return _client.queryBannedUsers(
      filter: filter,
      sort: sort,
      pagination: pagination,
    );
  }

  // Timer to keep track of mute expiration. This is used to update the channel
  // state when the mute expires.
  Timer? _muteExpirationTimer;

  /// Mutes the channel.
  Future<EmptyResponse> mute({Duration? expiration}) {
    _checkInitialized();

    // If there is a expiration set, we will set a timer to automatically unmute
    // the channel when the mute expires.
    if (expiration != null) {
      _muteExpirationTimer?.cancel();
      _muteExpirationTimer = Timer(expiration, unmute);
    }

    return _client.muteChannel(cid!, expiration: expiration);
  }

  /// Unmute the channel.
  Future<EmptyResponse> unmute() {
    _checkInitialized();

    // Cancel the mute expiration timer if it is set.
    _muteExpirationTimer?.cancel();
    _muteExpirationTimer = null;

    return _client.unmuteChannel(cid!);
  }

  /// Bans the member with given [userID] from the channel.
  Future<EmptyResponse> banMember(
    String userID,
    Map<String, dynamic> options,
  ) async {
    _checkInitialized();
    final opts = Map<String, dynamic>.from(options)
      ..addAll({
        'type': type,
        'id': id,
      });
    return _client.banUser(userID, opts);
  }

  /// Remove the ban for the member with given [userID] in the channel.
  Future<EmptyResponse> unbanMember(String userID) async {
    _checkInitialized();
    return _client.unbanUser(userID, {
      'type': type,
      'id': id,
    });
  }

  /// Shadow bans the user with the given [userID] from the channel.
  Future<EmptyResponse> shadowBan(
    String userID,
    Map<String, dynamic> options,
  ) async {
    _checkInitialized();
    final opts = Map<String, dynamic>.from(options)
      ..addAll({
        'type': type,
        'id': id,
      });
    return _client.shadowBan(userID, opts);
  }

  /// Remove the shadow ban for the user with the given [userID] in the channel.
  Future<EmptyResponse> removeShadowBan(String userID) async {
    _checkInitialized();
    return _client.removeShadowBan(userID, {
      'type': type,
      'id': id,
    });
  }

  /// Hides the channel from [StreamChatClient.queryChannels] for the user
  /// until a message is added.
  ///
  /// If [clearHistory] is set to true - all messages
  /// will be removed for the user.
  Future<EmptyResponse> hide({bool clearHistory = false}) async {
    _checkInitialized();
    return _client.hideChannel(
      id!,
      type,
      clearHistory: clearHistory,
    );
  }

  /// Removes the hidden status for the channel.
  Future<EmptyResponse> show() async {
    _checkInitialized();
    return _client.showChannel(id!, type);
  }

  /// Pins the channel for the current user.
  Future<Member> pin() async {
    _checkInitialized();

    final response = await _client.pinChannel(
      channelId: id!,
      channelType: type,
    );

    return response.channelMember;
  }

  /// Unpins the channel.
  Future<Member?> unpin() async {
    _checkInitialized();

    final response = await _client.unpinChannel(
      channelId: id!,
      channelType: type,
    );

    return response.channelMember;
  }

  /// Archives the channel.
  Future<Member?> archive() async {
    _checkInitialized();

    final response = await _client.archiveChannel(
      channelId: id!,
      channelType: type,
    );

    return response.channelMember;
  }

  /// Unarchives the channel for the current user.
  Future<Member?> unarchive() async {
    _checkInitialized();

    final response = await _client.unarchiveChannel(
      channelId: id!,
      channelType: type,
    );

    return response.channelMember;
  }

  /// Stream of [Event] coming from websocket connection specific for the
  /// channel. Pass an eventType as parameter in order to filter just a type
  /// of event.
  Stream<Event> on([
    String? eventType,
    String? eventType2,
    String? eventType3,
    String? eventType4,
  ]) => _client
      .on(
        eventType,
        eventType2,
        eventType3,
        eventType4,
      )
      .where((e) => e.cid == cid);

  late final _keyStrokeHandler = KeyStrokeHandler(
    onStartTyping: startTyping,
    onStopTyping: stopTyping,
  );

  // Whether sending typing events is allowed in the channel and by the user
  // privacy settings.
  bool get _canSendTypingEvents {
    final currentUser = client.state.currentUser;
    if (currentUser == null) return false;

    return canUseTypingEvents && currentUser.isTypingIndicatorsEnabled;
  }

  /// Sends the [Event.typingStart] event and schedules a timer to invoke the
  /// [Event.typingStop] event.
  ///
  /// This is meant to be called every time the user presses a key.
  Future<void> keyStroke([String? parentId]) async {
    if (!_canSendTypingEvents) return;

    client.logger.info('KeyStroke received');
    return _keyStrokeHandler(parentId);
  }

  /// Sends the [EventType.typingStart] event.
  Future<void> startTyping([String? parentId]) async {
    if (!_canSendTypingEvents) return;

    client.logger.info('start typing');
    await sendEvent(
      Event(
        type: EventType.typingStart,
        parentId: parentId,
      ),
    );
  }

  /// Sends the [EventType.typingStop] event.
  Future<void> stopTyping([String? parentId]) async {
    if (!_canSendTypingEvents) return;

    client.logger.info('stop typing');
    await sendEvent(
      Event(
        type: EventType.typingStop,
        parentId: parentId,
      ),
    );
  }

  /// Call this method to dispose the channel client.
  void dispose() {
    client.state.removeChannel('$cid');
    state?.dispose();
    state = null;
    _muteExpirationTimer?.cancel();
    _keyStrokeHandler.cancel();
  }

  void _checkInitialized() {
    if (_isInitialized) return;

    throw StateError(
      "Channel $cid hasn't been initialized yet or has been disposed. "
      'Make sure to call .watch() or instantiate the client using '
      '[Channel.fromState]',
    );
  }
}

/// The class that handles the state of the channel listening to the events.
class ChannelClientState {
  /// Creates a new instance listening to events and updating the state.
  ChannelClientState(
    this._channel,
    ChannelState channelState,
  ) {
    _retryQueue = RetryQueue(
      channel: _channel,
      logger: _client.detachedLogger(
        '🔄 (${generateHash([_channel.cid])})',
      ),
    );

    _channelStateController = BehaviorSubject.seeded(channelState);
    // Update the persistence storage with the seeded channel state.
    _debouncedUpdatePersistenceChannelState.call([channelState]);

    // region TYPING EVENTS
    _listenTypingEvents();
    // endregion

    // region MESSAGE EVENTS
    _listenMessageNew();
    _listenMessageDeleted();
    _listenMessageUpdated();
    // endregion

    // region DRAFT EVENTS
    _listenDraftUpdated();
    _listenDraftDeleted();
    // endregion

    // region REACTION EVENTS
    _listenReactionNew();
    _listenReactionUpdated();
    _listenReactionDeleted();
    // endregion

    // region POLL EVENTS
    _listenPollCreated();
    _listenPollUpdated();
    _listenPollClosed();
    _listenPollAnswerCasted();
    _listenPollVoteCasted();
    _listenPollVoteChanged();
    _listenPollAnswerRemoved();
    _listenPollVoteRemoved();
    // endregion

    // region READ EVENTS
    _listenReadEvents();
    // endregion

    // region CHANNEL EVENTS
    _listenChannelTruncated();
    _listenChannelUpdated();
    _listenChannelMessageCount();
    // endregion

    // region MEMBER EVENTS
    _listenMemberAdded();
    _listenMemberRemoved();
    _listenMemberUpdated();
    _listenMemberBanned();
    _listenMemberUnbanned();
    _listenUserMessagesDeleted();
    // endregion

    // region USER WATCHING EVENTS
    _listenUserStartWatching();
    _listenUserStopWatching();
    // endregion

    // region REMINDER EVENTS
    _listenReminderCreated();
    _listenReminderUpdated();
    _listenReminderDeleted();
    // endregion

    // region LOCATION EVENTS
    _listenLocationShared();
    _listenLocationUpdated();
    _listenLocationExpired();
    // endregion

    _startCleaningStaleTypingEvents();

    _startCleaningStalePinnedMessages();

    _startCleaningExpiredLocations();

    _listenChannelPushPreferenceUpdated();

    final persistenceClient = _client.chatPersistenceClient;
    persistenceClient
        ?.getChannelThreads(_channel.cid!)
        .then((threads) {
          // Load all the threads for the channel from the offline storage.
          if (threads.isNotEmpty) _threads = threads;
        })
        .then((_) => retryFailedMessages());
  }

  final Channel _channel;
  StreamChatClient get _client => _channel._client;
  final _subscriptions = CompositeSubscription();

  void _listenMemberAdded() {
    _subscriptions.add(
      _channel.on(EventType.memberAdded).listen((Event e) {
        final member = e.member!;
        final existingMembers = channelState.members ?? [];

        updateChannelState(
          channelState.copyWith(
            members: [...existingMembers, member],
          ),
        );
      }),
    );
  }

  void _listenMemberRemoved() {
    _subscriptions.add(
      _channel.on(EventType.memberRemoved).listen((Event e) {
        final user = e.user!;
        final existingRead = channelState.read ?? [];
        final existingMembers = channelState.members ?? [];

        updateChannelState(
          channelState.copyWith(
            read: [...existingRead.where((r) => r.user.id != user.id)],
            members: [...existingMembers.where((m) => m.userId != user.id)],
          ),
        );
      }),
    );
  }

  void _listenMemberUpdated() {
    _subscriptions
      // Listen to events containing member users
      ..add(
        _channel.on().listen(
          (event) {
            final user = event.user;
            if (user == null) return;

            final existingMembers = [...?channelState.members];
            final existingMembership = channelState.membership;

            // Return if the user is not a existing member of the channel.
            if (!existingMembers.any((m) => m.userId == user.id)) return;

            Member? maybeUpdateMemberUser(Member? existingMember) {
              if (existingMember == null) return null;
              if (existingMember.userId == user.id) {
                return existingMember.copyWith(user: user);
              }
              return existingMember;
            }

            updateChannelState(
              channelState.copyWith(
                membership: maybeUpdateMemberUser(existingMembership),
                members: [...existingMembers.map(maybeUpdateMemberUser).nonNulls],
              ),
            );
          },
        ),
      )
      // Listen to member updated events.
      ..add(
        _channel.on(EventType.memberUpdated).listen(
          (Event e) {
            final member = e.member!;
            final existingMembers = channelState.members ?? [];
            final existingMembership = channelState.membership;

            Member? maybeUpdateMember(Member? existingMember) {
              if (existingMember == null) return null;
              if (existingMember.userId == member.userId) return member;
              return existingMember;
            }

            updateChannelState(
              channelState.copyWith(
                membership: maybeUpdateMember(existingMembership),
                members: [...existingMembers.map(maybeUpdateMember).nonNulls],
              ),
            );
          },
        ),
      );
  }

  void _listenChannelUpdated() {
    _subscriptions.add(
      _channel.on(EventType.channelUpdated).listen((Event e) {
        final channel = e.channel!;
        updateChannelState(
          channelState.copyWith(
            channel: channelState.channel?.merge(channel),
            members: channel.members,
          ),
        );
      }),
    );
  }

  void _listenChannelMessageCount() {
    _subscriptions.add(
      _channel.on().listen(
        (Event e) {
          final messageCount = e.channelMessageCount;
          if (messageCount == null) return;

          updateChannelState(
            channelState.copyWith(
              channel: channelState.channel?.copyWith(
                messageCount: messageCount,
              ),
            ),
          );
        },
      ),
    );
  }

  void _listenChannelTruncated() {
    _subscriptions.add(
      _channel.on(EventType.channelTruncated, EventType.notificationChannelTruncated).listen((event) async {
        final channel = event.channel!;
        await _client.chatPersistenceClient?.deleteMessageByCid(channel.cid);
        truncate();
        if (event.message != null) {
          updateMessage(event.message!);
        }
      }),
    );
  }

  void _listenMemberBanned() {
    _subscriptions.add(
      _channel
          .on(EventType.userBanned)
          .where((it) => it.cid != null) // filters channel ban from app ban
          .listen(
            (event) async {
              final user = event.user!;
              final member = await _channel
                  .queryMembers(filter: Filter.equal('id', user.id))
                  .then((it) => it.members.first);

              _updateMember(member);
            },
          ),
    );
  }

  void _listenUserStartWatching() {
    _subscriptions.add(
      _channel.on(EventType.userWatchingStart).listen((event) {
        final watcher = event.user;
        if (watcher != null) {
          final existingWatchers = channelState.watchers;
          updateChannelState(
            channelState.copyWith(
              watchers: [
                watcher,
                ...?existingWatchers?.where((user) => user.id != watcher.id),
              ],
              watcherCount: event.watcherCount,
            ),
          );
        }
      }),
    );
  }

  void _listenUserStopWatching() {
    _subscriptions.add(
      _channel.on(EventType.userWatchingStop).listen((event) {
        final watcher = event.user;
        if (watcher != null) {
          final existingWatchers = channelState.watchers ?? const <User>[];
          _channelState = channelState.copyWith(
            watchers: existingWatchers.where((user) => user.id != watcher.id).toList(),
            watcherCount: event.watcherCount,
          );
        }
      }),
    );
  }

  void _listenMemberUnbanned() {
    _subscriptions.add(
      _channel
          .on(EventType.userUnbanned)
          .where((it) => it.cid != null) // filters channel ban from app ban
          .listen(
            (event) async {
              final user = event.user!;
              final member = await _channel
                  .queryMembers(filter: Filter.equal('id', user.id))
                  .then((it) => it.members.first);

              _updateMember(member);
            },
          ),
    );
  }

  void _updateMember(Member member) {
    final currentMembers = [...members];
    final memberIndex = currentMembers.indexWhere(
      (m) => m.userId == member.userId,
    );

    if (memberIndex == -1) return;
    currentMembers[memberIndex] = member;

    updateChannelState(
      channelState.copyWith(
        members: currentMembers,
      ),
    );
  }

  /// Flag which indicates if [ChannelClientState] contain latest/recent messages or not.
  ///
  /// This flag should be managed by UI sdks.
  ///
  /// When false, any new message received by WebSocket event
  /// [EventType.messageNew] will not be pushed on to message list.
  bool get isUpToDate => _isUpToDateController.value;

  set isUpToDate(bool isUpToDate) => _isUpToDateController.safeAdd(isUpToDate);

  /// [isUpToDate] flag count as a stream.
  Stream<bool> get isUpToDateStream => _isUpToDateController.stream;
  final _isUpToDateController = BehaviorSubject.seeded(true);

  /// The retry queue associated to this channel.
  late final RetryQueue _retryQueue;

  /// Retry failed message.
  Future<void> retryFailedMessages() async {
    final allMessages = [...messages, ...threads.values.flattened];
    final failedMessages = allMessages.where((it) => it.state.isFailed);

    if (failedMessages.isEmpty) return;
    _retryQueue.add(failedMessages);
  }

  Message? _findPollMessage(String pollId) {
    final message = messages.firstWhereOrNull((it) => it.pollId == pollId);
    if (message != null) return message;

    final threadMessage = threads.values.flattened.firstWhereOrNull((it) {
      return it.pollId == pollId;
    });

    return threadMessage;
  }

  void _listenPollCreated() {
    _subscriptions.add(
      _channel.on(EventType.pollCreated).listen((event) {
        final message = event.message;
        if (message == null || message.poll == null) return;

        return addNewMessage(message);
      }),
    );
  }

  void _listenPollUpdated() {
    _subscriptions.add(
      _channel.on(EventType.pollUpdated).listen((event) {
        final eventPoll = event.poll;
        if (eventPoll == null) return;

        final pollMessage = _findPollMessage(eventPoll.id);
        if (pollMessage == null) return;

        final oldPoll = pollMessage.poll;

        final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers;
        final ownVotesAndAnswers = oldPoll?.ownVotesAndAnswers ?? eventPoll.ownVotesAndAnswers;

        final poll = eventPoll.copyWith(
          latestAnswers: latestAnswers,
          ownVotesAndAnswers: ownVotesAndAnswers,
        );

        final message = pollMessage.copyWith(poll: poll);
        updateMessage(message);
      }),
    );
  }

  void _listenPollClosed() {
    _subscriptions.add(
      _channel.on(EventType.pollClosed).listen((event) {
        final eventPoll = event.poll;
        if (eventPoll == null) return;

        final pollMessage = _findPollMessage(eventPoll.id);
        if (pollMessage == null) return;

        final oldPoll = pollMessage.poll;
        final poll = oldPoll?.copyWith(isClosed: true) ?? eventPoll;

        final message = pollMessage.copyWith(poll: poll);
        updateMessage(message);
      }),
    );
  }

  void _listenPollAnswerCasted() {
    _subscriptions.add(
      _channel.on(EventType.pollAnswerCasted).listen((event) {
        final (eventPoll, eventPollVote) = (event.poll, event.pollVote);
        if (eventPoll == null || eventPollVote == null) return;

        final pollMessage = _findPollMessage(eventPoll.id);
        if (pollMessage == null) return;

        final oldPoll = pollMessage.poll;

        final latestAnswers = <String, PollVote>{
          for (final ans in oldPoll?.latestAnswers ?? []) ans.id: ans,
          eventPollVote.id!: eventPollVote,
        };

        final currentUserId = _client.state.currentUser?.id;
        final ownVotesAndAnswers = <String, PollVote>{
          for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote,
          if (eventPollVote.userId == currentUserId) eventPollVote.id!: eventPollVote,
        };

        final poll = eventPoll.copyWith(
          latestAnswers: [...latestAnswers.values],
          ownVotesAndAnswers: [...ownVotesAndAnswers.values],
        );

        final message = pollMessage.copyWith(poll: poll);
        updateMessage(message);
      }),
    );
  }

  void _listenPollVoteCasted() {
    _subscriptions.add(
      _channel.on(EventType.pollVoteCasted).listen((event) {
        final (eventPoll, eventPollVote) = (event.poll, event.pollVote);
        if (eventPoll == null || eventPollVote == null) return;

        final pollMessage = _findPollMessage(eventPoll.id);
        if (pollMessage == null) return;

        final oldPoll = pollMessage.poll;

        final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers;
        final currentUserId = _client.state.currentUser?.id;
        final ownVotesAndAnswers = <String, PollVote>{
          for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote,
          if (eventPollVote.userId == currentUserId) eventPollVote.id!: eventPollVote,
        };

        final poll = eventPoll.copyWith(
          latestAnswers: latestAnswers,
          ownVotesAndAnswers: [...ownVotesAndAnswers.values],
        );

        final message = pollMessage.copyWith(poll: poll);
        updateMessage(message);
      }),
    );
  }

  void _listenPollAnswerRemoved() {
    _subscriptions.add(
      _channel.on(EventType.pollAnswerRemoved).listen((event) {
        final (eventPoll, eventPollVote) = (event.poll, event.pollVote);
        if (eventPoll == null || eventPollVote == null) return;

        final pollMessage = _findPollMessage(eventPoll.id);
        if (pollMessage == null) return;

        final oldPoll = pollMessage.poll;

        final latestAnswers = <String, PollVote>{
          for (final ans in oldPoll?.latestAnswers ?? []) ans.id: ans,
        }..remove(eventPollVote.id);

        final ownVotesAndAnswers = <String, PollVote>{
          for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote,
        }..remove(eventPollVote.id);

        final poll = eventPoll.copyWith(
          latestAnswers: [...latestAnswers.values],
          ownVotesAndAnswers: [...ownVotesAndAnswers.values],
        );

        final message = pollMessage.copyWith(poll: poll);
        updateMessage(message);
      }),
    );
  }

  void _listenPollVoteRemoved() {
    _subscriptions.add(
      _channel.on(EventType.pollVoteRemoved).listen((event) {
        final (eventPoll, eventPollVote) = (event.poll, event.pollVote);
        if (eventPoll == null || eventPollVote == null) return;

        final pollMessage = _findPollMessage(eventPoll.id);
        if (pollMessage == null) return;

        final oldPoll = pollMessage.poll;

        final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers;
        final ownVotesAndAnswers = <String, PollVote>{
          for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote,
        }..remove(eventPollVote.id);

        final poll = eventPoll.copyWith(
          latestAnswers: latestAnswers,
          ownVotesAndAnswers: [...ownVotesAndAnswers.values],
        );

        final message = pollMessage.copyWith(poll: poll);
        updateMessage(message);
      }),
    );
  }

  void _listenPollVoteChanged() {
    _subscriptions.add(
      _channel.on(EventType.pollVoteChanged).listen((event) {
        final (eventPoll, eventPollVote) = (event.poll, event.pollVote);
        if (eventPoll == null || eventPollVote == null) return;

        final pollMessage = _findPollMessage(eventPoll.id);
        if (pollMessage == null) return;

        final oldPoll = pollMessage.poll;

        final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers;
        final currentUserId = _client.state.currentUser?.id;
        final ownVotesAndAnswers = <String, PollVote>{
          for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote,
          if (eventPollVote.userId == currentUserId) eventPollVote.id!: eventPollVote,
        };

        final poll = eventPoll.copyWith(
          latestAnswers: latestAnswers,
          ownVotesAndAnswers: [...ownVotesAndAnswers.values],
        );

        final message = pollMessage.copyWith(poll: poll);
        updateMessage(message);
      }),
    );
  }

  void _listenDraftUpdated() {
    _subscriptions.add(
      _channel.on(EventType.draftUpdated).listen((event) {
        final draft = event.draft;
        if (draft == null) return;

        return updateDraft(draft);
      }),
    );
  }

  void _listenDraftDeleted() {
    _subscriptions.add(
      _channel.on(EventType.draftDeleted).listen((event) {
        final draft = event.draft;
        if (draft == null) return;

        return deleteDraft(draft);
      }),
    );
  }

  void _listenReminderCreated() {
    _subscriptions.add(
      _channel.on(EventType.reminderCreated).listen((event) {
        final reminder = event.reminder;
        if (reminder == null) return;

        updateReminder(reminder);
      }),
    );
  }

  void _listenReminderUpdated() {
    _subscriptions.add(
      _channel.on(EventType.reminderUpdated).listen((event) {
        final reminder = event.reminder;
        if (reminder == null) return;

        updateReminder(reminder);
      }),
    );
  }

  void _listenReminderDeleted() {
    _subscriptions.add(
      _channel.on(EventType.reminderDeleted).listen((event) {
        final reminder = event.reminder;
        if (reminder == null) return;

        deleteReminder(reminder);
      }),
    );
  }

  /// Updates the [reminder] of the message if it exists.
  void updateReminder(MessageReminder reminder) {
    final messageId = reminder.messageId;
    // TODO: Improve once we have support for parentId in reminders.
    for (final message in [...messages, ...threads.values.flattened]) {
      if (message.id == messageId) {
        return updateMessage(
          message.copyWith(reminder: reminder),
        );
      }
    }
  }

  /// Deletes the [reminder] of the message if it exists.
  void deleteReminder(MessageReminder reminder) {
    final messageId = reminder.messageId;
    // TODO: Improve once we have support for parentId in reminders.
    for (final message in [...messages, ...threads.values.flattened]) {
      if (message.id == messageId) {
        return updateMessage(
          message.copyWith(reminder: null),
        );
      }
    }
  }

  Message? _findLocationMessage(String id) {
    final message = messages.firstWhereOrNull((it) {
      return it.sharedLocation?.messageId == id;
    });

    if (message != null) return message;

    final threadMessage = threads.values.flattened.firstWhereOrNull((it) {
      return it.sharedLocation?.messageId == id;
    });

    return threadMessage;
  }

  void _listenLocationShared() {
    _subscriptions.add(
      _channel.on(EventType.locationShared).listen((event) {
        final message = event.message;
        if (message == null || message.sharedLocation == null) return;

        return addNewMessage(message);
      }),
    );
  }

  void _listenLocationUpdated() {
    _subscriptions.add(
      _channel.on(EventType.locationUpdated).listen((event) {
        final location = event.message?.sharedLocation;
        if (location == null) return;

        final messageId = location.messageId;
        if (messageId == null) return;

        final oldMessage = _findLocationMessage(messageId);
        if (oldMessage == null) return;

        final updatedMessage = oldMessage.copyWith(sharedLocation: location);
        return updateMessage(updatedMessage);
      }),
    );
  }

  void _listenLocationExpired() {
    _subscriptions.add(
      _channel.on(EventType.locationExpired).listen((event) {
        final location = event.message?.sharedLocation;
        if (location == null) return;

        final messageId = location.messageId;
        if (messageId == null) return;

        final oldMessage = _findLocationMessage(messageId);
        if (oldMessage == null) return;

        final updatedMessage = oldMessage.copyWith(sharedLocation: location);
        return updateMessage(updatedMessage);
      }),
    );
  }

  void _listenReactionDeleted() {
    _subscriptions.add(
      _channel.on(EventType.reactionDeleted).listen((event) {
        final (eventReaction, eventMessage) = (event.reaction, event.message);
        if (eventReaction == null || eventMessage == null) return;

        final messageId = eventMessage.id;
        final parentId = eventMessage.parentId;

        for (final message in [...messages, ...?threads[parentId]]) {
          if (message.id == messageId) {
            final currentUserId = _channel.client.state.currentUser?.id;

            final currentMessage = switch (currentUserId) {
              final userId? when userId == eventReaction.userId => message.deleteMyReaction(
                reactionType: eventReaction.type,
              ),
              _ => message,
            };

            return updateMessage(
              eventMessage.copyWith(
                ownReactions: currentMessage.ownReactions,
              ),
            );
          }
        }
      }),
    );
  }

  void _listenReactionNew() {
    _subscriptions.add(
      _channel.on(EventType.reactionNew).listen((event) {
        final (eventReaction, eventMessage) = (event.reaction, event.message);
        if (eventReaction == null || eventMessage == null) return;

        final messageId = eventMessage.id;
        final parentId = eventMessage.parentId;

        for (final message in [...messages, ...?threads[parentId]]) {
          if (message.id == messageId) {
            final currentUserId = _channel.client.state.currentUser?.id;

            final currentMessage = switch (currentUserId) {
              final userId? when userId == eventReaction.userId => message.addMyReaction(eventReaction),
              _ => message,
            };

            return updateMessage(
              eventMessage.copyWith(
                ownReactions: currentMessage.ownReactions,
              ),
            );
          }
        }
      }),
    );
  }

  void _listenReactionUpdated() {
    _subscriptions.add(
      _channel.on(EventType.reactionUpdated).listen((event) {
        final (eventReaction, eventMessage) = (event.reaction, event.message);
        if (eventReaction == null || eventMessage == null) return;

        final messageId = eventMessage.id;
        final parentId = eventMessage.parentId;

        for (final message in [...messages, ...?threads[parentId]]) {
          if (message.id == messageId) {
            final currentUserId = _channel.client.state.currentUser?.id;

            final currentMessage = switch (currentUserId) {
              final userId? when userId == eventReaction.userId =>
                // reaction.updated is only called if enforce_unique is true
                message.addMyReaction(eventReaction, enforceUnique: true),
              _ => message,
            };

            return updateMessage(
              eventMessage.copyWith(
                ownReactions: currentMessage.ownReactions,
              ),
            );
          }
        }
      }),
    );
  }

  void _listenMessageUpdated() {
    _subscriptions.add(
      _channel.on(EventType.messageUpdated).listen((event) {
        final message = event.message;
        if (message == null) return;

        return updateMessage(message, upsert: false);
      }),
    );
  }

  void _listenMessageDeleted() {
    _subscriptions.add(
      _channel.on(EventType.messageDeleted).listen((event) {
        final hardDelete = event.hardDelete ?? false;

        final message = event.message!.copyWith(
          // TODO: Remove once deletedForMe is properly enriched on the backend.
          deletedForMe: event.deletedForMe,
        );

        // Decrement the locally-tracked unread count for hard-deleted
        // messages that would have counted as unread. Soft-deleted messages
        // keep their slot. Only applies to channels that track unread counts
        // locally (see [Channel.usesLocalUnreadCount]) — server-driven
        // channels get corrected counts from server read events instead.
        if (hardDelete && _channel.usesLocalUnreadCount && MessageRules.canCountAsUnread(message, _channel)) {
          unreadCount = math.max(0, unreadCount - 1);
        }

        return deleteMessage(message, hardDelete: hardDelete);
      }),
    );
  }

  void _listenMessageNew() {
    _subscriptions.add(
      _channel
          .on(
            EventType.messageNew,
            EventType.notificationMessageNew,
          )
          .listen((event) {
            final message = event.message;
            if (message == null) return;

            addNewMessage(message);

            // Only message.new carries a reliable watcher count;
            // notification.message_new targets non-watchers and reports 0.
            if (event.watcherCount case final watcherCount? when event.type == EventType.messageNew) {
              updateChannelState(
                channelState.copyWith(watcherCount: watcherCount),
              );
            }
          }),
    );
  }

  /// Adds a new message to the channel state and updates the unread count.
  void addNewMessage(Message message) {
    final isThreadMessage = message.parentId != null;
    final isNotShownInChannel = message.showInChannel != true;
    final isThreadOnlyMessage = isThreadMessage && isNotShownInChannel;

    // Only add the message if the channel is upToDate or if the message is
    // a thread-only message.
    if (isUpToDate || isThreadOnlyMessage) updateMessage(message);

    // Otherwise, check if we can count the message as unread.
    if (MessageRules.canCountAsUnread(message, _channel)) {
      unreadCount += 1; // Increment unread count
    }

    _client.channelDeliveryReporter.submitForDelivery([_channel]);
  }

  /// Updates the [read] in the state if it exists. Adds it otherwise.
  void updateRead([Iterable<Read>? read]) {
    final existingReads = channelState.read ?? const <Read>[];
    final updatedReads = existingReads.merge(
      read,
      key: (read) => read.user.id,
    );

    updateChannelState(
      channelState.copyWith(
        read: updatedReads.toList(),
      ),
    );
  }

  /// Updates the [draft] in the channel state or the message if it exists.
  void updateDraft(Draft draft) {
    if (draft.parentId case final parentId?) {
      for (final message in messages) {
        if (message.id == parentId) {
          return updateMessage(message.copyWith(draft: draft));
        }
      }
    }

    updateChannelState(
      channelState.copyWith(
        draft: draft,
      ),
    );
  }

  /// Deletes the [draft] from the state if it exists.
  void deleteDraft(Draft draft) async {
    // Delete the draft from the persistence client.
    await _client.chatPersistenceClient?.deleteDraftMessageByCid(
      draft.channelCid,
      parentId: draft.parentId,
    );

    if (draft.parentId case final parentId?) {
      for (final message in messages) {
        if (message.id == parentId) {
          return updateMessage(
            message.copyWith(draft: null),
          );
        }
      }
    }

    updateChannelState(
      channelState.copyWith(
        draft: null,
      ),
    );
  }

  /// Updates the [message] in the state.
  ///
  /// Reconciles via `Message.updateWith`, so locally-known enrichment
  /// (poll, sharedLocation, ownReactions, nested quotedMessage) is
  /// preserved when [message] omits those fields. Use [replaceMessage]
  /// for paths that need a strict overwrite.
  ///
  /// When [upsert] is `true` (the default) and [message] isn't already in
  /// the state, it's added. When `false`, an unknown [message] is skipped
  /// and the state is left unchanged; only a message already loaded in the
  /// state is updated.
  void updateMessage(Message message, {bool upsert = true}) => _updateMessages([message], upsert: upsert);

  /// Replaces the [message] in the state if it exists, no-op otherwise.
  ///
  /// Unlike [updateMessage], this does **not** merge with the existing
  /// state — [message] is used as-is. Useful for local rollbacks of an
  /// optimistic update, where the caller has the full prior snapshot and
  /// doesn't want the merge falling back to the optimistic values.
  void replaceMessage(Message message) => _updateMessages([message], update: _replaceUpdate);

  // Default `update` for [_updateMessages]: merge incoming with the
  // locally-known message via `Message.updateWith`, preserving enrichment
  // the server may strip on partial payloads.
  static Message _mergeUpdate(Message original, Message updated) => original.updateWith(updated);

  // Replace `update` for [_updateMessages]: take the incoming as-is. Used
  // by local rollback paths.
  static Message _replaceUpdate(Message _, Message updated) => updated;

  /// Cleans up all the stale error messages which requires no action.
  void cleanUpStaleErrorMessages() {
    final errorMessages = messages.where((message) {
      return message.isError && !message.isBounced;
    });

    if (errorMessages.isEmpty) return;
    return _removeMessages(errorMessages);
  }

  /// Remove a [message] from this [channelState].
  void removeMessage(Message message) => _removeMessages([message]);

  /// Removes/Updates the [message] based on the [hardDelete] value.
  void deleteMessage(Message message, {bool hardDelete = false}) {
    return _deleteMessages([message], hardDelete: hardDelete);
  }

  void _listenReadEvents() {
    _subscriptions
      ..add(
        _channel.on(EventType.messageRead).listen(
          (event) {
            // Skip handling the event if delivered for a thread
            if (event.thread != null) return;

            final user = event.user;
            if (user == null) return;

            final currentRead = userReadOf(userId: user.id);

            final updatedRead = Read(
              user: user,
              lastRead: event.createdAt,
              unreadMessages: 0, // Reset unread count
              lastReadMessageId: event.lastReadMessageId,
              // Preserve delivery info as it's not part of the read event.
              lastDeliveredAt: currentRead?.lastDeliveredAt,
              lastDeliveredMessageId: currentRead?.lastDeliveredMessageId,
            );

            updateRead([updatedRead]);

            // If the read event is from the current user, reconcile the
            // channel delivery status with the updated read state, and clear
            // any pending manual mark-unread — the user has read past it.
            final currentUser = _client.state.currentUser;
            if (event.isFromUser(userId: currentUser?.id)) {
              _client.channelDeliveryReporter.reconcileDelivery([_channel]);
              _isMarkedAsUnread = false;
            }
          },
        ),
      )
      ..add(
        _channel.on(EventType.notificationMarkUnread).listen(
          (event) {
            final user = event.user;
            if (user == null) return;

            final currentRead = userReadOf(userId: user.id);

            final updatedRead = Read(
              user: user,
              lastRead: event.lastReadAt!,
              unreadMessages: event.unreadMessages,
              lastReadMessageId: event.lastReadMessageId,
              // Preserve delivery info as it's not part of the read event.
              lastDeliveredAt: currentRead?.lastDeliveredAt,
              lastDeliveredMessageId: currentRead?.lastDeliveredMessageId,
            );

            updateRead([updatedRead]);

            // Only a mark-unread for the current user's own read state
            // should gate this device's auto mark-read.
            final currentUser = _client.state.currentUser;
            if (event.isFromUser(userId: currentUser?.id)) {
              _isMarkedAsUnread = true;
            }
          },
        ),
      )
      ..add(
        _channel.on(EventType.messageDelivered).listen(
          (event) {
            final user = event.user;
            if (user == null) return;

            final currentRead = userReadOf(userId: user.id);
            final never = DateTime.fromMillisecondsSinceEpoch(0, isUtc: true);

            final updatedRead = Read(
              user: user,
              lastDeliveredAt: event.lastDeliveredAt,
              lastDeliveredMessageId: event.lastDeliveredMessageId,
              // Preserve read info as it's not part of the delivery event.
              lastRead: currentRead?.lastRead ?? never,
              unreadMessages: currentRead?.unreadMessages,
              lastReadMessageId: currentRead?.lastReadMessageId,
            );

            updateRead([updatedRead]);

            // If the delivered event is from the current user, reconcile
            // the channel delivery with the updated read state.
            final currentUser = _client.state.currentUser;
            if (event.isFromUser(userId: currentUser?.id)) {
              _client.channelDeliveryReporter.reconcileDelivery([_channel]);
            }
          },
        ),
      );
  }

  /// Channel message list.
  List<Message> get messages => _channelState.messages ?? <Message>[];

  /// Channel message list as a stream.
  Stream<List<Message>> get messagesStream =>
      channelStateStream.map((cs) => cs.messages ?? <Message>[]).distinct(const ListEquality().equals);

  /// Channel pinned message list.
  List<Message> get pinnedMessages => _channelState.pinnedMessages ?? <Message>[];

  /// Channel pinned message list as a stream.
  Stream<List<Message>> get pinnedMessagesStream =>
      channelStateStream.map((cs) => cs.pinnedMessages ?? <Message>[]).distinct(const ListEquality().equals);

  /// Channel pending message list.
  List<Message> get pendingMessages => _channelState.pendingMessages ?? <Message>[];

  /// Channel pending message list as a stream.
  Stream<List<Message>> get pendingMessagesStream =>
      channelStateStream.map((cs) => cs.pendingMessages ?? <Message>[]).distinct(const ListEquality().equals);

  /// Get channel last message.
  Message? get lastMessage => messages.lastOrNull;

  /// Get channel last message as a stream.
  Stream<Message?> get lastMessageStream {
    return messagesStream.map((messages) => messages.lastOrNull);
  }

  /// Channel members list.
  List<Member> get members =>
      (_channelState.members ?? <Member>[]).map((e) => e.copyWith(user: _client.state.users[e.user!.id])).toList();

  /// Channel members list as a stream.
  Stream<List<Member>> get membersStream =>
      CombineLatestStream.combine2<List<Member?>?, Map<String?, User?>, List<Member>>(
        channelStateStream.map((cs) => cs.members),
        _client.state.usersStream,
        (members, users) => [...?members?.map((e) => e!.copyWith(user: users[e.user!.id]))],
      ).distinct(const ListEquality().equals);

  /// Channel watcher count.
  int? get watcherCount => _channelState.watcherCount;

  /// Channel watcher count as a stream.
  Stream<int?> get watcherCountStream => channelStateStream.map((cs) => cs.watcherCount);

  /// Channel watchers list.
  List<User> get watchers => (_channelState.watchers ?? <User>[]).map((e) => _client.state.users[e.id] ?? e).toList();

  /// Channel watchers list as a stream.
  Stream<List<User>> get watchersStream => CombineLatestStream.combine2<List<User>?, Map<String?, User?>, List<User>>(
    channelStateStream.map((cs) => cs.watchers),
    _client.state.usersStream,
    (watchers, users) => [...?watchers?.map((e) => users[e.id] ?? e)],
  ).distinct(const ListEquality().equals);

  /// Channel active live locations.
  List<Location> get activeLiveLocations {
    return _channelState.activeLiveLocations ?? <Location>[];
  }

  /// Channel active live locations as a stream.
  Stream<List<Location>> get activeLiveLocationsStream =>
      channelStateStream.map((cs) => cs.activeLiveLocations ?? <Location>[]).distinct(const ListEquality().equals);

  /// Channel draft.
  Draft? get draft => _channelState.draft;

  /// Channel draft as a stream.
  Stream<Draft?> get draftStream {
    return channelStateStream.map((cs) => cs.draft).distinct();
  }

  /// Channel member for the current user.
  Member? get currentUserMember => members.firstWhereOrNull(
    (m) => m.user?.id == _client.state.currentUser?.id,
  );

  /// Channel role for the current user
  String? get currentUserChannelRole => currentUserMember?.channelRole;

  /// Channel read list.
  List<Read> get read => _channelState.read ?? <Read>[];

  /// Channel read list as a stream.
  Stream<List<Read>> get readStream =>
      channelStateStream.map((cs) => cs.read ?? <Read>[]).distinct(const ListEquality().equals);

  /// Channel read for the logged in user.
  Read? get currentUserRead {
    final currentUser = _client.state.currentUser;
    return userReadOf(userId: currentUser?.id);
  }

  /// Channel read for the logged in user as a stream.
  ///
  /// Re-subscribes only when the user id actually changes; null still
  /// propagates downstream so consumers see the logged-out transition.
  Stream<Read?> get currentUserReadStream {
    final currentUserId = _client.state.currentUserStream.map((it) => it?.id).distinct();
    return currentUserId.switchMap((id) => userReadStreamOf(userId: id)).distinct();
  }

  /// Unread count getter as a stream.
  Stream<int> get unreadCountStream => currentUserReadStream.map((read) => read?.unreadMessages ?? 0).distinct();

  /// Unread count getter.
  int get unreadCount => currentUserRead?.unreadMessages ?? 0;

  /// Setter for unread count.
  set unreadCount(int count) {
    final currentUser = _client.state.currentUser;
    if (currentUser == null) return;

    var existingUserRead = currentUserRead;
    if (existingUserRead == null) {
      final lastMessageAt = _channelState.channel?.lastMessageAt;
      existingUserRead = Read(
        user: currentUser,
        lastRead: lastMessageAt ?? DateTime.now(),
      );
    }

    return updateRead([existingUserRead.copyWith(unreadMessages: count)]);
  }

  /// Whether the current user explicitly marked a message in this channel as
  /// unread during this session, without having read past that boundary
  /// since.
  ///
  /// Set by [markUnreadLocally] and by a `notification.mark_unread` event for
  /// the current user; cleared by [markReadLocally] and by a `message.read`
  /// event for the current user. Intended for UI-layer gating that shouldn't
  /// immediately undo a manual mark-unread — mirrors the iOS SDK's
  /// `ReadStateHandler.isMarkedAsUnread`.
  bool get isMarkedAsUnread => _isMarkedAsUnread;
  bool _isMarkedAsUnread = false;

  /// Marks the channel as read locally, without making a network request.
  ///
  /// Used for channels that track unread counts locally (see
  /// [Channel.usesLocalUnreadCount]), since the server rejects the mark-read
  /// endpoint for channels that have read events disabled.
  ///
  /// [messageId] only sets the resulting [Read.lastReadMessageId]; it does not
  /// narrow which messages stay unread. The count always drops to zero and
  /// [Read.lastRead] is always `now`, so messages newer than [messageId] are
  /// marked read as well. This differs from the server, which recomputes the
  /// count as the number of messages after [messageId], and from
  /// [markUnreadLocally], which does recompute from the locally-known
  /// messages. Callers that need a partial boundary should use
  /// [markUnreadLocally] instead.
  void markReadLocally({String? messageId}) {
    final currentUser = _client.state.currentUser;
    if (currentUser == null) return;

    final now = DateTime.now();
    final lastReadMessageId = messageId ?? messages.lastOrNull?.id;

    final existingUserRead = currentUserRead;
    updateRead([
      Read(
        user: currentUser,
        lastRead: now,
        lastReadMessageId: lastReadMessageId,
        lastDeliveredAt: existingUserRead?.lastDeliveredAt,
        lastDeliveredMessageId: existingUserRead?.lastDeliveredMessageId,
      ),
    ]);

    // Read supersedes delivered, so drop any pending delivery candidate the
    // new read boundary just made ineligible. `delivery_events` is configured
    // independently of `read_events`, so a channel tracking unread counts
    // locally can still have delivery receipts enabled. Mirrors what the
    // `message.read` event listener does for server-driven channels.
    _client.channelDeliveryReporter.reconcileDelivery([_channel]);

    _isMarkedAsUnread = false;
  }

  /// Marks the channel as unread locally, without making a network request.
  ///
  /// [lastRead] and [lastReadMessageId] define the new read boundary: any
  /// locally-known message that is still eligible per
  /// [MessageRules.canCountAsUnread] once this boundary is applied is counted
  /// as unread.
  ///
  /// Used for channels that track unread counts locally (see
  /// [Channel.usesLocalUnreadCount]), since the server rejects the
  /// mark-unread endpoint for channels that have read events disabled.
  void markUnreadLocally({
    required DateTime lastRead,
    String? lastReadMessageId,
  }) {
    final currentUser = _client.state.currentUser;
    if (currentUser == null) return;

    final existingUserRead = currentUserRead;

    // Apply the new read boundary first so `MessageRules.canCountAsUnread`
    // (which reads `channel.state?.currentUserRead`) evaluates against it.
    updateRead([
      Read(
        user: currentUser,
        lastRead: lastRead,
        lastReadMessageId: lastReadMessageId,
        lastDeliveredAt: existingUserRead?.lastDeliveredAt,
        lastDeliveredMessageId: existingUserRead?.lastDeliveredMessageId,
      ),
    ]);

    // Recompute the unread count from the locally-known messages now that
    // the boundary above is in effect.
    final unread = messages.where((it) => MessageRules.canCountAsUnread(it, _channel)).length;

    unreadCount = unread;
    _isMarkedAsUnread = true;
  }

  /// Counts the number of unread messages mentioning the current user.
  ///
  /// **NOTE**: The method relies on the [Channel.messages] list and doesn't do
  /// any API call. Therefore, the count might be not reliable as it relies on
  /// the local data.
  int countUnreadMentions() {
    final currentUserId = _client.state.currentUser?.id;

    var count = 0;
    for (final message in messages) {
      if (!MessageRules.canCountAsUnread(message, _channel)) continue;
      if (!message.mentionedUsers.any((it) => it.id == currentUserId)) continue;

      count++;
    }

    return count;
  }

  /// Delete all channel messages.
  void truncate() {
    _channelState = _channelState.copyWith(
      messages: [],
    );
  }

  /// Drops the oldest messages, keeping at most [maxMessages].
  ///
  /// No-op when [maxMessages] is non-positive, when the current count is
  /// already within the limit, or when [isUpToDate] is `false`.
  ///
  /// Prefer `StreamChannel.pruneOldest` when a [StreamChannel] is present:
  /// it also resets the widget-layer "top reached" marker so top-pagination
  /// can resume. Calling this directly leaves that marker untouched.
  void pruneOldest(int maxMessages) {
    if (maxMessages <= 0) return;
    if (!isUpToDate) return;

    final current = messages;
    if (current.length <= maxMessages) return;

    final pruned = current.sublist(current.length - maxMessages);
    _channelState = _channelState.copyWith(messages: pruned);
  }

  /// Update channelState with updated information.
  void updateChannelState(ChannelState updatedState) {
    final newMessages = messages.mergeSorted(
      updatedState.messages,
      key: (message) => message.id,
      update: _mergeUpdate,
      compare: _sortByCreatedAt,
    );

    final watchers = _channelState.watchers ?? const <User>[];
    final newWatchers = watchers.merge(
      updatedState.watchers,
      key: (watcher) => watcher.id,
    );

    final reads = _channelState.read ?? const <Read>[];
    final newReads = reads.merge(
      updatedState.read,
      key: (read) => read.user.id,
    );

    _channelState = _channelState.copyWith(
      messages: newMessages,
      channel: _channelState.channel?.merge(updatedState.channel),
      watchers: newWatchers.toList(),
      watcherCount: updatedState.watcherCount,
      members: updatedState.members,
      membership: updatedState.membership,
      read: newReads.toList(),
      draft: updatedState.draft,
      pinnedMessages: updatedState.pinnedMessages,
      pendingMessages: updatedState.pendingMessages,
      pushPreferences: updatedState.pushPreferences,
      activeLiveLocations: updatedState.activeLiveLocations,
    );
  }

  /// Applies a [remoteState] received from the server or offline storage
  /// (e.g. a `query`/`watch` response), merging it into local state.
  ///
  /// Unlike [updateChannelState], this preserves the current user's
  /// locally-tracked read state for channels that track unread counts
  /// on-device (see [Channel.usesLocalUnreadCount]) — their `lastRead`,
  /// `lastReadMessageId`, and `unreadMessages` are kept as-is instead of
  /// being overwritten by the remote payload; only delivery fields are
  /// still applied from it.
  ///
  /// Call this instead of [updateChannelState] whenever [remoteState]
  /// genuinely comes from the network or offline storage.
  void updateChannelStateFromServer(ChannelState remoteState) {
    updateChannelState(_preserveLocalUnreadState(remoteState));
  }

  /// Rewrites the current user's [Read] in [remoteState], if present, to
  /// keep the locally-tracked `lastRead` / `lastReadMessageId` /
  /// `unreadMessages` while still adopting the remote delivery fields.
  ///
  /// No-op unless [Channel.usesLocalUnreadCount] is enabled and a local read
  /// already exists for the current user.
  ChannelState _preserveLocalUnreadState(ChannelState remoteState) {
    if (!_channel.usesLocalUnreadCount) return remoteState;

    final localRead = currentUserRead;
    final remoteReads = remoteState.read;
    if (localRead == null || remoteReads == null) return remoteState;

    final currentUserId = localRead.user.id;
    final preservedReads = remoteReads.map((read) {
      if (read.user.id != currentUserId) return read;
      return localRead.copyWith(
        lastDeliveredAt: read.lastDeliveredAt,
        lastDeliveredMessageId: read.lastDeliveredMessageId,
      );
    });

    return remoteState.copyWith(read: preservedReads.toList());
  }

  int _sortByCreatedAt(Message a, Message b) => a.createdAt.compareTo(b.createdAt);

  /// The channel state related to this client.
  ChannelState get _channelState => _channelStateController.value;

  /// The channel state related to this client as a stream.
  Stream<ChannelState> get channelStateStream => _channelStateController.stream;

  /// The channel state related to this client.
  ChannelState get channelState => _channelStateController.value;
  late BehaviorSubject<ChannelState> _channelStateController;

  late final _debouncedUpdatePersistenceChannelState = debounce(
    (ChannelState state) {
      final persistenceClient = _client.chatPersistenceClient;
      return persistenceClient?.updateChannelState(state);
    },
    const Duration(seconds: 1),
  );

  set _channelState(ChannelState v) {
    _channelStateController.safeAdd(v);
    _debouncedUpdatePersistenceChannelState.call([v]);
  }

  late final _debouncedUpdatePersistenceChannelThreads = debounce(
    (Map<String, List<Message>> threads) async {
      final channelCid = _channel.cid;
      if (channelCid == null) return;

      final persistenceClient = _client.chatPersistenceClient;
      return persistenceClient?.updateChannelThreads(channelCid, threads);
    },
    const Duration(seconds: 1),
  );

  /// The channel threads related to this channel.
  Map<String, List<Message>> get threads => {..._threadsController.value};

  /// The channel threads related to this channel as a stream.
  Stream<Map<String, List<Message>>> get threadsStream => _threadsController;
  final _threadsController = BehaviorSubject.seeded(<String, List<Message>>{});
  set _threads(Map<String, List<Message>> threads) {
    _threadsController.safeAdd(threads);
    _debouncedUpdatePersistenceChannelThreads.call([threads]);
  }

  /// Clears all the replies in the thread identified by [parentId].
  void clearThread(String parentId) {
    final updatedThreads = {
      ...threads,
      parentId: <Message>[],
    };

    _threads = updatedThreads;
  }

  /// Update threads with updated information about messages.
  void updateThreadInfo(String parentId, List<Message> messages) {
    final updatedThreads = {...threads};

    final threadMessages = updatedThreads[parentId] ?? <Message>[];
    final updatedThreadMessages = _mergeMessagesIntoExisting(
      existing: threadMessages,
      toMerge: messages,
    );

    // Update the thread with the modified message list.
    updatedThreads[parentId] = updatedThreadMessages.toList();

    _threads = updatedThreads;
  }

  Draft? _getThreadDraft(String parentId, List<Message>? messages) {
    return messages?.firstWhereOrNull((it) => it.id == parentId)?.draft;
  }

  /// Draft for a specific thread identified by [parentId].
  Draft? threadDraft(String parentId) => _getThreadDraft(parentId, messages);

  /// Stream of draft for a specific thread identified by [parentId].
  ///
  /// This stream emits a new value whenever the draft associated with the
  /// specified thread is updated or removed.
  Stream<Draft?> threadDraftStream(String parentId) =>
      channelStateStream.map((cs) => _getThreadDraft(parentId, cs.messages)).distinct();

  /// Channel related typing users stream.
  Stream<Map<User, Event>> get typingEventsStream => _typingEventsController.stream;

  /// Channel related typing users last value.
  Map<User, Event> get typingEvents => _typingEventsController.value;
  final _typingEventsController = BehaviorSubject.seeded(<User, Event>{});

  void _listenTypingEvents() {
    _subscriptions
      ..add(
        _channel.on(EventType.typingStart).listen(
          (event) {
            final user = event.user;
            if (user == null) return;

            final currentUser = _client.state.currentUser;
            if (event.isFromUser(userId: currentUser?.id)) return;

            final events = {...typingEvents, user: event};
            _typingEventsController.safeAdd(events);
          },
        ),
      )
      ..add(
        _channel.on(EventType.typingStop).listen(
          (event) {
            final user = event.user;
            if (user == null) return;

            final currentUser = _client.state.currentUser;
            if (event.isFromUser(userId: currentUser?.id)) return;

            final events = {...typingEvents}..remove(user);
            _typingEventsController.safeAdd(events);
          },
        ),
      );
  }

  Timer? _staleTypingEventsCleanerTimer;

  // Checks and removes stale typing events that were not explicitly stopped by
  // the sender due to technical difficulties. e.g. process death, loss of
  // Internet connection or custom implementation.
  void _startCleaningStaleTypingEvents() {
    _staleTypingEventsCleanerTimer = Timer.periodic(
      const Duration(seconds: 1),
      (_) {
        final now = DateTime.now();
        typingEvents.forEach((user, event) {
          if (now.difference(event.createdAt).inSeconds > incomingTypingStartEventTimeout) {
            _client.handleEvent(
              Event(
                type: EventType.typingStop,
                user: user,
                cid: _channel.cid,
                parentId: event.parentId,
              ),
            );
          }
        });
      },
    );
  }

  Timer? _stalePinnedMessagesCleanerTimer;

  // Checks and removes stale pinned messages that are not valid anymore.
  void _startCleaningStalePinnedMessages() {
    _stalePinnedMessagesCleanerTimer = Timer.periodic(
      const Duration(seconds: 30),
      (_) {
        final now = DateTime.now();
        var expiredMessages = channelState.pinnedMessages?.where((m) => m.pinExpires?.isBefore(now) == true).toList();
        if (expiredMessages != null && expiredMessages.isNotEmpty) {
          expiredMessages = expiredMessages
              .map(
                (m) => m.copyWith(
                  pinExpires: null,
                  pinned: false,
                ),
              )
              .toList();

          updateChannelState(
            _channelState.copyWith(
              pinnedMessages: pinnedMessages.where(_pinIsValid).toList(),
              messages: expiredMessages,
            ),
          );
        }
      },
    );
  }

  Timer? _staleLiveLocationsCleanerTimer;
  void _startCleaningExpiredLocations() {
    _staleLiveLocationsCleanerTimer?.cancel();
    _staleLiveLocationsCleanerTimer = Timer.periodic(
      const Duration(seconds: 1),
      (_) {
        final currentUserId = _channel._client.state.currentUser?.id;
        if (currentUserId == null) return;

        final expired = activeLiveLocations.where((it) => it.isExpired);
        if (expired.isEmpty) return;

        for (final sharedLocation in expired) {
          // Skip if the location is shared by the current user,
          // as we are already handling them in the client.
          if (sharedLocation.userId == currentUserId) continue;

          final lastUpdatedAt = DateTime.timestamp();
          final locationExpiredEvent = Event(
            type: EventType.locationExpired,
            cid: sharedLocation.channelCid,
            message: Message(
              id: sharedLocation.messageId,
              updatedAt: lastUpdatedAt,
              sharedLocation: sharedLocation.copyWith(
                updatedAt: lastUpdatedAt,
              ),
            ),
          );

          _channel._client.handleEvent(locationExpiredEvent);
        }
      },
    );
  }

  // Listens to channel push preference update events and updates the state
  void _listenChannelPushPreferenceUpdated() {
    _subscriptions.add(
      _channel.on(EventType.channelPushPreferenceUpdated).listen(
        (event) {
          final pushPreferences = event.channelPushPreference;
          if (pushPreferences == null) return;

          updateChannelState(
            channelState.copyWith(
              pushPreferences: pushPreferences,
            ),
          );
        },
      ),
    );
  }

  Future<void> _deleteMessagesFromUser({
    required String userId,
    bool hardDelete = false,
    DateTime? deletedAt,
  }) async {
    // Delete messages from persistence.
    //
    // Note: We perform this operation separately even though [_removeMessages]
    // already handles it as we need to delete all messages from the user, not
    // only the ones present in the current state.
    final persistence = _channel.client.chatPersistenceClient;
    await persistence?.deleteMessagesFromUser(
      userId: userId,
      cid: _channel.cid,
      hardDelete: hardDelete,
      deletedAt: deletedAt,
    );

    // Gather messages to delete from state.
    final userMessages = <String, Message>{};
    for (final message in [...messages, ...threads.values.flattened]) {
      if (message.user?.id != userId) continue;
      userMessages[message.id] = message.copyWith(
        type: MessageType.deleted,
        deletedAt: deletedAt ?? DateTime.now(),
        state: switch (hardDelete) {
          true => MessageState.hardDeleted,
          false => MessageState.softDeleted,
        },
      );
    }

    final messagesToDelete = userMessages.values;
    return _deleteMessages(messagesToDelete, hardDelete: hardDelete);
  }

  void _deleteMessages(
    Iterable<Message> messages, {
    bool hardDelete = false,
  }) {
    if (messages.isEmpty) return;

    if (hardDelete) return _removeMessages(messages);
    return _updateMessages(messages, upsert: false);
  }

  void _updateMessages(
    Iterable<Message> messages, {
    Message Function(Message original, Message updated) update = _mergeUpdate,
    bool upsert = true,
  }) {
    if (messages.isEmpty) return;

    _updateThreadMessages(messages, update: update, upsert: upsert);
    _updateChannelMessages(messages, update: update, upsert: upsert);
    _updatePinnedMessages(messages, update: update);
    _updateActiveLiveLocations(messages);
  }

  void _updateThreadMessages(
    Iterable<Message> messages, {
    Message Function(Message original, Message updated) update = _mergeUpdate,
    bool upsert = true,
  }) {
    if (messages.isEmpty) return;

    // Group messages by parentId so each thread merge only sees its own
    // replies — passing the full batch to every thread would leak replies
    // across thread boundaries (the merge dedups by id, not by parentId).
    final messagesByThread = <String, List<Message>>{};
    for (final m in messages) {
      if (m.parentId case final parentId?) (messagesByThread[parentId] ??= []).add(m);
    }

    // If there are no affected threads, return early.
    if (messagesByThread.isEmpty) return;

    final updatedThreads = {...threads};
    for (final MapEntry(key: thread, :value) in messagesByThread.entries) {
      final existingThreadMessages = updatedThreads[thread];

      // Don't create a phantom entry for a thread that wasn't loaded: with
      // `upsert: false` an out-of-window reply is dropped, so there's nothing
      // to merge. Writing it back would make `threads.containsKey(parentId)`
      // report a thread that was never paged in.
      if (existingThreadMessages == null && !upsert) continue;

      final threadMessages = existingThreadMessages ?? <Message>[];
      final updatedThreadMessages = _mergeMessagesIntoExisting(
        existing: threadMessages,
        toMerge: value,
        update: update,
        upsert: upsert,
      );

      // Update the thread with the modified message list.
      updatedThreads[thread] = updatedThreadMessages.toList();
    }

    // Update the threads map.
    _threads = updatedThreads;
  }

  void _updateChannelMessages(
    Iterable<Message> messages, {
    Message Function(Message original, Message updated) update = _mergeUpdate,
    bool upsert = true,
  }) {
    if (messages.isEmpty) return;

    final affectedMessages = messages.map((it) {
      // If it's not a thread message, consider it affected.
      if (it.parentId == null) return it;
      // If it's a thread message shown in channel, consider it affected.
      if (it.showInChannel == true) return it;

      return null; // Thread message not shown in channel, ignore it.
    }).nonNulls;

    // If there are no affected messages, return early.
    if (affectedMessages.isEmpty) return;

    final channelMessages = [...this.messages];
    final updatedChannelMessages = _mergeMessagesIntoExisting(
      existing: channelMessages,
      toMerge: affectedMessages,
      update: update,
      upsert: upsert,
    );

    // Calculate the new last message at time.
    var lastMessageAt = _channelState.channel?.lastMessageAt;
    for (final message in affectedMessages) {
      if (MessageRules.canUpdateChannelLastMessageAt(message, _channel)) {
        lastMessageAt = [lastMessageAt, message.createdAt].nonNulls.max;
      }
    }

    _channelState = _channelState.copyWith(
      messages: updatedChannelMessages.toList(),
      channel: _channelState.channel?.copyWith(lastMessageAt: lastMessageAt),
    );
  }

  void _updatePinnedMessages(
    Iterable<Message> messages, {
    Message Function(Message original, Message updated) update = _mergeUpdate,
  }) {
    if (messages.isEmpty) return;

    // No-op fast path: nothing was pinned, and nothing in the batch is
    // becoming pinned — skip the merge/copyWith churn that would otherwise
    // land right back on an empty `pinnedMessages` list.
    if (pinnedMessages.isEmpty && messages.every((m) => !m.pinned)) return;

    final updatedPinnedMessages = _mergePinnedMessagesIntoExisting(
      existing: pinnedMessages,
      toMerge: messages,
      update: update,
    );

    _channelState = _channelState.copyWith(
      pinnedMessages: updatedPinnedMessages.toList(),
    );
  }

  void _updateActiveLiveLocations(Iterable<Message> messages) {
    if (messages.isEmpty) return;

    final activeLiveLocations = [...this.activeLiveLocations];
    final updatedActiveLiveLocations = _mergeActiveLocationsIntoExisting(
      existing: activeLiveLocations,
      toMerge: messages,
    );

    _channelState = _channelState.copyWith(
      activeLiveLocations: updatedActiveLiveLocations.toList(),
    );
  }

  Iterable<Location> _mergeActiveLocationsIntoExisting({
    required Iterable<Location> existing,
    required Iterable<Message> toMerge,
  }) {
    if (toMerge.isEmpty) return existing;

    final mergedLocations = existing.mergeFrom(
      toMerge,
      key: (it) => (it.userId, it.channelCid, it.createdByDeviceId),
      value: (message) => message.sharedLocation,
      update: (original, updated) => updated,
    );

    final toUpdateMap = {for (final m in toMerge) m.id: m};
    final updatedLocations = mergedLocations.where((it) {
      // Remove the location if it's expired.
      if (it.isExpired) return false;

      final updatedMessage = toUpdateMap[it.messageId];
      // Remove the location if the attached message is deleted.
      if (updatedMessage?.isDeleted == true) return false;

      return true;
    });

    return updatedLocations;
  }

  Iterable<Message> _mergePinnedMessagesIntoExisting({
    required Iterable<Message> existing,
    required Iterable<Message> toMerge,
    Message Function(Message original, Message updated) update = _mergeUpdate,
  }) {
    return _mergeMessagesIntoExisting(
      existing: existing,
      toMerge: toMerge,
      update: update,
    ).where(_pinIsValid);
  }

  Iterable<Message> _mergeMessagesIntoExisting({
    required Iterable<Message> existing,
    required Iterable<Message> toMerge,
    Message Function(Message original, Message updated) update = _mergeUpdate,
    bool upsert = true,
  }) {
    if (toMerge.isEmpty) return existing;

    // [update] decides whether each pair is reconciled (default — see
    // `_mergeUpdate`) or replaced (`_replaceUpdate`, used by local rollback
    // paths that don't want enrichment fallback to keep optimistic values).
    //
    // [upsert] controls whether ids not already in [existing] are inserted.
    // Event-driven paths (`message.updated`, `message.deleted` soft) pass
    // `upsert: false` so an out-of-window message isn't dropped into a gap
    // between the loaded slice and history the client hasn't paged in yet.
    final existingList = existing is List<Message> ? existing : existing.toList();
    var toMergeList = toMerge is List<Message> ? toMerge : toMerge.toList();

    // Single-message fast path. The hot ingest path (server echoes, edits,
    // reactions, read receipts) always lands here, and `lastIndexWhere` +
    // `sortedUpsertAt` skips the O(N) keymap build that the two-pointer
    // merge would otherwise do up front.
    if (toMergeList.length == 1) {
      final message = toMergeList.first;
      final oldIndex = existingList.lastIndexWhere((it) => it.id == message.id);

      // upsert: false — skip update if message is not loaded
      if (oldIndex == -1 && !upsert) return existingList;

      final resolved = oldIndex == -1 ? message : update(existingList[oldIndex], message);

      final mergedMessages = existingList.sortedUpsertAt(
        oldIndex,
        resolved,
        update: update,
        compare: _sortByCreatedAt,
      );

      // Non-delete updates can't change what embedded quotedMessage copies
      // should display, so we can skip the rewrite entirely.
      if (!resolved.isDeleted) return mergedMessages;

      return mergedMessages.updateIf(
        (it) => it.quotedMessageId == resolved.id,
        (it) => it.copyWith(quotedMessage: resolved),
      );
    }

    // upsert: false - skip messages not loaded in the window
    if (!upsert) {
      final existingIds = {for (final m in existingList) m.id};
      toMergeList = toMergeList.where((m) => existingIds.contains(m.id)).toList();
      if (toMergeList.isEmpty) return existingList;
    }

    // Batch path: receiver (`existingList`) is maintained sorted as a
    // state invariant; `mergeSorted` sorts `toMergeList` internally and
    // returns a sorted result.
    final mergedMessages = existingList.mergeSorted(
      toMergeList,
      key: (message) => message.id,
      update: update,
      compare: _sortByCreatedAt,
    );

    // Refresh embedded `quotedMessage` refs only for messages quoting an
    // incoming message that is now deleted. `updateIf` returns the same
    // list reference when nothing matches, so steady-state allocates
    // nothing for this step.
    final deletedIds = toMergeList.where((m) => m.isDeleted).map((m) => m.id).toSet();
    if (deletedIds.isEmpty) return mergedMessages;

    final mergedById = {for (final m in mergedMessages) m.id: m};
    return mergedMessages.updateIf(
      (it) => deletedIds.contains(it.quotedMessageId),
      (it) => it.copyWith(quotedMessage: mergedById[it.quotedMessageId]),
    );
  }

  void _removeMessages(Iterable<Message> messages) {
    if (messages.isEmpty) return;

    final messageIds = messages.map((m) => m.id).toSet().toList();
    final persistenceClient = _channel.client.chatPersistenceClient;
    // Remove the messages from the persistence client.
    persistenceClient?.deleteMessageByIds(messageIds);
    persistenceClient?.deletePinnedMessageByIds(messageIds);

    _removeThreadMessages(messages);
    _removeChannelMessages(messages);
    _removePinnedMessages(messages);
    _removeActiveLiveLocations(messages);
  }

  void _removeThreadMessages(Iterable<Message> messages) {
    if (messages.isEmpty) return;

    final affectedThreads = {...messages.map((it) => it.parentId).nonNulls};
    // If there are no affected threads, return early.
    if (affectedThreads.isEmpty) return;

    final updatedThreads = {...threads};
    for (final thread in affectedThreads) {
      final threadMessages = updatedThreads[thread];
      // Continue if the thread doesn't exist.
      if (threadMessages == null) continue;

      // Remove the deleted message from the thread messages and reference from
      // other messages quoting it.
      final updatedThreadMessages = _removeMessagesFromExisting(
        existing: threadMessages,
        toRemove: messages,
      );

      // If there are no more messages in the thread, remove the thread entry.
      if (updatedThreadMessages.isEmpty) {
        updatedThreads.remove(thread);
        continue;
      }

      // Otherwise, update the thread with the modified message list.
      updatedThreads[thread] = updatedThreadMessages.toList();
    }

    // Update the threads map.
    _threads = updatedThreads;
  }

  void _removeChannelMessages(Iterable<Message> messages) {
    if (messages.isEmpty) return;

    final affectedMessages = messages.map((it) {
      // If it's not a thread message, consider it affected.
      if (it.parentId == null) return it;
      // If it's a thread message shown in channel, consider it affected.
      if (it.showInChannel == true) return it;

      return null; // Thread message not shown in channel, ignore it.
    }).nonNulls;

    // If there are no affected messages, return early.
    if (affectedMessages.isEmpty) return;

    final channelMessages = [...this.messages];
    final updatedChannelMessages = _removeMessagesFromExisting(
      existing: channelMessages,
      toRemove: affectedMessages,
    );

    _channelState = _channelState.copyWith(
      messages: updatedChannelMessages.toList(),
    );
  }

  void _removePinnedMessages(Iterable<Message> messages) {
    if (messages.isEmpty) return;

    final pinnedMessages = [...this.pinnedMessages];
    final updatedPinnedMessages = _removePinnedMessagesFromExisting(
      existing: pinnedMessages,
      toRemove: messages,
    );

    _channelState = _channelState.copyWith(
      pinnedMessages: updatedPinnedMessages.toList(),
    );
  }

  void _removeActiveLiveLocations(Iterable<Message> messages) {
    if (messages.isEmpty) return;

    final activeLiveLocations = [...this.activeLiveLocations];
    final updatedActiveLiveLocations = _removeActiveLocationsFromExisting(
      existing: activeLiveLocations,
      toRemove: messages,
    );

    _channelState = _channelState.copyWith(
      activeLiveLocations: updatedActiveLiveLocations.toList(),
    );
  }

  Iterable<Location> _removeActiveLocationsFromExisting({
    required Iterable<Location> existing,
    required Iterable<Message> toRemove,
  }) {
    if (toRemove.isEmpty) return existing;

    final toRemoveIds = toRemove.map((m) => m.id).toSet();
    final updatedLocations = existing.where(
      // Remove the location if its attached message is in the toRemove list.
      (it) => !toRemoveIds.contains(it.messageId),
    );

    return updatedLocations;
  }

  Iterable<Message> _removePinnedMessagesFromExisting({
    required Iterable<Message> existing,
    required Iterable<Message> toRemove,
  }) {
    return _removeMessagesFromExisting(
      existing: existing,
      toRemove: toRemove,
    ).where(_pinIsValid);
  }

  Iterable<Message> _removeMessagesFromExisting({
    required Iterable<Message> existing,
    required Iterable<Message> toRemove,
  }) {
    if (toRemove.isEmpty) return existing;

    final toRemoveIds = toRemove.map((m) => m.id).toSet();
    final updatedMessages = existing
        .where((it) {
          // Remove the message if it's in the toRemove list.
          return !toRemoveIds.contains(it.id);
        })
        .map((it) {
          // Continue if the message doesn't quote any of the deleted messages.
          if (!toRemoveIds.contains(it.quotedMessageId)) return it;

          // Setting it to null will remove the quoted message from the message.
          return it.copyWith(quotedMessageId: null, quotedMessage: null);
        });

    return updatedMessages;
  }

  // Listens to user message deleted events and marks messages from that user
  // as either soft or hard deleted based on the event data.
  void _listenUserMessagesDeleted() {
    _subscriptions.add(
      _channel.on(EventType.userMessagesDeleted).listen((event) async {
        final user = event.user;
        if (user == null) return;

        return _deleteMessagesFromUser(
          userId: user.id,
          hardDelete: event.hardDelete ?? false,
          deletedAt: event.createdAt,
        );
      }),
    );
  }

  /// Call this method to dispose this object.
  void dispose() {
    _debouncedUpdatePersistenceChannelThreads.cancel();
    _debouncedUpdatePersistenceChannelState.cancel();
    _retryQueue.dispose();
    _subscriptions.cancel();
    _channelStateController.close();
    _isUpToDateController.close();
    _threadsController.close();
    _staleTypingEventsCleanerTimer?.cancel();
    _stalePinnedMessagesCleanerTimer?.cancel();
    _staleLiveLocationsCleanerTimer?.cancel();
    _typingEventsController.close();
  }
}

bool _pinIsValid(Message message) {
  // If the message is deleted, the pin is not valid.
  if (message.isDeleted) return false;

  // If the message is not pinned, it's not valid.
  if (message.pinned != true) return false;

  // If there's no expiration, the pin is valid.
  final pinExpires = message.pinExpires;
  if (pinExpires == null) return true;

  // If there's an expiration, check if it's still valid.
  return pinExpires.isAfter(DateTime.now());
}

/// Extension methods for reading related operations on a ChannelClientState.
extension ChannelReadHelper on ChannelClientState {
  /// Get the [Read] object for a specific user identified by [userId].
  Read? userReadOf({String? userId}) => read.userReadOf(userId: userId);

  /// Stream of [Read] object for a specific user identified by [userId].
  Stream<Read?> userReadStreamOf({String? userId}) {
    return readStream.map((read) => read.userReadOf(userId: userId));
  }

  /// Returns the list of [Read]s that have marked the given [msg] as read.
  ///
  /// The [Read] is considered to have read the message if:
  /// - The read user is not the sender of the message.
  /// - The read's lastRead is after or equal to the message's createdAt.
  List<Read> readsOf({required Message message}) {
    return read.readsOf(message: message);
  }

  /// Stream of list of [Read]s that have marked the given [msg] as read.
  ///
  /// The [Read] is considered to have read the message if:
  /// - The read user is not the sender of the message.
  /// - The read's lastRead is after or equal to the message's createdAt.
  Stream<List<Read>> readsOfStream({required Message message}) {
    return readStream.map((read) => read.readsOf(message: message));
  }

  /// Returns the list of [Read]s that have marked the given [message] as
  /// delivered.
  ///
  /// The [Read] is considered to have delivered the message if:
  /// - The read user is not the sender of the message.
  /// - The read contains a non-null lastDeliveredAt.
  /// - The read's lastDeliveredAt is after or equal to the message's createdAt.
  List<Read> deliveriesOf({required Message message}) {
    return read.deliveriesOf(message: message);
  }

  /// Stream of list of [Read]s that have marked the given [message] as
  /// delivered.
  ///
  /// The [Read] is considered to have delivered the message if:
  /// - The read user is not the sender of the message.
  /// - The read contains a non-null lastDeliveredAt.
  /// - The read's lastDeliveredAt is after or equal to the message's createdAt.
  Stream<List<Read>> deliveriesOfStream({required Message message}) {
    return readStream.map((read) => read.deliveriesOf(message: message));
  }
}

/// Extension methods for checking channel capabilities on a Channel instance.
///
/// These methods provide a convenient way to check if the current user has
/// specific capabilities in a channel.
extension ChannelCapabilityCheck on Channel {
  /// True, if the current user can send a message to this channel.
  bool get canSendMessage {
    return ownCapabilities.contains(ChannelCapability.sendMessage);
  }

  /// True, if the current user can send a reply to this channel.
  bool get canSendReply {
    return ownCapabilities.contains(ChannelCapability.sendReply);
  }

  /// True, if the current user can send a message with restricted visibility.
  bool get canSendRestrictedVisibilityMessage {
    return ownCapabilities.contains(
      ChannelCapability.sendRestrictedVisibilityMessage,
    );
  }

  /// True, if the current user can send reactions.
  bool get canSendReaction {
    return ownCapabilities.contains(ChannelCapability.sendReaction);
  }

  /// True, if the current user can attach links to messages.
  bool get canSendLinks {
    return ownCapabilities.contains(ChannelCapability.sendLinks);
  }

  /// True, if the current user can attach files to messages.
  bool get canCreateAttachment {
    return ownCapabilities.contains(ChannelCapability.createAttachment);
  }

  /// True, if the current user can freeze or unfreeze channel.
  bool get canFreezeChannel {
    return ownCapabilities.contains(ChannelCapability.freezeChannel);
  }

  /// True, if the current user can enable or disable slow mode.
  bool get canSetChannelCooldown {
    return ownCapabilities.contains(ChannelCapability.setChannelCooldown);
  }

  /// True, if the current user can leave channel (remove own membership).
  bool get canLeaveChannel {
    return ownCapabilities.contains(ChannelCapability.leaveChannel);
  }

  /// True, if the current user can join channel (add own membership).
  bool get canJoinChannel {
    return ownCapabilities.contains(ChannelCapability.joinChannel);
  }

  /// True, if the current user can pin a message.
  bool get canPinMessage {
    return ownCapabilities.contains(ChannelCapability.pinMessage);
  }

  /// True, if the current user can delete any message from the channel.
  bool get canDeleteAnyMessage {
    return ownCapabilities.contains(ChannelCapability.deleteAnyMessage);
  }

  /// True, if the current user can delete own messages from the channel.
  bool get canDeleteOwnMessage {
    return ownCapabilities.contains(ChannelCapability.deleteOwnMessage);
  }

  /// True, if the current user can update any message in the channel.
  bool get canUpdateAnyMessage {
    return ownCapabilities.contains(ChannelCapability.updateAnyMessage);
  }

  /// True, if the current user can update own messages in the channel.
  bool get canUpdateOwnMessage {
    return ownCapabilities.contains(ChannelCapability.updateOwnMessage);
  }

  /// True, if the current user can use message search.
  bool get canSearchMessages {
    return ownCapabilities.contains(ChannelCapability.searchMessages);
  }

  /// True, if the current user can send typing events.
  @Deprecated('Use canUseTypingEvents instead')
  bool get canSendTypingEvents {
    if (canUseTypingEvents) return true;
    return ownCapabilities.contains(ChannelCapability.sendTypingEvents);
  }

  /// True, if the current user can upload message attachments.
  bool get canUploadFile {
    return ownCapabilities.contains(ChannelCapability.uploadFile);
  }

  /// True, if the current user can delete channel.
  bool get canDeleteChannel {
    return ownCapabilities.contains(ChannelCapability.deleteChannel);
  }

  /// True, if the current user can update channel data.
  bool get canUpdateChannel {
    return ownCapabilities.contains(ChannelCapability.updateChannel);
  }

  /// True, if the current user can update channel members.
  bool get canUpdateChannelMembers {
    return ownCapabilities.contains(ChannelCapability.updateChannelMembers);
  }

  /// True, if the current user can update thread data.
  bool get canUpdateThread {
    return ownCapabilities.contains(ChannelCapability.updateThread);
  }

  /// True, if the current user can quote a message.
  bool get canQuoteMessage {
    return ownCapabilities.contains(ChannelCapability.quoteMessage);
  }

  /// True, if the current user can ban channel members.
  bool get canBanChannelMembers {
    return ownCapabilities.contains(ChannelCapability.banChannelMembers);
  }

  /// True, if the current user can flag a message.
  bool get canFlagMessage {
    return ownCapabilities.contains(ChannelCapability.flagMessage);
  }

  /// True, if the current user can mute a channel.
  bool get canMuteChannel {
    return ownCapabilities.contains(ChannelCapability.muteChannel);
  }

  /// True, if the current user can send custom events.
  bool get canSendCustomEvents {
    return ownCapabilities.contains(ChannelCapability.sendCustomEvents);
  }

  /// True, if the current user has read events capability.
  @Deprecated('Use canUseReadReceipts instead')
  bool get canReceiveReadEvents => canUseReadReceipts;

  /// True, if the current user has read events capability.
  bool get canUseReadReceipts {
    return ownCapabilities.contains(ChannelCapability.readEvents);
  }

  /// True, if unread counts for this channel should be tracked locally,
  /// on-device, rather than relying on the server.
  ///
  /// This is the case when [StreamChatClient.isLocalUnreadCountEnabled] is
  /// enabled and the channel doesn't support read receipts (for example,
  /// livestream channel types that disable read events). Channels that
  /// support read receipts always rely on server-driven unread counts.
  bool get usesLocalUnreadCount {
    return _client.isLocalUnreadCountEnabled && !canUseReadReceipts;
  }

  /// True, if the current user has connect events capability.
  bool get canReceiveConnectEvents {
    return ownCapabilities.contains(ChannelCapability.connectEvents);
  }

  /// True, if the current user can send and receive typing events.
  bool get canUseTypingEvents {
    return ownCapabilities.contains(ChannelCapability.typingEvents);
  }

  /// True, if channel slow mode is active.
  bool get isInSlowMode {
    return ownCapabilities.contains(ChannelCapability.slowMode);
  }

  /// True, if the current user is allowed to post messages as usual even if the
  /// channel is in slow mode.
  bool get canSkipSlowMode {
    return ownCapabilities.contains(ChannelCapability.skipSlowMode);
  }

  /// True, if the current user can create a poll.
  bool get canSendPoll {
    return ownCapabilities.contains(ChannelCapability.sendPoll);
  }

  /// True, if the current user can vote in a poll.
  bool get canCastPollVote {
    return ownCapabilities.contains(ChannelCapability.castPollVote);
  }

  /// True, if the current user can query poll votes.
  bool get canQueryPollVotes {
    return ownCapabilities.contains(ChannelCapability.queryPollVotes);
  }

  /// True, if the current user has delivery events capability.
  bool get canUseDeliveryReceipts {
    return ownCapabilities.contains(ChannelCapability.deliveryEvents);
  }

  /// True, if the current user can share location in the channel.
  bool get canShareLocation {
    return ownCapabilities.contains(ChannelCapability.shareLocation);
  }

  /// True, if the current user can send an "@channel" mention that notifies
  /// all channel members.
  bool get canNotifyChannel {
    return ownCapabilities.contains(ChannelCapability.notifyChannel);
  }

  /// True, if the current user can send an "@here" mention that notifies all
  /// online channel members.
  bool get canNotifyHere {
    return ownCapabilities.contains(ChannelCapability.notifyHere);
  }

  /// True, if the current user can mention one or more roles in a message.
  bool get canNotifyRole {
    return ownCapabilities.contains(ChannelCapability.notifyRole);
  }

  /// True, if the current user can mention one or more user groups in a
  /// message.
  bool get canNotifyGroup {
    return ownCapabilities.contains(ChannelCapability.notifyGroup);
  }
}