Merge pull request #1598 from GetStream/release/v6.3.0

.github/workflows/dart_code_metrics.yaml

@@ -1,7 +1,7 @@
 name: Dart Code Metrics
 
 env:
-  flutter_version: "3.10.0"
+  flutter_version: "3.10.4"
   folders: "lib, test"
 
 on:

.github/workflows/stream_flutter_workflow.yml

@@ -2,7 +2,7 @@ name: stream_flutter_workflow
 
 env:
   ACTIONS_ALLOW_UNSECURE_COMMANDS: 'true'
-  flutter_version: "3.10.0"
+  flutter_version: "3.10.4"
 
 on:
   pull_request:
@@ -17,7 +17,7 @@ on:
     branches:
       - master
       - develop
-      
+
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
   cancel-in-progress: true

packages/stream_chat/CHANGELOG.md

@@ -1,3 +1,14 @@
+## 6.3.0
+
+🐞 Fixed
+
+- [[#1585]](https://github.com/GetStream/stream-chat-flutter/issues/1585) Fixed channels left not being removed from
+  the persistent storage.
+
+🔄 Changed
+
+- Updated `dio` dependency to `^5.2.0`.
+
 ## 6.2.0
 
 🐞 Fixed
@@ -8,6 +19,9 @@
 ✅ Added
 
 - Added support for `ChatPersistenceClient.isConnected` for checking if the client is connected to the database.
+- Added support for `ChatPersistenceClient.userId` for getting the current connected user id.
+- Added two new methods `ChatPersistenceClient.disconnect` and `ChatPersistenceClient.connect` for disconnecting and
+  connecting to the database.
 
 ## 6.1.0
 

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

@@ -1379,13 +1379,14 @@ class Channel {
       this.state?.updateChannelState(updatedState);
       return updatedState;
     } catch (e) {
-      if (!_client.persistenceEnabled) {
-        rethrow;
+      if (_client.persistenceEnabled) {
+        return _client.chatPersistenceClient!.getChannelStateByCid(
+          cid!,
+          messagePagination: messagesPagination,
+        );
       }
-      return _client.chatPersistenceClient!.getChannelStateByCid(
-        cid!,
-        messagePagination: messagesPagination,
-      );
+
+      rethrow;
     }
   }
 
@@ -1841,9 +1842,7 @@ class ChannelClientState {
 
   /// [isUpToDate] flag count as a stream.
   Stream<bool> get isUpToDateStream => _isUpToDateController.stream;
-
-  final BehaviorSubject<bool> _isUpToDateController =
-      BehaviorSubject.seeded(true);
+  final _isUpToDateController = BehaviorSubject.seeded(true);
 
   /// The retry queue associated to this channel.
   late final RetryQueue _retryQueue;

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

@@ -125,10 +125,6 @@ class StreamChatClient {
   final _tokenManager = TokenManager();
   final _connectionIdManager = ConnectionIdManager();
 
-  set chatPersistenceClient(ChatPersistenceClient? value) {
-    _originalChatPersistenceClient = value;
-  }
-
   /// Default user agent for all requests
   static String defaultUserAgent =
       'stream-chat-dart-client-${CurrentPlatform.name}';
@@ -139,15 +135,15 @@ class StreamChatClient {
   /// The current package version
   static const packageVersion = PACKAGE_VERSION;
 
-  ChatPersistenceClient? _originalChatPersistenceClient;
-
   /// Chat persistence client
-  ChatPersistenceClient? get chatPersistenceClient => _chatPersistenceClient;
+  ChatPersistenceClient? chatPersistenceClient;
 
-  ChatPersistenceClient? _chatPersistenceClient;
-
-  /// Whether the chat persistence is available or not
-  bool get persistenceEnabled => _chatPersistenceClient != null;
+  /// Returns `True` if the [chatPersistenceClient] is available and connected.
+  /// Otherwise, returns `False`.
+  bool get persistenceEnabled {
+    final client = chatPersistenceClient;
+    return client != null && client.isConnected;
+  }
 
   late final RetryPolicy _retryPolicy;
 
@@ -324,27 +320,68 @@ class StreamChatClient {
     final ownUser = OwnUser.fromUser(user);
     state.currentUser = ownUser;
 
-    if (!connectWebSocket) return ownUser;
-
     try {
-      if (_originalChatPersistenceClient != null) {
-        _chatPersistenceClient = _originalChatPersistenceClient;
-        await _chatPersistenceClient!.connect(ownUser.id);
+      // Connect to persistence client if its set.
+      if (chatPersistenceClient != null) {
+        await openPersistenceConnection(ownUser);
       }
-      final connectedUser = await openConnection(
-        includeUserDetailsInConnectCall: true,
-      );
-      return state.currentUser = connectedUser;
+
+      // Connect to websocket if [connectWebSocket] is true.
+      //
+      // This is useful when you want to connect to websocket
+      // at a later stage or use the client in connection-less mode.
+      if (connectWebSocket) {
+        final connectedUser = await openConnection(
+          includeUserDetailsInConnectCall: true,
+        );
+        state.currentUser = connectedUser;
+      }
+
+      return state.currentUser!;
     } catch (e, stk) {
       if (e is StreamWebSocketError && e.isRetriable) {
-        final event = await _chatPersistenceClient?.getConnectionInfo();
+        final event = await chatPersistenceClient?.getConnectionInfo();
         if (event != null) return ownUser.merge(event.me);
       }
       logger.severe('error connecting user : ${ownUser.id}', e, stk);
       rethrow;
     }
   }
 
+  /// Connects the [chatPersistenceClient] to the given [user].
+  Future<void> openPersistenceConnection(User user) async {
+    final client = chatPersistenceClient;
+    if (client == null) {
+      throw const StreamChatError('Chat persistence client is not set');
+    }
+
+    if (client.isConnected) {
+      // If the persistence client is already connected to the userId,
+      // we don't need to connect again.
+      if (client.userId == user.id) return;
+
+      throw const StreamChatError('''
+        Chat persistence client is already connected to a different user,
+        please close the connection before connecting a new one.''');
+    }
+
+    // Connect the persistence client to the userId.
+    return client.connect(user.id);
+  }
+
+  /// Disconnects the [chatPersistenceClient] from the current user.
+  Future<void> closePersistenceConnection({bool flush = false}) async {
+    final client = chatPersistenceClient;
+    // If the persistence client is never connected, we don't need to close it.
+    if (client == null || !client.isConnected) {
+      logger.info('Chat persistence client is not connected');
+      return;
+    }
+
+    // Disconnect the persistence client.
+    return client.disconnect(flush: flush);
+  }
+
   /// Creates a new WebSocket connection with the current user.
   /// If [includeUserDetailsInConnectCall] is true it will include the current
   /// user details in the connect call.
@@ -422,7 +459,7 @@ class StreamChatClient {
     final connectionId = event.connectionId;
     if (connectionId != null) {
       _connectionIdManager.setConnectionId(connectionId);
-      _chatPersistenceClient?.updateConnectionInfo(event);
+      chatPersistenceClient?.updateConnectionInfo(event);
     }
   }
 
@@ -460,9 +497,9 @@ class StreamChatClient {
         // channels are empty, assuming it's a fresh start
         // and making sure `lastSyncAt` is initialized
         if (persistenceEnabled) {
-          final lastSyncAt = await _chatPersistenceClient?.getLastSyncAt();
+          final lastSyncAt = await chatPersistenceClient?.getLastSyncAt();
           if (lastSyncAt == null) {
-            await _chatPersistenceClient?.updateLastSyncAt(DateTime.now());
+            await chatPersistenceClient?.updateLastSyncAt(DateTime.now());
           }
         }
       }
@@ -493,13 +530,12 @@ class StreamChatClient {
   /// Will automatically fetch [cids] and [lastSyncedAt] if [persistenceEnabled]
   Future<void> sync({List<String>? cids, DateTime? lastSyncAt}) {
     return synchronized(() async {
-      final channels = cids ?? await _chatPersistenceClient?.getChannelCids();
+      final channels = cids ?? await chatPersistenceClient?.getChannelCids();
       if (channels == null || channels.isEmpty) {
         return;
       }
 
-      final syncAt =
-          lastSyncAt ?? await _chatPersistenceClient?.getLastSyncAt();
+      final syncAt = lastSyncAt ?? await chatPersistenceClient?.getLastSyncAt();
       if (syncAt == null) {
         return;
       }
@@ -520,7 +556,7 @@ class StreamChatClient {
 
         final now = DateTime.now();
         _lastSyncedAt = now;
-        _chatPersistenceClient?.updateLastSyncAt(now);
+        chatPersistenceClient?.updateLastSyncAt(now);
       } catch (e, stk) {
         logger.severe('Error during sync', e, stk);
       }
@@ -532,9 +568,8 @@ class StreamChatClient {
   /// Requests channels with a given query.
   Stream<List<Channel>> queryChannels({
     Filter? filter,
-    @Deprecated('''
-    sort has been deprecated. 
-    Please use channelStateSort instead.''') List<SortOption<ChannelModel>>? sort,
+    @Deprecated('Use channelStateSort instead.')
+    List<SortOption<ChannelModel>>? sort,
     List<SortOption<ChannelState>>? channelStateSort,
     bool state = true,
     bool watch = true,
@@ -679,7 +714,7 @@ class StreamChatClient {
 
     final updateData = _mapChannelStateToChannel(channels);
 
-    await _chatPersistenceClient?.updateChannelQueries(
+    await chatPersistenceClient?.updateChannelQueries(
       filter,
       channels.map((c) => c.channel!.cid).toList(),
       clearQueryCache: paginationParams.offset == 0,
@@ -694,11 +729,12 @@ class StreamChatClient {
     Filter? filter,
     @Deprecated('''
     sort has been deprecated. 
-    Please use channelStateSort instead.''') List<SortOption<ChannelModel>>? sort,
+    Please use channelStateSort instead.''')
+    List<SortOption<ChannelModel>>? sort,
     List<SortOption<ChannelState>>? channelStateSort,
     PaginationParams paginationParams = const PaginationParams(),
   }) async {
-    final offlineChannels = (await _chatPersistenceClient?.getChannelStates(
+    final offlineChannels = (await chatPersistenceClient?.getChannelStates(
           filter: filter,
           // ignore: deprecated_member_use_from_same_package
           sort: sort,
@@ -1362,7 +1398,7 @@ class StreamChatClient {
     final response =
         await _chatApi.message.deleteMessage(messageId, hard: hard);
     if (hard == true) {
-      await _chatPersistenceClient?.deleteMessageById(messageId);
+      await chatPersistenceClient?.deleteMessageById(messageId);
     }
     return response;
   }
@@ -1468,34 +1504,33 @@ class StreamChatClient {
   Future<void> disconnectUser({bool flushChatPersistence = false}) async {
     logger.info('Disconnecting user : ${state.currentUser?.id}');
 
-    // resetting state
+    // resetting state.
     state.dispose();
     state = ClientState(this);
     _lastSyncedAt = null;
 
-    // resetting credentials
+    // resetting credentials.
     _tokenManager.reset();
     _connectionIdManager.reset();
 
-    // disconnecting persistence client
-    await _chatPersistenceClient?.disconnect(flush: flushChatPersistence);
-    _chatPersistenceClient = null;
+    // closing persistence connection.
+    await closePersistenceConnection(flush: flushChatPersistence);
 
     // closing web-socket connection
-    closeConnection();
+    return closeConnection();
   }
 
   /// Call this function to dispose the client
   Future<void> dispose() async {
     logger.info('Disposing new StreamChatClient');
 
-    // disposing state
+    // disposing state.
     state.dispose();
 
-    // disconnecting persistence client
-    await _chatPersistenceClient?.disconnect();
+    // closing persistence connection.
+    await closePersistenceConnection();
 
-    // closing web-socket connection
+    // closing web-socket connection.
     closeConnection();
 
     await _eventController.close();
@@ -1541,6 +1576,8 @@ class ClientState {
         currentUser = currentUser?.copyWith(totalUnreadCount: count);
       }));
 
+    _listenChannelLeft();
+
     _listenChannelDeleted();
 
     _listenChannelHidden();
@@ -1601,12 +1638,30 @@ class ClientState {
     );
   }
 
+  void _listenChannelLeft() {
+    _eventsSubscription?.add(
+      _client
+          .on(
+        EventType.memberRemoved,
+        EventType.notificationRemovedFromChannel,
+      )
+          .listen((event) async {
+        final isCurrentUser = event.user!.id == currentUser!.id;
+        if (isCurrentUser) {
+          final eventChannel = event.channel!;
+          await _client.chatPersistenceClient
+              ?.deleteChannels([eventChannel.cid]);
+          channels.remove(eventChannel.cid)?.dispose();
+        }
+      }),
+    );
+  }
+
   void _listenChannelDeleted() {
     _eventsSubscription?.add(
       _client
           .on(
         EventType.channelDeleted,
-        EventType.notificationRemovedFromChannel,
         EventType.notificationChannelDeleted,
       )
           .listen((Event event) async {