Merge remote-tracking branch 'origin/master' into v10.0.0

# Conflicts:
#	melos.yaml
#	packages/stream_chat/CHANGELOG.md
#	packages/stream_chat/example/pubspec.yaml
#	packages/stream_chat/lib/src/client/channel.dart
#	packages/stream_chat/lib/src/core/models/channel_config.dart
#	packages/stream_chat/lib/src/core/models/channel_config.g.dart
#	packages/stream_chat/lib/src/core/models/message.dart
#	packages/stream_chat/lib/src/core/models/message.g.dart
#	packages/stream_chat/lib/version.dart
#	packages/stream_chat/pubspec.yaml
#	packages/stream_chat_flutter/CHANGELOG.md
#	packages/stream_chat_flutter/example/pubspec.yaml
#	packages/stream_chat_flutter/pubspec.yaml
#	packages/stream_chat_flutter_core/CHANGELOG.md
#	packages/stream_chat_flutter_core/example/pubspec.yaml
#	packages/stream_chat_flutter_core/pubspec.yaml
#	packages/stream_chat_localizations/CHANGELOG.md
#	packages/stream_chat_localizations/example/pubspec.yaml
#	packages/stream_chat_localizations/pubspec.yaml
#	packages/stream_chat_persistence/CHANGELOG.md
#	packages/stream_chat_persistence/example/pubspec.yaml
#	packages/stream_chat_persistence/lib/src/db/drift_chat_database.dart
#	packages/stream_chat_persistence/pubspec.yaml
#	sample_app/pubspec.yaml

Author: xsahil03x

.github/workflows/check_db_entities.yml

@@ -48,7 +48,7 @@ jobs:
 
       - name: 🔍 Find Comment
         if: steps.check_entities.outputs.has_changes == 'true'
-        uses: peter-evans/find-comment@v3
+        uses: peter-evans/find-comment@v4
         id: find_comment
         with:
           issue-number: ${{ github.event.pull_request.number }}
@@ -57,7 +57,7 @@ jobs:
 
       - name: 💬 Create or Update Comment
         if: steps.check_entities.outputs.has_changes == 'true'
-        uses: peter-evans/create-or-update-comment@v4
+        uses: peter-evans/create-or-update-comment@v5
         with:
           comment-id: ${{ steps.find_comment.outputs.comment-id }}
           issue-number: ${{ github.event.pull_request.number }}

.github/workflows/stream_flutter_workflow.yml

@@ -8,6 +8,7 @@ on:
   pull_request:
     paths:
       - 'packages/**'
+      - 'sample_app/**'
       - '.github/workflows/stream_flutter_workflow.yml'
     types:
       - opened

.github/workflows/update_goldens.yml

@@ -30,7 +30,7 @@ jobs:
         run: melos run update:goldens
 
       - name: 📤 Commit Changes
-        uses: stefanzweifel/git-auto-commit-action@v6
+        uses: stefanzweifel/git-auto-commit-action@v7
         with:
           commit_message: "chore: Update Goldens"
           file_pattern: "**/test/**/goldens/*.png"

packages/stream_chat/CHANGELOG.md

@@ -25,6 +25,26 @@ For more details, please refer to the [migration guide](../../migrations/v10-mig
   - `StreamChatClient.removeImage()` - Remove an image from the Stream CDN
   - `StreamChatClient.removeFile()` - Remove a file from the Stream CDN
 
+## 9.18.0
+
+🐞 Fixed
+
+- Improved sync reliability and error handling with enhanced `lastSyncAt` initialization, 400
+  error recovery, and automatic flushing of stale persistence data after 30 days of inactivity.
+
+✅ Added
+
+- Added support for `Message.channelRole` field to provide access to the sender's channel role.
+- Added support for `Channel.messageCount` field.
+- Added support for Pending Messages. Pending messages can be accessed via
+  `ChannelState.pendingMessages` or `ChannelState.pendingMessagesStream`.
+
+🐞 Fixed
+
+- Fixed thread messages increasing the unread count in the main channel.
+- Fixed `ChannelState.memberCount`, `ChannelState.config` and `ChannelState.extraData` getting reset
+  on first load.
+
 ## 10.0.0-beta.6
 
 - Included the changes from version [`9.17.0`](https://pub.dev/packages/stream_chat/changelog).

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

@@ -407,6 +407,24 @@ class Channel {
     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);
+  }
+
   /// Channel id.
   String? get id => state?._channelState.channel?.id ?? _id;
 
@@ -2256,23 +2274,19 @@ class ChannelClientState {
   ChannelClientState(
     this._channel,
     ChannelState channelState,
-  ) : _debouncedUpdatePersistenceChannelState = debounce(
-          (ChannelState state) {
-            final persistenceClient = _channel._client.chatPersistenceClient;
-            return persistenceClient?.updateChannelState(state);
-          },
-          const Duration(seconds: 1),
-        ) {
+  ) {
     _retryQueue = RetryQueue(
       channel: _channel,
       logger: _channel.client.detachedLogger(
         '⟳ (${generateHash([_channel.cid])})',
       ),
     );
 
-    _checkExpiredAttachmentMessages(channelState);
-
     _channelStateController = BehaviorSubject.seeded(channelState);
+    // Update the persistence storage with the seeded channel state.
+    _debouncedUpdatePersistenceChannelState.call([channelState]);
+
+    _checkExpiredAttachmentMessages(channelState);
 
     // region TYPING EVENTS
     _listenTypingEvents();
@@ -2313,6 +2327,7 @@ class ChannelClientState {
     // region CHANNEL EVENTS
     _listenChannelTruncated();
     _listenChannelUpdated();
+    _listenChannelMessageCount();
     // endregion
 
     // region MEMBER EVENTS
@@ -2348,20 +2363,11 @@ class ChannelClientState {
 
     _listenChannelPushPreferenceUpdated();
 
-    _channel._client.chatPersistenceClient
-        ?.getChannelThreads(_channel.cid!)
-        .then((threads) {
-      _threads = threads;
-    }).then((_) {
-      _channel._client.chatPersistenceClient
-          ?.getChannelStateByCid(_channel.cid!)
-          .then((state) {
-        // Replacing the persistence state members with the latest
-        // `channelState.members` as they may have changes over the time.
-        updateChannelState(state.copyWith(members: channelState.members));
-        retryFailedMessages();
-      });
-    });
+    final persistenceClient = _channel.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;
@@ -2496,6 +2502,23 @@ class ChannelClientState {
     }));
   }
 
+  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)
@@ -3478,6 +3501,15 @@ class ChannelClientState {
       .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 =>
       _channelState.messages != null && _channelState.messages!.isNotEmpty
@@ -3602,7 +3634,7 @@ class ChannelClientState {
     if (message.isEphemeral) return false;
 
     // Don't count thread replies which are not shown in the channel as unread.
-    if (message.parentId != null && message.showInChannel == false) {
+    if (message.parentId != null && message.showInChannel != true) {
       return false;
     }
 
@@ -3624,6 +3656,18 @@ class ChannelClientState {
     final isMuted = currentUser.mutes.any((it) => it.user.id == messageUser.id);
     if (isMuted) return false;
 
+    final lastRead = currentUserRead?.lastRead;
+    // Don't count messages created before the last read time as unread.
+    if (lastRead case final read? when message.createdAt.isBefore(read)) {
+      return false;
+    }
+
+    final lastReadMessageId = currentUserRead?.lastReadMessageId;
+    // Don't count if the last read message id is the same as the message id.
+    if (lastReadMessageId case final id? when message.id == id) {
+      return false;
+    }
+
     // If we've passed all checks, count the message as unread.
     return true;
   }
@@ -3700,6 +3744,7 @@ class ChannelClientState {
       read: newReads,
       draft: updatedState.draft,
       pinnedMessages: updatedState.pinnedMessages,
+      pendingMessages: updatedState.pendingMessages,
       pushPreferences: updatedState.pushPreferences,
       activeLiveLocations: updatedState.activeLiveLocations,
     );
@@ -3718,13 +3763,30 @@ class ChannelClientState {
   ChannelState get channelState => _channelStateController.value;
   late BehaviorSubject<ChannelState> _channelStateController;
 
-  final Debounce _debouncedUpdatePersistenceChannelState;
+  late final _debouncedUpdatePersistenceChannelState = debounce(
+    (ChannelState state) {
+      final persistenceClient = _channel._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 = _channel._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};
 
@@ -3733,10 +3795,7 @@ class ChannelClientState {
   final _threadsController = BehaviorSubject.seeded(<String, List<Message>>{});
   set _threads(Map<String, List<Message>> threads) {
     _threadsController.safeAdd(threads);
-    _channel.client.chatPersistenceClient?.updateChannelThreads(
-      _channel.cid!,
-      threads,
-    );
+    _debouncedUpdatePersistenceChannelThreads.call([threads]);
   }
 
   /// Clears all the replies in the thread identified by [parentId].
@@ -3934,6 +3993,7 @@ class ChannelClientState {
 
   /// Call this method to dispose this object.
   void dispose() {
+    _debouncedUpdatePersistenceChannelThreads.cancel();
     _debouncedUpdatePersistenceChannelState.cancel();
     _retryQueue.dispose();
     _subscriptions.cancel();

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

@@ -1,5 +1,6 @@
 import 'dart:async';
 
+import 'package:collection/collection.dart';
 import 'package:dio/dio.dart';
 import 'package:logging/logging.dart';
 import 'package:meta/meta.dart';
@@ -192,9 +193,6 @@ class StreamChatClient {
 
   late final RetryPolicy _retryPolicy;
 
-  /// the last dateTime at the which all the channels were synced
-  DateTime? _lastSyncedAt;
-
   /// The retry policy options getter
   RetryPolicy get retryPolicy => _retryPolicy;
 
@@ -523,25 +521,17 @@ class StreamChatClient {
 
     if (connectionRecovered) {
       // connection recovered
-      final cids = state.channels.keys.toList(growable: false);
+      final cids = [...state.channels.keys.toSet()];
       if (cids.isNotEmpty) {
         await queryChannelsOnline(
           filter: Filter.in_('cid', cids),
           paginationParams: const PaginationParams(limit: 30),
         );
-        if (persistenceEnabled) {
-          await sync(cids: cids, lastSyncAt: _lastSyncedAt);
-        }
-      } else {
-        // channels are empty, assuming it's a fresh start
-        // and making sure `lastSyncAt` is initialized
-        if (persistenceEnabled) {
-          final lastSyncAt = await chatPersistenceClient?.getLastSyncAt();
-          if (lastSyncAt == null) {
-            await chatPersistenceClient?.updateLastSyncAt(DateTime.now());
-          }
-        }
+
+        // Sync the persistence client if available
+        if (persistenceEnabled) await sync(cids: cids);
       }
+
       handleEvent(Event(
         type: EventType.connectionRecovered,
         online: true,
@@ -573,34 +563,45 @@ class StreamChatClient {
   Future<void> sync({List<String>? cids, DateTime? lastSyncAt}) {
     return _syncLock.synchronized(() async {
       final channels = cids ?? await chatPersistenceClient?.getChannelCids();
-      if (channels == null || channels.isEmpty) {
-        return;
-      }
+      if (channels == null || channels.isEmpty) return;
 
       final syncAt = lastSyncAt ?? await chatPersistenceClient?.getLastSyncAt();
       if (syncAt == null) {
-        return;
+        logger.info('Fresh sync start: lastSyncAt initialized to now.');
+        return chatPersistenceClient?.updateLastSyncAt(DateTime.now());
       }
 
       try {
+        logger.info('Syncing events since $syncAt for channels: $channels');
+
         final res = await _chatApi.general.sync(channels, syncAt);
-        final events = res.events
-          ..sort((a, b) => a.createdAt.compareTo(b.createdAt));
+        final events = res.events.sorted(
+          (a, b) => a.createdAt.compareTo(b.createdAt),
+        );
 
         for (final event in events) {
-          logger.fine('event.type: ${event.type}');
-          final messageText = event.message?.text;
-          if (messageText != null) {
-            logger.fine('event.message.text: $messageText');
-          }
+          logger.fine('Syncing event: ${event.type}');
           handleEvent(event);
         }
 
-        final now = DateTime.now();
-        _lastSyncedAt = now;
-        chatPersistenceClient?.updateLastSyncAt(now);
-      } catch (e, stk) {
-        logger.severe('Error during sync', e, stk);
+        final updatedSyncAt = events.lastOrNull?.createdAt ?? DateTime.now();
+        return chatPersistenceClient?.updateLastSyncAt(updatedSyncAt);
+      } catch (error, stk) {
+        // If we got a 400 error, it means that either the sync time is too
+        // old or the channel list is too long or too many events need to be
+        // synced. In this case, we should just flush the persistence client
+        // and start over.
+        if (error is StreamChatNetworkError && error.statusCode == 400) {
+          logger.warning(
+            'Failed to sync events due to stale or oversized state. '
+            'Resetting the persistence client to enable a fresh start.',
+          );
+
+          await chatPersistenceClient?.flush();
+          return chatPersistenceClient?.updateLastSyncAt(DateTime.now());
+        }
+
+        logger.warning('Error syncing events', error, stk);
       }
     });
   }
@@ -2193,7 +2194,6 @@ class StreamChatClient {
     // resetting state.
     state.dispose();
     state = ClientState(this);
-    _lastSyncedAt = null;
 
     // resetting credentials.
     _tokenManager.reset();