feat(llc, core, ui, localization): threads v2 (#2076)

Co-authored-by: xsahil03x <[email protected]>

melos.yaml

@@ -37,7 +37,7 @@ command:
       http_parser: ^4.0.2
       image_gallery_saver_plus: ^3.0.5
       image_picker: ^1.1.2
-      image_size_getter: ^2.1.3
+      image_size_getter: ^2.3.0
       jiffy: ^6.2.1
       jose: ^0.3.4
       json_annotation: ^4.9.0

packages/stream_chat/CHANGELOG.md

@@ -1,3 +1,9 @@
+## Upcoming
+
+✅ Added
+
+- Added support for Threads v2 feature, which allows users to query all the threads.
+
 ## 9.0.0
 
 ✅ Added

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

@@ -1519,6 +1519,18 @@ class Channel {
     return response;
   }
 
+  /// Mark the thread with [threadId] in the channel as read.
+  Future<EmptyResponse> markThreadRead(String threadId) {
+    _checkInitialized();
+    return client.markThreadRead(id!, type, threadId);
+  }
+
+  /// Mark the thread with [threadId] in the channel as unread.
+  Future<EmptyResponse> markThreadUnread(String threadId) {
+    _checkInitialized();
+    return client.markThreadUnread(id!, type, threadId);
+  }
+
   void _initState(ChannelState channelState) {
     state = ChannelClientState(this, channelState);
 

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

@@ -196,7 +196,7 @@ class StreamChatClient {
 
   StreamSubscription<ConnectionStatus>? _connectionStatusSubscription;
 
-  final _eventController = BehaviorSubject<Event>();
+  final _eventController = PublishSubject<Event>();
 
   /// Stream of [Event] coming from [_ws] connection
   /// Listen to this or use the [on] method to filter specific event types
@@ -491,10 +491,12 @@ class StreamChatClient {
     final previousState = wsConnectionStatus;
     final currentState = _wsConnectionStatus = status;
 
-    handleEvent(Event(
-      type: EventType.connectionChanged,
-      online: status == ConnectionStatus.connected,
-    ));
+    if (previousState != currentState) {
+      handleEvent(Event(
+        type: EventType.connectionChanged,
+        online: status == ConnectionStatus.connected,
+      ));
+    }
 
     if (currentState == ConnectionStatus.connected &&
         previousState != ConnectionStatus.connected) {
@@ -1213,6 +1215,32 @@ class StreamChatClient {
         messageId,
       );
 
+  /// Mark the thread with [threadId] in the channel with [channelId] of type
+  /// [channelType] as read.
+  Future<EmptyResponse> markThreadRead(
+    String channelId,
+    String channelType,
+    String threadId,
+  ) =>
+      _chatApi.channel.markThreadRead(
+        channelId,
+        channelType,
+        threadId,
+      );
+
+  /// Mark the thread with [threadId] in the channel with [channelId] of type
+  /// [channelType] as unread.
+  Future<EmptyResponse> markThreadUnread(
+    String channelId,
+    String channelType,
+    String threadId,
+  ) =>
+      _chatApi.channel.markThreadUnread(
+        channelId,
+        channelType,
+        threadId,
+      );
+
   /// Creates a new Poll
   Future<CreatePollResponse> createPoll(Poll poll) =>
       _chatApi.polls.createPoll(poll);
@@ -1659,6 +1687,43 @@ class StreamChatClient {
   Future<OGAttachmentResponse> enrichUrl(String url) =>
       _chatApi.general.enrichUrl(url);
 
+  /// Queries threads with the given [options] and [pagination] params.
+  Future<QueryThreadsResponse> queryThreads({
+    ThreadOptions options = const ThreadOptions(),
+    PaginationParams pagination = const PaginationParams(),
+  }) =>
+      _chatApi.threads.queryThreads(
+        options: options,
+        pagination: pagination,
+      );
+
+  /// Retrieves a thread with the given [messageId].
+  ///
+  /// Optionally pass [options] to limit the response.
+  Future<GetThreadResponse> getThread(
+    String messageId, {
+    ThreadOptions options = const ThreadOptions(),
+  }) =>
+      _chatApi.threads.getThread(
+        messageId,
+        options: options,
+      );
+
+  /// Partially updates the thread with the given [messageId].
+  ///
+  /// Use [set] to define values to be set.
+  /// Use [unset] to define values to be unset.
+  Future<UpdateThreadResponse> partialUpdateThread(
+    String messageId, {
+    Map<String, Object?>? set,
+    List<String>? unset,
+  }) =>
+      _chatApi.threads.partialUpdateThread(
+        messageId,
+        set: set,
+        unset: unset,
+      );
+
   /// Closes the [_ws] connection and resets the [state]
   /// If [flushChatPersistence] is true the client deletes all offline
   /// user's data.
@@ -1712,30 +1777,32 @@ class ClientState {
       cancelEventSubscription();
     }
 
-    _eventsSubscription = CompositeSubscription();
-    _eventsSubscription!
-      ..add(_client
-          .on()
-          .where((event) =>
-              event.me != null && event.type != EventType.healthCheck)
-          .map((e) => e.me!)
-          .listen((user) {
-        currentUser = currentUser?.merge(user) ?? user;
-      }))
-      ..add(_client
-          .on()
-          .map((event) => event.unreadChannels)
-          .whereType<int>()
-          .listen((count) {
-        currentUser = currentUser?.copyWith(unreadChannels: count);
-      }))
-      ..add(_client
-          .on()
-          .map((event) => event.totalUnreadCount)
-          .whereType<int>()
-          .listen((count) {
-        currentUser = currentUser?.copyWith(totalUnreadCount: count);
-      }));
+    _eventsSubscription = CompositeSubscription()
+      ..add(
+        _client.on().listen((event) {
+          // Update the current user only if the event is not a health check.
+          if (event.me case final user?) {
+            if (event.type != EventType.healthCheck) {
+              currentUser = currentUser?.merge(user) ?? user;
+            }
+          }
+
+          // Update the total unread count.
+          if (event.totalUnreadCount case final count?) {
+            currentUser = currentUser?.copyWith(totalUnreadCount: count);
+          }
+
+          // Update the unread channels count.
+          if (event.unreadChannels case final count?) {
+            currentUser = currentUser?.copyWith(unreadChannels: count);
+          }
+
+          // Update the unread threads count.
+          if (event.unreadThreads case final count?) {
+            currentUser = currentUser?.copyWith(unreadThreads: count);
+          }
+        }),
+      );
 
     _listenChannelLeft();
 
@@ -1873,6 +1940,12 @@ class ClientState {
   /// The current unread channels count as a stream
   Stream<int> get unreadChannelsStream => _unreadChannelsController.stream;
 
+  /// The current unread thread count.
+  int get unreadThreads => _unreadThreadsController.value;
+
+  /// The current unread threads count as a stream.
+  Stream<int> get unreadThreadsStream => _unreadThreadsController.stream;
+
   /// The current total unread messages count
   int get totalUnreadCount => _totalUnreadCountController.value;
 
@@ -1909,28 +1982,32 @@ class ClientState {
   }
 
   void _computeUnreadCounts(OwnUser? user) {
-    final totalUnreadCount = user?.totalUnreadCount;
-    if (totalUnreadCount != null) {
-      _totalUnreadCountController.add(totalUnreadCount);
+    if (user?.totalUnreadCount case final count?) {
+      _totalUnreadCountController.add(count);
+    }
+
+    if (user?.unreadChannels case final count?) {
+      _unreadChannelsController.add(count);
     }
 
-    final unreadChannels = user?.unreadChannels;
-    if (unreadChannels != null) {
-      _unreadChannelsController.add(unreadChannels);
+    if (user?.unreadThreads case final count?) {
+      _unreadThreadsController.add(count);
     }
   }
 
   final _channelsController = BehaviorSubject<Map<String, Channel>>.seeded({});
   final _currentUserController = BehaviorSubject<OwnUser?>();
   final _usersController = BehaviorSubject<Map<String, User>>.seeded({});
   final _unreadChannelsController = BehaviorSubject<int>.seeded(0);
+  final _unreadThreadsController = BehaviorSubject<int>.seeded(0);
   final _totalUnreadCountController = BehaviorSubject<int>.seeded(0);
 
   /// Call this method to dispose this object
   void dispose() {
     cancelEventSubscription();
     _currentUserController.close();
     _unreadChannelsController.close();
+    _unreadThreadsController.close();
     _totalUnreadCountController.close();
 
     final channels = [...this.channels.keys];

packages/stream_chat/lib/src/core/api/channel_api.dart

@@ -337,6 +337,32 @@ class ChannelApi {
     return EmptyResponse.fromJson(response.data);
   }
 
+  /// Mark the provided [threadId] of the channel as read.
+  Future<EmptyResponse> markThreadRead(
+    String channelId,
+    String channelType,
+    String threadId,
+  ) async {
+    final response = await _client.post(
+      '${_getChannelUrl(channelId, channelType)}/read',
+      data: {'thread_id': threadId},
+    );
+    return EmptyResponse.fromJson(response.data);
+  }
+
+  /// Mark the provided [threadId] of the channel as unread.
+  Future<EmptyResponse> markThreadUnread(
+    String channelId,
+    String channelType,
+    String threadId,
+  ) async {
+    final response = await _client.post(
+      '${_getChannelUrl(channelId, channelType)}/unread',
+      data: {'thread_id': threadId},
+    );
+    return EmptyResponse.fromJson(response.data);
+  }
+
   /// Stop watching the channel
   Future<EmptyResponse> stopWatching(
     String channelId,

packages/stream_chat/lib/src/core/api/requests.dart

@@ -207,3 +207,43 @@ class PartialUpdateUserRequest extends Equatable {
   @override
   List<Object?> get props => [id, set, unset];
 }
+
+/// {@template threadOptions}
+/// Options for querying threads.
+/// {@endtemplate}
+@JsonSerializable(createFactory: false)
+class ThreadOptions extends Equatable {
+  /// {@macro threadOptions}
+  const ThreadOptions({
+    this.watch = true,
+    this.replyLimit = 2,
+    this.participantLimit = 100,
+    this.memberLimit = 100,
+  });
+
+  /// If true, the client will watch for changes in the thread.
+  ///
+  /// Defaults to true.
+  final bool watch;
+
+  /// The number of most recent replies to return per thread.
+  ///
+  /// Defaults to 2.
+  final int replyLimit;
+
+  /// The number of thread participants to return per thread.
+  ///
+  /// Defaults to 100.
+  final int participantLimit;
+
+  /// The number of members to return per thread.
+  ///
+  /// Defaults to 100.
+  final int memberLimit;
+
+  /// Serialize model to json
+  Map<String, dynamic> toJson() => _$ThreadOptionsToJson(this);
+
+  @override
+  List<Object?> get props => [watch, replyLimit, participantLimit, memberLimit];
+}