Expose ThreadTracked/ThreadUntracked events to easier react to threads becoming relevant for local user

Author: sierpinskid

Assets/Plugins/StreamChat/Core/IStreamChatClient.cs

@@ -70,6 +70,32 @@ public interface IStreamChatClient : IDisposable, IStreamChatClientEventsListene
         /// </summary>
         event ChannelMemberRemovedHandler RemovedFromChannelAsMember;
 
+        /// <summary>
+        /// Raised when an <see cref="IStreamThread"/> becomes available locally. Use this to bind
+        /// per-thread UI and to subscribe to the thread's own events such as
+        /// <see cref="IStreamThread.Updated"/>, <see cref="IStreamThread.ReplyReceived"/> and
+        /// <see cref="IStreamThread.ReadStateChanged"/>.
+        ///
+        /// Fires when:
+        ///   - A channel watch (<see cref="GetOrCreateChannelWithIdAsync"/>, <see cref="QueryChannelsAsync"/>)
+        ///     returns a channel that contains threads.
+        ///   - You call <see cref="GetThreadAsync"/> or <see cref="QueryThreadsAsync"/>.
+        ///
+        /// A thread started by another user in a channel you are watching but where you are NOT a
+        /// thread participant will not raise this event - the server delivers only the reply, with
+        /// no thread payload. To learn about such threads call <see cref="QueryThreadsAsync"/>.
+        /// </summary>
+        event StreamThreadChangeHandler ThreadTracked;
+
+        /// <summary>
+        /// Raised when an <see cref="IStreamThread"/> is no longer available locally. Use this to
+        /// tear down per-thread UI.
+        ///
+        /// Fires when the thread's parent message is hard-deleted - the thread is destroyed and
+        /// will no longer appear in <see cref="QueryThreadsAsync"/> results.
+        /// </summary>
+        event StreamThreadChangeHandler ThreadUntracked;
+
         /// <summary>
         /// Current connection state
         /// </summary>

Assets/Plugins/StreamChat/Core/State/Caches/Cache.cs

@@ -51,6 +51,7 @@ public Cache(StreamChatClient stateClient, ISerializer serializer, ILogs logs)
 
             Threads.RegisterDtoIdMapping<StreamThread, ThreadStateResponseInternalDTO>(dto => dto.ParentMessageId);
             Threads.RegisterDtoIdMapping2<StreamThread, ThreadResponseInternalDTO>(dto => dto.ParentMessageId);
+            Threads.RegisterDtoIdMapping3<StreamThread, ThreadStateInternalDTO>(dto => dto.ParentMessageId);
         }
 
         public ICacheRepository<StreamChannel> Channels { get; }

Assets/Plugins/StreamChat/Core/State/Caches/CacheRepository.cs

@@ -11,6 +11,9 @@ namespace StreamChat.Core.State.Caches
     internal sealed class CacheRepository<TStatefulModel> : ICacheRepository<TStatefulModel>
         where TStatefulModel : class, IStreamStatefulModel
     {
+        public event Action<TStatefulModel> Tracked;
+        public event Action<TStatefulModel> Untracked;
+
         public IReadOnlyList<TStatefulModel> AllItems => _statefulModels;
 
         public bool TryGet(string uniqueId, out TStatefulModel trackedObject)
@@ -71,6 +74,7 @@ public TType CreateOrUpdate<TType, TDto>(TDto dto, out bool wasCreated)
         {
             var typedStatefulModel = GetOrCreateStatefulModel<TType, TDto>(dto, out wasCreated);
             typedStatefulModel.UpdateFromDto(dto, _cache);
+            RaiseTrackedIfCreated(typedStatefulModel, wasCreated);
             return typedStatefulModel;
         }
 
@@ -79,6 +83,7 @@ public TType CreateOrUpdate2<TType, TDto>(TDto dto, out bool wasCreated)
         {
             var typedStatefulModel = GetOrCreateStatefulModel<TType, TDto>(dto, out wasCreated);
             typedStatefulModel.UpdateFromDto(dto, _cache);
+            RaiseTrackedIfCreated(typedStatefulModel, wasCreated);
             return typedStatefulModel;
         }
 
@@ -87,6 +92,7 @@ public TType CreateOrUpdate3<TType, TDto>(TDto dto, out bool wasCreated)
         {
             var typedStatefulModel = GetOrCreateStatefulModel<TType, TDto>(dto, out wasCreated);
             typedStatefulModel.UpdateFromDto(dto, _cache);
+            RaiseTrackedIfCreated(typedStatefulModel, wasCreated);
             return typedStatefulModel;
         }
 
@@ -95,6 +101,7 @@ public TType CreateOrUpdate4<TType, TDto>(TDto dto, out bool wasCreated)
         {
             var typedStatefulModel = GetOrCreateStatefulModel<TType, TDto>(dto, out wasCreated);
             typedStatefulModel.UpdateFromDto(dto, _cache);
+            RaiseTrackedIfCreated(typedStatefulModel, wasCreated);
             return typedStatefulModel;
         }
 
@@ -103,9 +110,21 @@ public TType CreateOrUpdate5<TType, TDto>(TDto dto, out bool wasCreated)
         {
             var typedStatefulModel = GetOrCreateStatefulModel<TType, TDto>(dto, out wasCreated);
             typedStatefulModel.UpdateFromDto(dto, _cache);
+            RaiseTrackedIfCreated(typedStatefulModel, wasCreated);
             return typedStatefulModel;
         }
 
+        // Defer Tracked emission until AFTER the first UpdateFromDto so subscribers always observe
+        // a fully-hydrated object. Track() itself runs from the StreamStatefulModelBase constructor,
+        // before any DTO is applied, so emitting from there would surface a blank instance.
+        private void RaiseTrackedIfCreated(TStatefulModel trackedObject, bool wasCreated)
+        {
+            if (wasCreated)
+            {
+                Tracked?.Invoke(trackedObject);
+            }
+        }
+
         /// <summary>
         /// This is called from <see cref="IStreamStatefulModel"/> constructor
         /// </summary>
@@ -137,6 +156,8 @@ public void Remove(TStatefulModel trackedObject)
 
             _statefulModels.Remove(trackedObject);
             _statefulModelById.Remove(trackedObject.UniqueId);
+
+            Untracked?.Invoke(trackedObject);
         }
 
         internal delegate TStatefulModel ConstructorHandler(string uniqueId);

Assets/Plugins/StreamChat/Core/State/Caches/ICacheExt.cs

@@ -85,5 +85,8 @@ public static StreamThread TryCreateOrUpdate(this ICache cache, ThreadStateRespo
 
         public static StreamThread TryCreateOrUpdate(this ICache cache, ThreadResponseInternalDTO dto)
             => dto == null ? null : cache.Threads.CreateOrUpdate2<StreamThread, ThreadResponseInternalDTO>(dto, out _);
+
+        public static StreamThread TryCreateOrUpdate(this ICache cache, ThreadStateInternalDTO dto)
+            => dto == null ? null : cache.Threads.CreateOrUpdate3<StreamThread, ThreadStateInternalDTO>(dto, out _);
     }
 }

Assets/Plugins/StreamChat/Core/State/Caches/ICacheRepository.cs

@@ -10,6 +10,18 @@ namespace StreamChat.Core.State.Caches
     internal interface ICacheRepository<TTrackedObject>
         where TTrackedObject : class, IStreamStatefulModel
     {
+        /// <summary>
+        /// Raised after a tracked object is first hydrated by a CreateOrUpdate call (the first
+        /// time a given DTO id is observed). Fires after the initial UpdateFromDto completes,
+        /// so subscribers always observe a populated object - never a blank instance.
+        /// </summary>
+        event Action<TTrackedObject> Tracked;
+
+        /// <summary>
+        /// Raised after a tracked object is removed from the repository.
+        /// </summary>
+        event Action<TTrackedObject> Untracked;
+
         IReadOnlyList<TTrackedObject> AllItems { get; }
 
         bool TryGet(string uniqueId, out TTrackedObject trackedObject);

Assets/Plugins/StreamChat/Core/StatefulModels/StreamChannel.cs

@@ -684,6 +684,7 @@ void IUpdateableFrom<ChannelStateResponseInternalDTO, StreamChannel>.UpdateFromD
             _pendingMessages.TryReplaceRegularObjectsFromDto(dto.PendingMessages, cache);
             _pinnedMessages.TryReplaceTrackedObjects2(dto.PinnedMessages, cache.Messages);
             _read.TryReplaceRegularObjectsFromDto2(dto.Read, cache);
+            SeedThreadsFromDto(dto.Threads, cache);
             WatcherCount = GetOrDefault(dto.WatcherCount, WatcherCount);
             _watchers.TryAppendUniqueTrackedObjects2(dto.Watchers, cache.Users);
 
@@ -709,6 +710,7 @@ void IUpdateableFrom3<ChannelStateResponseFieldsInternalDTO, StreamChannel>.Upda
             _pendingMessages.TryReplaceRegularObjectsFromDto(dto.PendingMessages, cache);
             _pinnedMessages.TryReplaceTrackedObjects2(dto.PinnedMessages, cache.Messages);
             _read.TryReplaceRegularObjectsFromDto2(dto.Read, cache);
+            SeedThreadsFromDto(dto.Threads, cache);
             WatcherCount = GetOrDefault(dto.WatcherCount, WatcherCount);
             _watchers.TryAppendUniqueTrackedObjects2(dto.Watchers, cache.Users);
 
@@ -1151,6 +1153,25 @@ private Task InternalBanUserAsync(IStreamUser user, bool isShadowBan = false, st
 
         private void SortMessagesByCreatedAt() => _messages.Sort(MessageCreatedAtComparer.Instance);
 
+        // Seed the global Cache.Threads from the channel-state threads carried by the watch
+        // response. Without this the WS thread handlers (thread.updated, notification.thread_message_new,
+        // mark read/unread) would early-return on an unknown thread id and silently drop every
+        // mutation on threads the watcher has not explicitly fetched.
+        // Each first-time insertion fans out as IStreamChatClient.ThreadTracked through the
+        // CacheRepository.Tracked subscription wired up in StreamChatClient.
+        private static void SeedThreadsFromDto(List<ThreadStateInternalDTO> dtos, ICache cache)
+        {
+            if (dtos == null)
+            {
+                return;
+            }
+
+            foreach (var threadDto in dtos)
+            {
+                cache.TryCreateOrUpdate(threadDto);
+            }
+        }
+
         private UpdateChannelRequestInternalDTO GetUpdateRequestWithCurrentData()
             => new UpdateChannelRequestInternalDTO
             {

Assets/Plugins/StreamChat/Core/StatefulModels/StreamThread.cs

@@ -16,6 +16,7 @@ namespace StreamChat.Core.StatefulModels
     internal sealed class StreamThread : StreamStatefulModelBase<StreamThread>,
         IUpdateableFrom<ThreadStateResponseInternalDTO, StreamThread>,
         IUpdateableFrom2<ThreadResponseInternalDTO, StreamThread>,
+        IUpdateableFrom3<ThreadStateInternalDTO, StreamThread>,
         IStreamThread
     {
         public event StreamThreadChangeHandler Updated;
@@ -222,6 +223,66 @@ void IUpdateableFrom<ThreadStateResponseInternalDTO, StreamThread>.UpdateFromDto
             Updated?.Invoke(this);
         }
 
+        // ChannelStateResponse[Fields]InternalDTO carries threads as ThreadStateInternalDTO
+        // (the embedded variant). It differs from ThreadStateResponseInternalDTO mainly in:
+        //   - Channel is the lightweight ChannelInternalDTO (no nested config / messages / read).
+        //     Resolve via cache by CID instead of constructing a partial channel from this payload.
+        //   - Read is List<ReadInternalDTO> (no last_read_message_id), uses the v1 IStateLoadableFrom path.
+        // Replies / parent / participants / custom are otherwise applied with the same semantics
+        // as the response variant so that thread events that arrive after a channel watch can
+        // mutate the now-cached thread instead of being silently dropped.
+        void IUpdateableFrom3<ThreadStateInternalDTO, StreamThread>.UpdateFromDto(
+            ThreadStateInternalDTO dto, ICache cache)
+        {
+            ActiveParticipantCount = GetOrDefault(dto.ActiveParticipantCount, ActiveParticipantCount);
+
+            var cid = dto.Channel?.Cid ?? dto.ChannelCid;
+            if (!string.IsNullOrEmpty(cid) && cache.Channels.TryGet(cid, out var existingChannel))
+            {
+                Channel = existingChannel;
+            }
+
+            ChannelCid = GetOrDefault(cid, ChannelCid);
+            CreatedAt = dto.CreatedAt;
+
+            if (dto.CreatedBy != null)
+            {
+                CreatedBy = cache.TryCreateOrUpdate(dto.CreatedBy);
+            }
+
+            DeletedAt = dto.DeletedAt;
+            LastMessageAt = dto.LastMessageAt;
+
+            if (dto.LatestReplies != null)
+            {
+                _latestReplies.TryReplaceTrackedObjects(dto.LatestReplies, cache.Messages);
+                SortLatestRepliesByCreatedAt();
+            }
+
+            if (dto.ParentMessage != null)
+            {
+                ParentMessage = cache.TryCreateOrUpdate(dto.ParentMessage);
+            }
+
+            ParentMessageId = GetOrDefault(dto.ParentMessageId, ParentMessageId);
+            ParticipantCount = GetOrDefault(dto.ParticipantCount, ParticipantCount);
+            ReplyCount = dto.ReplyCount;
+
+            _read.TryReplaceRegularObjectsFromDto(dto.Read, cache);
+
+            if (dto.ThreadParticipants != null)
+            {
+                _threadParticipants.TryReplaceRegularObjectsFromDto(dto.ThreadParticipants, cache);
+            }
+
+            Title = GetOrDefault(dto.Title, Title);
+            UpdatedAt = dto.UpdatedAt;
+
+            LoadAdditionalCustom(dto.Custom);
+
+            Updated?.Invoke(this);
+        }
+
         void IUpdateableFrom2<ThreadResponseInternalDTO, StreamThread>.UpdateFromDto(
             ThreadResponseInternalDTO dto, ICache cache)
         {

Assets/Plugins/StreamChat/Core/StreamChatClient.cs

@@ -86,6 +86,9 @@ public sealed class StreamChatClient : IStreamChatClient
         public event ChannelMemberAddedHandler AddedToChannelAsMember;
         public event ChannelMemberRemovedHandler RemovedFromChannelAsMember;
 
+        public event StreamThreadChangeHandler ThreadTracked;
+        public event StreamThreadChangeHandler ThreadUntracked;
+
         public const int QueryUsersLimitMaxValue = 30;
         public const int QueryUsersOffsetMaxValue = 1000;
 
@@ -619,6 +622,12 @@ public void Dispose()
                 InternalLowLevelClient.Dispose();
             }
 
+            if (_cache?.Threads != null)
+            {
+                _cache.Threads.Tracked -= OnThreadEnteredCache;
+                _cache.Threads.Untracked -= OnThreadLeftCache;
+            }
+
             _isDisposed = true;
             Disposed?.Invoke();
         }
@@ -754,9 +763,16 @@ private StreamChatClient(IWebsocketClient websocketClient, IHttpClient httpClien
             _cache = new Cache(this, serializer, _logs);
             _pollsApi = new StreamPollsApi(InternalLowLevelClient, _cache);
 
+            _cache.Threads.Tracked += OnThreadEnteredCache;
+            _cache.Threads.Untracked += OnThreadLeftCache;
+
             SubscribeTo(InternalLowLevelClient);
         }
 
+        private void OnThreadEnteredCache(StreamThread thread) => ThreadTracked?.Invoke(thread);
+
+        private void OnThreadLeftCache(StreamThread thread) => ThreadUntracked?.Invoke(thread);
+
         private void InternalDeleteChannel(StreamChannel channel)
         {
             //StreamTodo: mark StreamChannel object as deleted + probably silent clear all internal data?