diff --git a/Assets/Plugins/StreamChat/Core/Configs/IStreamClientConfig.cs b/Assets/Plugins/StreamChat/Core/Configs/IStreamClientConfig.cs index 69887678..2af70921 100644 --- a/Assets/Plugins/StreamChat/Core/Configs/IStreamClientConfig.cs +++ b/Assets/Plugins/StreamChat/Core/Configs/IStreamClientConfig.cs @@ -36,5 +36,12 @@ public interface IStreamClientConfig /// Does not change server history. See . /// MessageCacheWindow DefaultMessageCacheWindow { get; set; } + + /// + /// How the client restores local state after the websocket reconnects. + /// Default is . + /// See for the other options. + /// + StateRecoveryStrategy StateRecoveryStrategy { get; set; } } } \ No newline at end of file diff --git a/Assets/Plugins/StreamChat/Core/Configs/StateRecoveryStrategy.cs b/Assets/Plugins/StreamChat/Core/Configs/StateRecoveryStrategy.cs new file mode 100644 index 00000000..f10497cd --- /dev/null +++ b/Assets/Plugins/StreamChat/Core/Configs/StateRecoveryStrategy.cs @@ -0,0 +1,58 @@ +namespace StreamChat.Core.Configs +{ + /// + /// How restores local state after the websocket reconnects. + /// Set this on . + /// + /// + /// After a reconnect, the server always drops the old watches. + /// and start watching those channels again. + /// does not - you must do that yourself. + /// + public enum StateRecoveryStrategy + { + /// + /// Default. + /// + /// Missed events are replayed as normal events + /// (, + /// , and so on). + /// The client also refreshes those channels and starts watching them again. + /// + /// Use this if your UI updates from per-event callbacks. + /// After a long disconnect on a busy channel this can replay many events at once + /// and cause a hitch. + /// + ReplayEvents = 0, + + /// + /// Same recovery as , but missed events update local state + /// without raising per-event callbacks. Listen to + /// and rebuild your UI from channel state + /// (, and similar). + /// + /// Some events are still raised because they are not stored in channel state: + /// , + /// , and membership or invite notifications. + /// + /// Use this if a long disconnect causes a hitch when the app resumes. + /// + BatchStateUpdate = 1, + + /// + /// The SDK does not restore state after a reconnect. + /// It does not refresh channels, does not start watching them again, + /// and does not raise . + /// Local state and stay as they were + /// before the disconnect. + /// + /// Use this only if you handle recovery yourself. When the connection is + /// again, query the channels you need. + /// Example: QueryChannelsAsync(new[] { ChannelFilter.Cid.In(cids) }, limit: 30). + /// That call refreshes state and starts watching. + /// Do not use here: + /// is still true, so WatchAsync does nothing. + /// + Disabled = 2, + } +} diff --git a/Assets/Plugins/StreamChat/Core/Configs/StateRecoveryStrategy.cs.meta b/Assets/Plugins/StreamChat/Core/Configs/StateRecoveryStrategy.cs.meta new file mode 100644 index 00000000..5ce3c44b --- /dev/null +++ b/Assets/Plugins/StreamChat/Core/Configs/StateRecoveryStrategy.cs.meta @@ -0,0 +1,2 @@ +fileFormatVersion: 2 +guid: deb96db72983f9e4ca61a28cb444a7be \ No newline at end of file diff --git a/Assets/Plugins/StreamChat/Core/Configs/StreamClientConfig.cs b/Assets/Plugins/StreamChat/Core/Configs/StreamClientConfig.cs index 54b698f9..a48396d2 100644 --- a/Assets/Plugins/StreamChat/Core/Configs/StreamClientConfig.cs +++ b/Assets/Plugins/StreamChat/Core/Configs/StreamClientConfig.cs @@ -12,5 +12,7 @@ public class StreamClientConfig : IStreamClientConfig public bool OptimisticMessageInsert { get; set; } = true; public MessageCacheWindow DefaultMessageCacheWindow { get; set; } = null; + + public StateRecoveryStrategy StateRecoveryStrategy { get; set; } = Configs.StateRecoveryStrategy.ReplayEvents; } } \ No newline at end of file diff --git a/Assets/Plugins/StreamChat/Core/IStreamChatClient.cs b/Assets/Plugins/StreamChat/Core/IStreamChatClient.cs index e72e016b..63eabffe 100644 --- a/Assets/Plugins/StreamChat/Core/IStreamChatClient.cs +++ b/Assets/Plugins/StreamChat/Core/IStreamChatClient.cs @@ -33,6 +33,24 @@ public interface IStreamChatClient : IDisposable, IStreamChatClientEventsListene /// event Action Disconnected; + /// + /// Raised once when reconnect recovery is done. + /// This includes full success, partial success, and cases with nothing to recover. + /// Not raised on the first login. + /// Not raised when is + /// . + /// + /// Channels in have fresh state and are watched again. + /// Cids in are still stale and not watched. + /// + /// Use this to rebuild your UI from channel state after a disconnect. + /// You need this for , + /// because per-event callbacks do not fire during recovery. + /// It is also useful for , + /// because some state updates after reconnect do not raise per-message callbacks. + /// + event StateRecoveredHandler StateRecovered; + /// /// Event fired when connection state with Stream Chat server has changed /// @@ -129,6 +147,12 @@ public interface IStreamChatClient : IDisposable, IStreamChatClientEventsListene /// methods may not be watched - check on a specific /// channel to know its state. /// + /// + /// This list is cleared when the connection drops, because the server drops every watch. + /// Recovery fills it again. + /// If you read this list to restore watches yourself, do it before the disconnect, + /// or use so the list is not cleared. + /// /// IReadOnlyList WatchedChannels { get; } diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/HistorySyncApplyResult.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/HistorySyncApplyResult.cs new file mode 100644 index 00000000..a1a4419c --- /dev/null +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/HistorySyncApplyResult.cs @@ -0,0 +1,21 @@ +using System; + +namespace StreamChat.Core.LowLevelClient +{ + /// + /// Result of applying one /sync history batch. + /// + internal sealed class HistorySyncApplyResult + { + /// + /// created_at of the newest event that applied, or null. + /// The /sync last_sync_at cursor advances to this. + /// + public DateTimeOffset? MaxAppliedCreatedAt { get; set; } + + /// + /// Events that threw while applying. Skipped, not retried in this batch. + /// + public int FailedEventCount { get; set; } + } +} diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/HistorySyncApplyResult.cs.meta b/Assets/Plugins/StreamChat/Core/LowLevelClient/HistorySyncApplyResult.cs.meta new file mode 100644 index 00000000..f131c2ae --- /dev/null +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/HistorySyncApplyResult.cs.meta @@ -0,0 +1,2 @@ +fileFormatVersion: 2 +guid: 59eb11f6ed758d54fa5b796cd4f8a5c7 \ No newline at end of file diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index 45a29261..1afb9459 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -425,7 +425,7 @@ public void Update(float deltaTime) #if STREAM_DEBUG_ENABLED _logs.Info(_authCredentials.UserId + " WS message: " + msg); #endif - HandleNewWebsocketMessage(msg); + HandleNewWebsocketMessage(msg, isLiveEvent: true); } } @@ -443,43 +443,150 @@ public void SetReconnectStrategySettings(ReconnectStrategy reconnectStrategy, fl public async Task FetchAndProcessEventsSinceLastReceivedEvent(IEnumerable channelCids) { - if (!channelCids.Any() || !_disconnectionLastEventReceivedAt.HasValue) + var response = await TrySyncHistoryAsync(channelCids); + ReplayHistoryEvents(response?.Events); + } + + /// + /// /sync counts events across all requested cids. The server rejects the request + /// at about 1000 events. Fewer cids makes success more likely. Matches Swift and Android (100). + /// + internal const int MaxSyncChannelCids = 100; + + /// + /// Best-effort /sync for up to channels. + /// Returns null when there is no sync point, or when it is older than 30 days. + /// + internal async Task TrySyncHistoryAsync(IEnumerable channelCids) + { + if (channelCids == null || !_disconnectionLastEventReceivedAt.HasValue) { - return; + return null; } var lastEventReceivedAt = _disconnectionLastEventReceivedAt.Value; - // Check if less than 30 days. Past that the server rejects LastSyncAt, so this only - // skips a request that was certain to fail — it is not a recovery path. The SDK has no - // re-hydrate fallback of its own, so bridging a gap this large is the consumer's job. - TimeSpan diff = _timeService.Now - lastEventReceivedAt; - if (diff.TotalDays > 30) + if ((_timeService.Now - lastEventReceivedAt).TotalDays > 30) + { + return null; + } + + using (new ListPoolScope(out var tempCids)) + { + foreach (var cid in channelCids) + { + if (tempCids.Count == MaxSyncChannelCids) + { + break; + } + + tempCids.Add(cid); + } + + if (tempCids.Count == 0) + { + return null; + } + + return await ChannelApi.SyncAsync(new SyncRequest + { + ChannelCids = tempCids, + LastSyncAt = lastEventReceivedAt, + Watch = true, + + // Lets recovery skip channels the server will never return again + // (deleted, or the user lost access while offline). + WithInaccessibleCids = true, + }); + } + } + + /// + /// Replay history events as live events, so public callbacks fire the same way. + /// Used by . + /// + internal void ReplayHistoryEvents(IEnumerable events) + { + if (events == null) { return; } - //StreamTodo: according to Android SDK there's an error if there are > 1000 events + foreach (var e in events) + { + // Isolated in the handler so one bad event does not stop the rest. + HandleNewWebsocketMessage(e); + } + } - var response = await ChannelApi.SyncAsync(new SyncRequest + /// + /// Apply history events to local state without raising per-event public callbacks. + /// Used by . + /// Matches Android isFromHistorySync and Swift postNotifications: false. + /// + internal HistorySyncApplyResult ApplyHistoryEvents(IEnumerable events) + { + var result = new HistorySyncApplyResult(); + if (events == null) { - ChannelCids = channelCids.ToList(), - LastSyncAt = lastEventReceivedAt, - Watch = true, - }); + return result; + } - if (response.Events.Count == 0) + _isApplyingHistoryEvents = true; + _historyMaxAppliedCreatedAt = null; + + try { - return; + foreach (var e in events) + { + try + { + HandleNewWebsocketMessage(e); + } + catch (Exception ex) + { + // Do not abort the batch. The /sync last_sync_at cursor only moves to + // events that applied, so a failed event is retried on the next reconnect. + result.FailedEventCount++; + _logs.Exception(ex); + } + } } + finally + { + result.MaxAppliedCreatedAt = _historyMaxAppliedCreatedAt; + _historyMaxAppliedCreatedAt = null; + _isApplyingHistoryEvents = false; + + if (result.MaxAppliedCreatedAt.HasValue) + { + TryAdvanceLastEventReceivedAt(result.MaxAppliedCreatedAt.Value, HistorySyncWatermarkSource); + } + } + + return result; + } - foreach (var e in response.Events) + /// + /// True while is running. + /// + internal bool IsApplyingHistoryEvents => _isApplyingHistoryEvents; + + private const string HistorySyncWatermarkSource = "history.sync"; + + // Track the newest created_at applied in this batch. + // _lastEventReceivedAt (the /sync last_sync_at cursor) is advanced once at the end. + // Advancing per event would skip a failed event on the next reconnect. + private void TrackMaxAppliedCreatedAt(DateTimeOffset createdAt) + { + if (createdAt == DateTimeOffset.MinValue) { - // StreamTodo: check if we can not serialized this again. Investigate adding a custom EventsJsonConverter that would populate the list as serialized strings - var serializedMsg = _serializer.Serialize(e); + return; + } - //StreamTodo: try block? - HandleNewWebsocketMessage(serializedMsg); + if (!_historyMaxAppliedCreatedAt.HasValue || createdAt > _historyMaxAppliedCreatedAt.Value) + { + _historyMaxAppliedCreatedAt = createdAt; } } @@ -577,8 +684,8 @@ internal async Task ConnectUserAsync(string apiKey, string u private readonly IStreamClientConfig _config; private readonly ReconnectScheduler _reconnectScheduler; - private readonly Dictionary> _eventKeyToHandler = - new Dictionary>(); + private readonly Dictionary> _eventKeyToHandler = + new Dictionary>(); private readonly object _websocketConnectionFailedFlagLock = new object(); private readonly object _websocketDisconnectedFlagLock = new object(); @@ -615,6 +722,9 @@ internal async Task ConnectUserAsync(string apiKey, string u /// private DateTimeOffset? _disconnectionLastEventReceivedAt; + private bool _isApplyingHistoryEvents; + private DateTimeOffset? _historyMaxAppliedCreatedAt; + private async Task RefreshAuthTokenFromProvider() { #if STREAM_DEBUG_ENABLED @@ -983,7 +1093,7 @@ private void RegisterEventType(string key, return; } - _eventKeyToHandler.Add(key, serializedContent => + _eventKeyToHandler.Add(key, payload => { try { @@ -991,45 +1101,149 @@ private void RegisterEventType(string key, var ignoreKeys = new[] { WSEventType.HealthCheck }; if (!ignoreKeys.Contains(key)) { - _logs.Warning("WS event received KEY: " + key + " CONTENT: " + serializedContent); + _logs.Warning("WS event received KEY: " + key + " CONTENT: " + payload); } #endif - var eventObj = DeserializeEvent(serializedContent, out var dto); + var eventObj = DeserializeEvent(payload, out var dto); postprocess?.Invoke(dto); - TryAdvanceLastEventReceivedAt(eventObj.CreatedAt, key); - handler?.Invoke(eventObj, dto); + + if (_isApplyingHistoryEvents) + { + TrackMaxAppliedCreatedAt(eventObj.CreatedAt); + } + else + { + TryAdvanceLastEventReceivedAt(eventObj.CreatedAt, key); + + // The low-level client has no state to read after a silent batch. + // Skip its public callbacks so they match BatchStateUpdate. + handler?.Invoke(eventObj, dto); + } + + // Updates local state even when public callbacks are skipped. internalHandler?.Invoke(dto); } catch (Exception e) { _logs.Exception(e); + + // ApplyHistoryEvents counts failures and needs the throw. Live events stay isolated here. + if (_isApplyingHistoryEvents) + { + throw; + } } }); } - private TEvent DeserializeEvent(string content, out TDto dto) + private TEvent DeserializeEvent(object payload, out TDto dto) where TEvent : ILoadableFrom, new() { + dto = DeserializePayload(payload); + + var response = new TEvent(); + response.LoadFromDto(dto); + + return response; + } + + private TDto DeserializePayload(object payload) + { + if (payload is string content) + { + try + { + return _serializer.Deserialize(content); + } + catch (Exception e) + { + throw new StreamDeserializationException(content, typeof(TDto), e); + } + } + try { - dto = _serializer.Deserialize(content); + var converted = _serializer.TryConvertTo(payload); + if (converted != null) + { + return converted; + } } catch (Exception e) { - throw new StreamDeserializationException(content, typeof(TDto), e); + throw new StreamDeserializationException(PayloadToLog(payload), typeof(TDto), e); } - var response = new TEvent(); - response.LoadFromDto(dto); + // Unknown object shape. Same as the old serialize-then-deserialize path. + try + { + return _serializer.Deserialize(_serializer.Serialize(payload)); + } + catch (Exception e) + { + throw new StreamDeserializationException(PayloadToLog(payload), typeof(TDto), e); + } + } - return response; + private void HandleNewWebsocketMessage(object payload, bool isLiveEvent = false) + { + if (payload is string json) + { + HandleNewWebsocketMessage(json, isLiveEvent); + return; + } + + if (payload == null) + { + HandleNewWebsocketMessage(_serializer.Serialize(payload), isLiveEvent); + return; + } + + WebsocketEventEnvelopePeek peek; + try + { + peek = _serializer.TryConvertTo(payload); + } + catch (Exception) + { + HandleNewWebsocketMessage(_serializer.Serialize(payload), isLiveEvent); + return; + } + + if (peek == null || (peek.Error == null && string.IsNullOrEmpty(peek.Type))) + { + HandleNewWebsocketMessage(_serializer.Serialize(payload), isLiveEvent); + return; + } + + DispatchWebsocketPayload(payload, peek.Error, peek.Type, isLiveEvent); } - private void HandleNewWebsocketMessage(string msg) + private void HandleNewWebsocketMessage(string msg, bool isLiveEvent = false) { const string ErrorKey = "error"; - if (_serializer.TryPeekValue(msg, ErrorKey, out var apiError)) + APIError apiError = null; + if (_serializer.TryPeekValue(msg, ErrorKey, out var peekedError)) + { + apiError = peekedError; + } + + const string TypeKey = "type"; + + string type = null; + if (apiError == null && !_serializer.TryPeekValue(msg, TypeKey, out type)) + { + _logs.Error($"Failed to find `{TypeKey}` in msg: " + msg); + return; + } + + DispatchWebsocketPayload(msg, apiError, type, isLiveEvent); + } + + private void DispatchWebsocketPayload(object payload, APIError apiError, string type, bool isLiveEvent) + { + if (apiError != null) { _errorSb.Length = 0; apiError.AppendFullLog(_errorSb); @@ -1038,15 +1252,15 @@ private void HandleNewWebsocketMessage(string msg) return; } - const string TypeKey = "type"; - - if (!_serializer.TryPeekValue(msg, TypeKey, out var type)) + // Stamp here, not in the health-check handler. That handler runs after other + // callbacks; a slow one can miss the timeout and drop the connection. + // Only live socket events count. A /sync replay does not prove the socket is alive. + if (isLiveEvent && type == WSEventType.HealthCheck) { - _logs.Error($"Failed to find `{TypeKey}` in msg: " + msg); - return; + _lastHealthCheckReceivedTime = _timeService.Time; } - if (EventReceived != null) + if (EventReceived != null && !_isApplyingHistoryEvents) { var time = DateTime.Now.TimeOfDay.ToString(@"hh\:mm\:ss"); EventReceived.Invoke($"{time} - Event received: {type}"); @@ -1054,35 +1268,45 @@ private void HandleNewWebsocketMessage(string msg) if (!_eventKeyToHandler.TryGetValue(type, out var handler)) { - if (TryHandleCustomChannelEvent(msg, type)) + if (TryHandleCustomChannelEvent(payload, type)) { return; } if (_config.LogLevel.IsDebugEnabled()) { - _logs.Warning($"No message handler registered for `{type}`. Message not handled: " + msg); + _logs.Warning($"No message handler registered for `{type}`. Message not handled: " + payload); } return; } - handler(msg); + handler(payload); } - private bool TryHandleCustomChannelEvent(string serializedContent, string eventType) + private bool TryHandleCustomChannelEvent(object payload, string eventType) { - if (!_serializer.TryPeekValue(serializedContent, "cid", out var cid) - || string.IsNullOrEmpty(cid)) + if (!TryGetEventCid(payload, out var cid) || string.IsNullOrEmpty(cid)) { return false; } try { - var dto = _serializer.Deserialize(serializedContent); - TryAdvanceLastEventReceivedAt(dto.CreatedAt, eventType); + var dto = DeserializePayload(payload); + + if (_isApplyingHistoryEvents) + { + TrackMaxAppliedCreatedAt(dto.CreatedAt); + } + else + { + TryAdvanceLastEventReceivedAt(dto.CreatedAt, eventType); + } + // Always raise. Custom events are not stored in channel state, so skipping them + // would lose the payload. JS, Swift, and Android drop custom events from /sync; + // an app that ports between SDKs must not rely on receiving them. var evt = new EventCustom(); ((ILoadableFrom)evt).LoadFromDto(dto); CustomEventReceived?.Invoke(evt); @@ -1096,6 +1320,45 @@ private bool TryHandleCustomChannelEvent(string serializedContent, string eventT } } + private bool TryGetEventCid(object payload, out string cid) + { + if (payload is string json) + { + return _serializer.TryPeekValue(json, "cid", out cid); + } + + try + { + var peek = _serializer.TryConvertTo(payload); + cid = peek?.Cid; + return !string.IsNullOrEmpty(cid); + } + catch (Exception) + { + var serialized = _serializer.Serialize(payload); + return _serializer.TryPeekValue(serialized, "cid", out cid); + } + } + + private static string PayloadToLog(object payload) + => payload as string ?? payload?.ToString() ?? string.Empty; + + /// + /// Enough of a websocket /sync event to dispatch: type, cid, and error. + /// Used so history events can be applied from JObject without a string round-trip. + /// + private class WebsocketEventEnvelopePeek + { + [Newtonsoft.Json.JsonProperty("type")] + public string Type { get; set; } + + [Newtonsoft.Json.JsonProperty("cid")] + public string Cid { get; set; } + + [Newtonsoft.Json.JsonProperty("error")] + public APIError Error { get; set; } + } + private void UpdateHealthCheck() { if (ConnectionState != ConnectionState.Connected) @@ -1133,8 +1396,6 @@ private void PingHealthCheck() private void HandleHealthCheckEvent(EventHealthCheck healthCheckEvent, HealthCheckEventInternalDTO dto) { - _lastHealthCheckReceivedTime = _timeService.Time; - if (ConnectionState == ConnectionState.Connecting) { OnConnectionConfirmed(healthCheckEvent, dto); diff --git a/Assets/Plugins/StreamChat/Core/Responses/StreamStateRecoveredEventArgs.cs b/Assets/Plugins/StreamChat/Core/Responses/StreamStateRecoveredEventArgs.cs new file mode 100644 index 00000000..af528352 --- /dev/null +++ b/Assets/Plugins/StreamChat/Core/Responses/StreamStateRecoveredEventArgs.cs @@ -0,0 +1,44 @@ +using System; +using System.Collections.Generic; +using StreamChat.Core.StatefulModels; + +namespace StreamChat.Core.Responses +{ + /// + /// Data for . + /// + public sealed class StreamStateRecoveredEventArgs + { + public StreamStateRecoveredEventArgs(IReadOnlyList channels, + IReadOnlyList unrecoveredChannelCids) + { + Channels = channels ?? Array.Empty(); + UnrecoveredChannelCids = unrecoveredChannelCids ?? Array.Empty(); + } + + /// + /// Channels that were refreshed and are watched again. + /// Their and other collections are up to date when this event is raised. + /// + /// Recovery adds the newest messages to what was already loaded. + /// If many messages arrived while you were disconnected, there can be a gap in the list. + /// cannot fill that gap, + /// because it loads from the oldest loaded message. + /// + public IReadOnlyList Channels { get; } + + /// + /// Channels that were watched before the disconnect but could not be recovered. + /// The server no longer returns them (deleted, or the user lost access), or the client could not refresh them. + /// Their local state is still stale and they are not watched. + /// + /// Empty when recovery fully succeeds. Use this to close or mark the related UI. + /// + public IReadOnlyList UnrecoveredChannelCids { get; } + + /// + /// True when every watched channel from before the disconnect was recovered. + /// + public bool IsComplete => UnrecoveredChannelCids.Count == 0; + } +} diff --git a/Assets/Plugins/StreamChat/Core/Responses/StreamStateRecoveredEventArgs.cs.meta b/Assets/Plugins/StreamChat/Core/Responses/StreamStateRecoveredEventArgs.cs.meta new file mode 100644 index 00000000..ed989ec5 --- /dev/null +++ b/Assets/Plugins/StreamChat/Core/Responses/StreamStateRecoveredEventArgs.cs.meta @@ -0,0 +1,2 @@ +fileFormatVersion: 2 +guid: 343e17afd03f6814b9154f5e482fbfcb \ No newline at end of file diff --git a/Assets/Plugins/StreamChat/Core/State/StreamStatefulModelBase.cs b/Assets/Plugins/StreamChat/Core/State/StreamStatefulModelBase.cs index bc7e6e67..a46c027a 100644 --- a/Assets/Plugins/StreamChat/Core/State/StreamStatefulModelBase.cs +++ b/Assets/Plugins/StreamChat/Core/State/StreamStatefulModelBase.cs @@ -51,6 +51,17 @@ internal StreamStatefulModelBase(string uniqueId, ICacheRepository Repository { get; } + /// + /// True while a /sync batch is applied with + /// . + /// State still updates. Do not raise per-event callbacks when that change + /// is already visible in state. Use instead. + /// + /// Still raise callbacks that are not stored in state: custom events, + /// channel deleted, and local-user membership or invite notifications. + /// + protected bool IsSilentHistorySync => Client.IsApplyingHistorySync; + protected void LoadAdditionalProperties(Dictionary additionalProperties) { //StreamTodo: investigate if there's a case we don't want to clear here diff --git a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamChannel.cs b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamChannel.cs index a1bba234..6406721c 100644 --- a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamChannel.cs +++ b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamChannel.cs @@ -116,7 +116,7 @@ public bool Hidden get => _hidden; internal set { - if (TrySet(ref _hidden, value)) + if (TrySet(ref _hidden, value) && !IsSilentHistorySync) { VisibilityChanged?.Invoke(this, Hidden); } @@ -148,7 +148,10 @@ internal set } _muted = value; - MuteChanged?.Invoke(this, value); + if (!IsSilentHistorySync) + { + MuteChanged?.Invoke(this, value); + } } } @@ -873,7 +876,10 @@ internal void HandleMessageUpdatedEvent(MessageUpdatedEventInternalDTO dto) } message.TryUpdateFromDto(dto.Message, Cache); - MessageUpdated?.Invoke(this, message); + if (!IsSilentHistorySync) + { + MessageUpdated?.Invoke(this, message); + } } internal void HandleMessageDeletedEvent(MessageDeletedEventInternalDTO dto) @@ -898,7 +904,10 @@ internal void HandleMessageDeletedEvent(MessageDeletedEventInternalDTO dto) message.InternalHandleSoftDelete(); } - MessageDeleted?.Invoke(this, message, isHardDelete); + if (!IsSilentHistorySync) + { + MessageDeleted?.Invoke(this, message, isHardDelete); + } } internal void HandleChannelUpdatedEvent(ChannelUpdatedEventInternalDTO eventDto) @@ -908,7 +917,10 @@ internal void HandleChannelUpdatedEvent(ChannelUpdatedEventInternalDTO eventDto) UpdateChannelFieldsFromDtoOverwrite(eventDto.Channel, Cache); MemberCount = eventDto.ChannelMemberCount; - Updated?.Invoke(this); + if (!IsSilentHistorySync) + { + Updated?.Invoke(this); + } } internal void HandleChannelTruncatedEvent(ChannelTruncatedEventInternalDTO eventDto) @@ -933,8 +945,11 @@ internal void InternalAddMember(StreamChannelMember member) } _members.Add(member); - MemberAdded?.Invoke(this, member); - MembersChanged?.Invoke(this, member, OperationType.Added); + if (!IsSilentHistorySync) + { + MemberAdded?.Invoke(this, member); + MembersChanged?.Invoke(this, member, OperationType.Added); + } } internal void InternalRemoveMember(StreamChannelMember member) @@ -945,8 +960,11 @@ internal void InternalRemoveMember(StreamChannelMember member) } _members.Remove(member); - MemberRemoved?.Invoke(this, member); - MembersChanged?.Invoke(this, member, OperationType.Removed); + if (!IsSilentHistorySync) + { + MemberRemoved?.Invoke(this, member); + MembersChanged?.Invoke(this, member, OperationType.Removed); + } } internal void InternalUpdateMember(StreamChannelMember member) @@ -956,8 +974,11 @@ internal void InternalUpdateMember(StreamChannelMember member) _members.Add(member); } - MemberUpdated?.Invoke(this, member); - MembersChanged?.Invoke(this, member, OperationType.Updated); + if (!IsSilentHistorySync) + { + MemberUpdated?.Invoke(this, member); + MembersChanged?.Invoke(this, member, OperationType.Updated); + } } protected override StreamChannel Self => this; @@ -1037,37 +1058,38 @@ private bool InternalAppendOrUpdateMessage(MessageInternalDTO dto, out StreamMes _messages.Sort(MessageCreatedAtComparer.Instance); } - MessageReceived?.Invoke(this, streamMessage); + if (!IsSilentHistorySync) + { + MessageReceived?.Invoke(this, streamMessage); + } - // Trim after MessageReceived so a message is never removed from cache before it is received. + // Trim after MessageReceived so the callback sees the message first. + // Still trim during a silent batch, or a long /sync can exceed MaxMessages. TrimMessageCacheIfNeeded(); return true; } + /// + /// Query merge appends messages and does not trim. Recovery calls this after merge. + /// + internal void InternalTrimMessageCache() => TrimMessageCacheIfNeeded(); + //StreamTodo: This deleteBeforeCreatedAt date is the date of event, it does not equal the passed TruncatedAt //Therefore the only way to detect partial truncate in the past would be to query the history private void InternalTruncateMessages(DateTimeOffset? deleteBeforeCreatedAt = null, MessageInternalDTO systemMessageDto = null) { - if (deleteBeforeCreatedAt.HasValue) + for (int i = _messages.Count - 1; i >= 0; i--) { - for (int i = _messages.Count - 1; i >= 0; i--) + var msg = _messages[i]; + if (deleteBeforeCreatedAt.HasValue && msg.CreatedAt >= deleteBeforeCreatedAt) { - var msg = _messages[i]; - if (msg.CreatedAt < deleteBeforeCreatedAt) - { - _messages.RemoveAt(i); - Cache.Messages.Remove(msg); - } - } - } - else - { - for (int i = _messages.Count - 1; i >= 0; i--) - { - _messages.RemoveAt(i); - Cache.Messages.Remove(_messages[i]); + continue; } + + _messages.RemoveAt(i); + _pinnedMessages.Remove(msg); + Cache.Messages.Remove(msg); } if (systemMessageDto != null) @@ -1075,7 +1097,10 @@ private void InternalTruncateMessages(DateTimeOffset? deleteBeforeCreatedAt = nu InternalAppendOrUpdateMessage(systemMessageDto, out _); } - Truncated?.Invoke(this); + if (!IsSilentHistorySync) + { + Truncated?.Invoke(this); + } } private void TrimMessageCacheIfNeeded() @@ -1116,8 +1141,9 @@ private void TrimMessageCacheIfNeeded() var handler = MessagesRemovedFromCache; - // Not pooled - this is handed to subscribers, so the SDK does not control its lifetime. - var removed = handler == null ? null : new List(removeCount); + // Not pooled: subscribers keep this list. Skip it during a silent batch + // (callback is not raised, and a long /sync can trim many times). + var removed = (handler == null || IsSilentHistorySync) ? null : new List(removeCount); using (new ListPoolScope(out var tempUntrackCandidates)) using (new HashSetPoolScope(out var tempPinnedMessageIds)) @@ -1142,8 +1168,12 @@ private void TrimMessageCacheIfNeeded() Cache.Messages.RemoveMany(tempUntrackCandidates); } - // Raised after the pooled buffers are returned so subscribers can trim or send safely. - handler?.Invoke(this, removed); + // After pooled lists are returned, so handlers can allocate safely. + // Do not raise during BatchStateUpdate: StateRecovered already rebuilds from Messages. + if (!IsSilentHistorySync) + { + handler?.Invoke(this, removed); + } } // Live messages are still appended while paused, so a channel that is never resumed keeps @@ -1304,7 +1334,10 @@ internal void InternalHandleUserWatchingStartEvent(UserWatchingStartEventInterna { WatcherCount += 1; _watchers.Add(user); - WatcherAdded?.Invoke(this, user); + if (!IsSilentHistorySync) + { + WatcherAdded?.Invoke(this, user); + } } } @@ -1321,7 +1354,11 @@ internal void InternalHandleUserWatchingStop(UserWatchingStopEventInternalDTO ev { var user = Cache.TryCreateOrUpdate(eventDto.User, out var wasCreated); _watchers.RemoveAt(i); - WatcherRemoved?.Invoke(this, user); + if (!IsSilentHistorySync) + { + WatcherRemoved?.Invoke(this, user); + } + return; } } @@ -1338,8 +1375,12 @@ internal void InternalHandleTypingStopped(TypingStopEventInternalDTO eventDto) if (_typingUsers[i].Id == eventDto.User.Id) { _typingUsers.RemoveAt(i); - UserStoppedTyping?.Invoke(this, user); - TypingUsersChanged?.Invoke(this); + if (!IsSilentHistorySync) + { + UserStoppedTyping?.Invoke(this, user); + TypingUsersChanged?.Invoke(this); + } + return; } } @@ -1354,8 +1395,11 @@ internal void InternalHandleTypingStarted(TypingStartEventInternalDTO eventDto) if (!_typingUsers.ContainsNoAlloc(user)) { _typingUsers.Add(user); - UserStartedTyping?.Invoke(this, user); - TypingUsersChanged?.Invoke(this); + if (!IsSilentHistorySync) + { + UserStartedTyping?.Invoke(this, user); + TypingUsersChanged?.Invoke(this); + } } } @@ -1380,17 +1424,33 @@ internal void InternalHandleCustomEvent(CustomEventInternalDTO dto) var customEvent = new StreamCustomEvent(dto.Type, user, dto.CreatedAt, new StreamCustomData(custom, Serializer)); + // Always raise. Custom events are not stored in channel state, so skipping them would lose the payload. CustomEventReceived?.Invoke(this, customEvent); } internal void InternalNotifyReactionReceived(StreamMessage message, StreamReaction reaction) - => ReactionAdded?.Invoke(this, message, reaction); + { + if (!IsSilentHistorySync) + { + ReactionAdded?.Invoke(this, message, reaction); + } + } internal void InternalNotifyReactionUpdated(StreamMessage message, StreamReaction reaction) - => ReactionUpdated?.Invoke(this, message, reaction); + { + if (!IsSilentHistorySync) + { + ReactionUpdated?.Invoke(this, message, reaction); + } + } public void InternalNotifyReactionDeleted(StreamMessage message, StreamReaction reaction) - => ReactionRemoved?.Invoke(this, message, reaction); + { + if (!IsSilentHistorySync) + { + ReactionRemoved?.Invoke(this, message, reaction); + } + } //StreamTodo: implement some timeout for typing users in case we dont' receive, this could be configurable private readonly List _typingUsers = new List(); diff --git a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamMessage.cs b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamMessage.cs index 5e5e4382..32ecdd36 100644 --- a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamMessage.cs +++ b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamMessage.cs @@ -409,7 +409,10 @@ internal void HandleReactionNewEvent(ReactionNewEventInternalDTO eventDto, Strea //StreamTodo: verify if this how we should update the message + what about events for customer to get notified Cache.TryCreateOrUpdate(eventDto.Message); - ReactionAdded?.Invoke(channel, this, reaction); + if (!IsSilentHistorySync) + { + ReactionAdded?.Invoke(channel, this, reaction); + } } internal void HandleReactionUpdatedEvent(ReactionUpdatedEventInternalDTO eventDto, StreamChannel channel, StreamReaction reaction) @@ -422,7 +425,10 @@ internal void HandleReactionUpdatedEvent(ReactionUpdatedEventInternalDTO eventDt eventDto.Message.OwnReactions = null; Cache.TryCreateOrUpdate(eventDto.Message); - ReactionUpdated?.Invoke(channel, this, reaction); + if (!IsSilentHistorySync) + { + ReactionUpdated?.Invoke(channel, this, reaction); + } } internal void HandleReactionDeletedEvent(ReactionDeletedEventInternalDTO eventDto, StreamChannel channel, StreamReaction reaction) @@ -435,7 +441,10 @@ internal void HandleReactionDeletedEvent(ReactionDeletedEventInternalDTO eventDt eventDto.Message.OwnReactions = null; Cache.TryCreateOrUpdate(eventDto.Message); - ReactionRemoved?.Invoke(channel, this, reaction); + if (!IsSilentHistorySync) + { + ReactionRemoved?.Invoke(channel, this, reaction); + } } protected override StreamMessage Self => this; diff --git a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamPoll.cs b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamPoll.cs index df4864ae..de7a8516 100644 --- a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamPoll.cs +++ b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamPoll.cs @@ -55,7 +55,7 @@ private set _isClosed = value; - if (value == true) + if (value && !IsSilentHistorySync) { Closed?.Invoke(this); } @@ -275,7 +275,10 @@ void IUpdateableFrom.UpdateFromDto(Poll LoadAdditionalProperties(dto.Custom); - Updated?.Invoke(this); + if (!IsSilentHistorySync) + { + Updated?.Invoke(this); + } } internal void HandlePollClosedEvent(PollClosedEventInternalDTO dto) @@ -292,7 +295,7 @@ internal void HandlePollVoteCastedEvent(PollVoteCastedEventInternalDTO dto) { this.TryUpdateFromDto(dto.Poll, Cache); - if (dto.PollVote != null) + if (dto.PollVote != null && !IsSilentHistorySync) { var vote = new StreamPollVote().TryLoadFromDto(dto.PollVote, Cache); VoteCasted?.Invoke(this, vote); @@ -303,7 +306,7 @@ internal void HandlePollVoteChangedEvent(PollVoteChangedEventInternalDTO dto) { this.TryUpdateFromDto(dto.Poll, Cache); - if (dto.PollVote != null) + if (dto.PollVote != null && !IsSilentHistorySync) { var vote = new StreamPollVote().TryLoadFromDto(dto.PollVote, Cache); VoteChanged?.Invoke(this, vote); @@ -314,7 +317,7 @@ internal void HandlePollVoteRemovedEvent(PollVoteRemovedEventInternalDTO dto) { this.TryUpdateFromDto(dto.Poll, Cache); - if (dto.PollVote != null) + if (dto.PollVote != null && !IsSilentHistorySync) { var vote = new StreamPollVote().TryLoadFromDto(dto.PollVote, Cache); VoteRemoved?.Invoke(this, vote); diff --git a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamThread.cs b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamThread.cs index f562e38a..2d478614 100644 --- a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamThread.cs +++ b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamThread.cs @@ -195,7 +195,10 @@ void IUpdateableFrom.UpdateFromDto // and stream-chat-js (constructCustomDataObject collects non-reserved top-level fields). LoadAdditionalProperties(dto.AdditionalProperties); - Updated?.Invoke(this); + if (!IsSilentHistorySync) + { + Updated?.Invoke(this); + } } // ChannelStateResponse[Fields]InternalDTO carries threads as ThreadStateInternalDTO @@ -259,7 +262,10 @@ void IUpdateableFrom3.UpdateFromDto( // and stream-chat-js (constructCustomDataObject collects non-reserved top-level fields). LoadAdditionalProperties(dto.AdditionalProperties); - Updated?.Invoke(this); + if (!IsSilentHistorySync) + { + Updated?.Invoke(this); + } } void IUpdateableFrom2.UpdateFromDto( @@ -307,7 +313,10 @@ void IUpdateableFrom2.UpdateFromDto( // and stream-chat-js (constructCustomDataObject collects non-reserved top-level fields). LoadAdditionalProperties(dto.AdditionalProperties); - Updated?.Invoke(this); + if (!IsSilentHistorySync) + { + Updated?.Invoke(this); + } } internal StreamThread(string uniqueId, ICacheRepository repository, @@ -332,7 +341,11 @@ internal void HandleNewReply(IStreamMessage reply) var isInsert = !_latestReplies.Contains(streamReply); if (!isInsert) { - ReplyReceived?.Invoke(this, reply); + if (!IsSilentHistorySync) + { + ReplyReceived?.Invoke(this, reply); + } + return; } @@ -355,7 +368,10 @@ internal void HandleNewReply(IStreamMessage reply) UpsertReplySenderAsParticipant(streamReply); IncrementUnreadForOtherReaders(streamReply); - ReplyReceived?.Invoke(this, reply); + if (!IsSilentHistorySync) + { + ReplyReceived?.Invoke(this, reply); + } } internal void HandleReplyDeleted(string messageId, bool isHardDelete) @@ -427,7 +443,10 @@ internal void HandleMarkReadByUser(string userId, DateTimeOffset createdAt) } } - ReadStateChanged?.Invoke(this); + if (!IsSilentHistorySync) + { + ReadStateChanged?.Invoke(this); + } } // Mirrors Android's Thread.markAsUnreadByUser. notification.mark_unread carries no @@ -452,7 +471,10 @@ internal void HandleMarkUnreadByUser(string userId, DateTimeOffset? lastReadAt) } } - ReadStateChanged?.Invoke(this); + if (!IsSilentHistorySync) + { + ReadStateChanged?.Invoke(this); + } } protected override string InternalUniqueId @@ -521,7 +543,7 @@ private void IncrementUnreadForOtherReaders(StreamMessage reply) } } - if (anyChanged) + if (anyChanged && !IsSilentHistorySync) { ReadStateChanged?.Invoke(this); } diff --git a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamUser.cs b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamUser.cs index edf721ea..8b209dd7 100644 --- a/Assets/Plugins/StreamChat/Core/StatefulModels/StreamUser.cs +++ b/Assets/Plugins/StreamChat/Core/StatefulModels/StreamUser.cs @@ -48,7 +48,7 @@ private set var prev = _online; _online = value; - if (prev != value) + if (prev != value && !IsSilentHistorySync) { PresenceChanged?.Invoke(this, Online, LastActive); } diff --git a/Assets/Plugins/StreamChat/Core/StreamChatClient.cs b/Assets/Plugins/StreamChat/Core/StreamChatClient.cs index cec74b01..eb22abc0 100644 --- a/Assets/Plugins/StreamChat/Core/StreamChatClient.cs +++ b/Assets/Plugins/StreamChat/Core/StreamChatClient.cs @@ -53,8 +53,6 @@ namespace StreamChat.Core /// public delegate void ChannelDeleteHandler(string channelCid, string channelId, ChannelType channelType); - //StreamTodo: Handle restoring state after lost connection - public delegate void ChannelInviteHandler(IStreamChannel channel, IStreamUser invitee); /// @@ -67,6 +65,9 @@ namespace StreamChat.Core /// public delegate void ChannelMemberRemovedHandler(IStreamChannel channel, IStreamChannelMember member); + /// + public delegate void StateRecoveredHandler(StreamStateRecoveredEventArgs eventArgs); + /// public sealed class StreamChatClient : IStreamChatClient { @@ -90,6 +91,8 @@ public sealed class StreamChatClient : IStreamChatClient public event StreamThreadChangeHandler ThreadTracked; public event StreamThreadChangeHandler ThreadUntracked; + public event StateRecoveredHandler StateRecovered; + public const int QueryUsersLimitMaxValue = 30; public const int QueryUsersOffsetMaxValue = 1000; @@ -222,6 +225,10 @@ var ownUserDto public Task DisconnectUserAsync() { TryCancelWaitingForUserConnection(); + + // End the session so the next Connected is a new login, not a reconnect. + _hasConnectedBefore = false; + return InternalLowLevelClient.DisconnectAsync(permanent: true); } @@ -286,28 +293,7 @@ public async Task> QueryChannelsAsync(IEnumerable _.GenerateFilterEntry()).ToDictionary(x => x.Key, x => x.Value), - Limit = limit, - MemberLimit = null, - MessageLimit = null, - Offset = offset, - Presence = true, - - /* - * StreamTodo: Allowing to sort query can potentially lead to mixed sorting in WatchedChannels - * But there seems no other choice because its too limiting to force only a global sorting for channels - * e.g. user may want to show channels in multiple ways with different sorting which would not work with global only sorting - */ - Sort = sort?.ToSortParamRequestList(), - State = true, - Watch = true, - }; - - var channelsResponseDto - = await InternalLowLevelClient.InternalChannelApi.QueryChannelsAsync(requestBodyDto); + var channelsResponseDto = await FetchQueryChannelsResponseAsync(filters, sort, limit, offset); if (channelsResponseDto.Channels == null || channelsResponseDto.Channels.Count == 0) { return Enumerable.Empty(); @@ -332,28 +318,8 @@ public async Task> QueryChannelsAsync(IDictionary x.Key, x => x.Value), - Limit = limit, - MemberLimit = null, - MessageLimit = null, - Offset = offset, - Presence = true, - - /* - * StreamTodo: Allowing to sort query can potentially lead to mixed sorting in WatchedChannels - * But there seems no other choice because its too limiting to force only a global sorting for channels - * e.g. user may want to show channels in multiple ways with different sorting which would not work with global only sorting - */ - Sort = sort?.ToSortParamRequestList(), - State = true, - Watch = true, - }; - - var channelsResponseDto - = await InternalLowLevelClient.InternalChannelApi.QueryChannelsAsync(requestBodyDto); + var channelsResponseDto = await InternalLowLevelClient.InternalChannelApi.QueryChannelsAsync( + CreateQueryChannelsRequest(filters, sort, limit, offset)); if (channelsResponseDto.Channels == null || channelsResponseDto.Channels.Count == 0) { return Enumerable.Empty(); @@ -967,6 +933,38 @@ internal Task RefreshChannelState(string cid) private readonly StreamPollsApi _pollsApi; private readonly List _watchedChannels = new List(); + /// + /// Cids watched when the connection dropped, newest activity first. + /// Recovery uses this because is cleared on disconnect. + /// + private readonly List _recoveryChannelCids = new List(); + + /// + /// Cids /sync reported as inaccessible. Never queried again for this client. + /// + private readonly HashSet _inaccessibleCids = new HashSet(); + + private int _recoveryGeneration; + private bool _hasConnectedBefore; + + /// + /// Cap so reconnect does not run an unbounded number of queries. + /// Same cap as /sync. JS and Android cap at 30. + /// + internal const int MaxRecoveredChannels = StreamChatLowLevelClient.MaxSyncChannelCids; + + /// + /// QueryChannelsAsync allows at most 30 channels per request. + /// + private const int MaxChannelsPerRecoveryQuery = 30; + + // List.Sort is unstable. Ties only happen when LastMessageAt is equal. + private static readonly Comparison ByLastMessageAtDescending = (a, b) + => (b.LastMessageAt ?? DateTimeOffset.MinValue).CompareTo(a.LastMessageAt ?? DateTimeOffset.MinValue); + + /// + internal bool IsApplyingHistorySync => InternalLowLevelClient.IsApplyingHistoryEvents; + private TaskCompletionSource _connectUserTaskSource; private CancellationToken _connectUserCancellationToken; private CancellationTokenSource _connectUserCancellationTokenSource; @@ -1022,10 +1020,8 @@ private void InternalDeleteChannel(StreamChannel channel) ChannelDeleted?.Invoke(channel.Cid, channel.Id, channel.Type); } - // Flip IsWatched=true and add to _watchedChannels. Call from every path that issued - // Watch=true to the server. Channels that land in the cache via non-watching paths - // (search hits, threads with Watch=false, ban-info / mute payloads) stay IsWatched=false. - // Idempotent: a no-op when the channel is already watched. + // Call from every path that sent Watch=true to the server. + // Search hits, threads with Watch=false, and similar cache-only paths stay unwatched. private void MarkChannelWatched(StreamChannel channel) { if (channel == null || channel.IsWatched) @@ -1037,11 +1033,19 @@ private void MarkChannelWatched(StreamChannel channel) _watchedChannels.Add(channel); } - // Counterpart to MarkChannelWatched. Called from StreamChannel.StopWatchingAsync - // after the server confirms the unwatch. Idempotent. + // Called from StreamChannel.StopWatchingAsync after the server confirms. internal void InternalMarkChannelUnwatched(StreamChannel channel) { - if (channel == null || !channel.IsWatched) + if (channel == null) + { + return; + } + + // Also remove it from the reconnect recovery list. If the user stopped watching + // while disconnected, recovery would otherwise start watching this channel again. + _recoveryChannelCids.Remove(channel.Cid); + + if (!channel.IsWatched) { return; } @@ -1050,7 +1054,11 @@ internal void InternalMarkChannelUnwatched(StreamChannel channel) _watchedChannels.Remove(channel); } - private void OnChannelLeftCache(StreamChannel channel) => _watchedChannels.Remove(channel); + private void OnChannelLeftCache(StreamChannel channel) + { + _watchedChannels.Remove(channel); + _recoveryChannelCids.Remove(channel.Cid); + } private void TryCancelWaitingForUserConnection() { @@ -1143,20 +1151,347 @@ private void OnConnected(HealthCheckEventInternalDTO dto) RestoreStateLostDuringDisconnect().LogIfFailed(); } - private Task RestoreStateLostDuringDisconnect() + /// + /// After reconnect the new websocket has no watches. Without this, channels stay + /// in local state but stop receiving events. + /// + /// 1. /sync catch-up (best effort). Must run first: a replayed channel.truncated + /// would wipe messages fetched in step 2. + /// 2. Re-query and re-watch, even if step 1 failed or was skipped. + /// 3. Raise once. + /// + /// A failure in one channel or request must not stop the others. + /// + private async Task RestoreStateLostDuringDisconnect() { - if (!WatchedChannels.Any()) + // First login is not a recovery. Clear any leftover snapshot from a previous session. + if (!_hasConnectedBefore) { - return Task.CompletedTask; + _hasConnectedBefore = true; + _recoveryChannelCids.Clear(); + _inaccessibleCids.Clear(); + return; + } + + if (InternalLowLevelClient.Config.StateRecoveryStrategy == StateRecoveryStrategy.Disabled) + { + return; } - return LowLevelClient.FetchAndProcessEventsSinceLastReceivedEvent(WatchedChannels.Select(c => c.Cid)); + var generation = ++_recoveryGeneration; + + using (new ListPoolScope(out var tempRecoveryChannelCids)) + { + FillRecoveryChannelCids(tempRecoveryChannelCids); + + var refreshedChannels = new List(); + + try + { + if (tempRecoveryChannelCids.Count > 0) + { + await TryCatchUpWithHistoryAsync(tempRecoveryChannelCids, generation); + if (!IsRecoveryGenerationCurrent(generation)) + { + return; + } + + await RehydrateAndRewatchChannelsAsync(tempRecoveryChannelCids, generation, refreshedChannels); + if (!IsRecoveryGenerationCurrent(generation)) + { + return; + } + } + } + catch (Exception e) + { + // Each step already handles its own errors. Still raise StateRecovered so the + // app knows recovery finished and which channels are stale. + _logs.Exception(e); + } + + if (!IsRecoveryGenerationCurrent(generation)) + { + return; + } + + var unrecovered = new List(); + + using (new HashSetPoolScope(out var tempRecovered)) + { + for (var i = 0; i < refreshedChannels.Count; i++) + { + tempRecovered.Add(refreshedChannels[i].Cid); + } + + for (var i = 0; i < tempRecoveryChannelCids.Count; i++) + { + if (!tempRecovered.Contains(tempRecoveryChannelCids[i])) + { + unrecovered.Add(tempRecoveryChannelCids[i]); + } + } + } + + if (unrecovered.Count > 0) + { + _logs.Warning( + $"Reconnect recovery could not restore {unrecovered.Count} channel(s): {string.Join(", ", unrecovered)}. " + + "Their local state is stale and they are no longer watched. See " + + nameof(StreamStateRecoveredEventArgs) + "." + nameof(StreamStateRecoveredEventArgs.UnrecoveredChannelCids)); + } + + StateRecovered?.Invoke(new StreamStateRecoveredEventArgs(refreshedChannels, unrecovered)); + } + } + + /// + /// Copy up to cids from the disconnect snapshot. + /// + private void FillRecoveryChannelCids(List recoveryChannelCids) + { + if (_recoveryChannelCids.Count > MaxRecoveredChannels) + { + _logs.Warning( + $"{_recoveryChannelCids.Count} channels were being watched when the connection dropped, but reconnect " + + $"recovery restores at most {MaxRecoveredChannels}. The {MaxRecoveredChannels} most recently active are " + + "recovered; the rest keep stale state and are not re-watched. Watch fewer channels concurrently, or set " + + nameof(IStreamClientConfig) + "." + nameof(IStreamClientConfig.StateRecoveryStrategy) + " to " + + nameof(StateRecoveryStrategy.Disabled) + " and recover them yourself."); + } + + var count = Math.Min(_recoveryChannelCids.Count, MaxRecoveredChannels); + for (var i = 0; i < count; i++) + { + recoveryChannelCids.Add(_recoveryChannelCids[i]); + } + } + + private async Task TryCatchUpWithHistoryAsync(IReadOnlyList recoveryChannelCids, int generation) + { + try + { + var response = await InternalLowLevelClient.TrySyncHistoryAsync(recoveryChannelCids); + + // null = we never called /sync (no _lastEventReceivedAt yet, or the gap is older than 30 days). + // Channels are still valid; RehydrateAndRewatchChannelsAsync recovers them. + // Contrast inaccessible_cids below: those are gone and must not be queried. + if (response == null) + { + return; + } + + if (!IsRecoveryGenerationCurrent(generation)) + { + return; + } + + if (response.InaccessibleCids != null) + { + // These channels will never come back. Record them so we do not query them again. + // Must run even when Events is empty. + foreach (var cid in response.InaccessibleCids) + { + _inaccessibleCids.Add(cid); + } + } + + if (response.Events == null || response.Events.Count == 0) + { + return; + } + + if (InternalLowLevelClient.Config.StateRecoveryStrategy == StateRecoveryStrategy.BatchStateUpdate) + { + InternalLowLevelClient.ApplyHistoryEvents(response.Events); + } + else + { + InternalLowLevelClient.ReplayHistoryEvents(response.Events); + } + } + catch (StreamApiException ex) when (ex.IsInputError()) + { + // Too many events to sync. The re-query below recovers state without those events. + _logs.Warning("The /sync catch-up was refused because too many events accumulated during the outage. " + + "Recovering channel state with a re-query instead. " + ex.Message); + } + catch (Exception ex) + { + _logs.Warning("The /sync catch-up failed. Recovering channel state with a re-query instead. " + + ex.Message); + } + } + + /// + /// Re-hydrate and re-watch in one request per cids. + /// Unlike Android, we do not re-watch omitted cids one by one. This query is by cid, + /// so a missing cid is deleted or no longer readable. A get-or-create watch would + /// recreate a deleted channel. Those cids go to + /// . + /// + private async Task RehydrateAndRewatchChannelsAsync(IReadOnlyList recoveryChannelCids, int generation, + List refreshed) + { + var sort = ChannelSort.OrderByDescending(ChannelSortFieldName.LastMessageAt); + + for (var i = 0; i < recoveryChannelCids.Count; i += MaxChannelsPerRecoveryQuery) + { + if (!IsRecoveryGenerationCurrent(generation)) + { + return; + } + + using (new ListPoolScope(out var tempChunk)) + { + var chunkEnd = Math.Min(i + MaxChannelsPerRecoveryQuery, recoveryChannelCids.Count); + for (var j = i; j < chunkEnd; j++) + { + if (!_inaccessibleCids.Contains(recoveryChannelCids[j])) + { + tempChunk.Add(recoveryChannelCids[j]); + } + } + + if (tempChunk.Count == 0) + { + continue; + } + + var filters = new IFieldFilterRule[] + { + ChannelFilter.Cid.In(tempChunk), + }; + + QueryChannelsResponseInternalDTO response; + try + { + // Fetch only. Apply after the generation check so a late response from an old + // connection cannot overwrite newer members, read, or pinned messages. + response = await FetchQueryChannelsResponseAsync(filters, sort, tempChunk.Count); + } + catch (Exception e) + { + // Do not skip the rest. There is no later retry on this connection. + _logs.Warning($"Recovery query failed for {tempChunk.Count} channel(s). Continuing with the rest. " + + e.Message); + continue; + } + + if (!IsRecoveryGenerationCurrent(generation)) + { + return; + } + + if (response?.Channels == null) + { + continue; + } + + foreach (var channelDto in response.Channels) + { + var channel = _cache.TryCreateOrUpdate(channelDto); + if (channel == null) + { + continue; + } + + MarkChannelWatched(channel); + // Query merge does not trim, so Messages can exceed MessageCacheWindow. + channel.InternalTrimMessageCache(); + refreshed.Add(channel); + } + } + } + } + + /// + /// Same request as public QueryChannelsAsync, but does not update cache or watches. + /// A late response from an old connection must not overwrite newer state. + /// + private Task FetchQueryChannelsResponseAsync( + IEnumerable filters, ChannelSortObject sort, int limit, int offset = 0) + { + var filterConditions = filters?.Select(_ => _.GenerateFilterEntry()).ToDictionary(x => x.Key, x => x.Value); + return InternalLowLevelClient.InternalChannelApi.QueryChannelsAsync( + CreateQueryChannelsRequest(filterConditions, sort, limit, offset)); + } + + //StreamTodo: Perhaps MessageLimit and MemberLimit should be configurable + /* + * StreamTodo: Allowing to sort query can potentially lead to mixed sorting in WatchedChannels + * But there seems no other choice because its too limiting to force only a global sorting for channels + * e.g. user may want to show channels in multiple ways with different sorting which would not work with global only sorting + */ + private static QueryChannelsRequestInternalDTO CreateQueryChannelsRequest( + IDictionary filterConditions, ChannelSortObject sort, int limit, int offset) + => new QueryChannelsRequestInternalDTO + { + FilterConditions = filterConditions as Dictionary + ?? filterConditions?.ToDictionary(x => x.Key, x => x.Value), + Limit = limit, + MemberLimit = null, + MessageLimit = null, + Offset = offset, + Presence = true, + Sort = sort?.ToSortParamRequestList(), + State = true, + Watch = true, + }; + + private bool IsRecoveryGenerationCurrent(int generation) => generation == _recoveryGeneration; + + /// + /// Save watched cids, then mark them unwatched. The server already dropped the watches. + /// + private void SnapshotRecoverySetAndClearWatches() + { + // A failed reconnect also hits Disconnected with an empty watch list. + // Do not overwrite the snapshot, or the next success recovers nothing. + if (_watchedChannels.Count == 0) + { + return; + } + + _recoveryChannelCids.Clear(); + + using (new ListPoolScope(out var tempOrdered)) + { + tempOrdered.AddRange(_watchedChannels); + tempOrdered.Sort(ByLastMessageAtDescending); + + for (var i = 0; i < tempOrdered.Count; i++) + { + _recoveryChannelCids.Add(tempOrdered[i].Cid); + } + } + + for (var i = 0; i < _watchedChannels.Count; i++) + { + ((StreamChannel)_watchedChannels[i]).IsWatched = false; + } + + _watchedChannels.Clear(); } private void OnDisconnected() => Disconnected?.Invoke(); private void OnConnectionStateChanged(ConnectionState previous, ConnectionState current) - => ConnectionStateChanged?.Invoke(previous, current); + { + if (current == ConnectionState.Disconnected) + { + // Ignore in-flight recovery. A late query can replace members, read state, + // and pinned messages with older data. + _recoveryGeneration++; + + if (InternalLowLevelClient.Config.StateRecoveryStrategy != StateRecoveryStrategy.Disabled) + { + SnapshotRecoverySetAndClearWatches(); + } + } + + ConnectionStateChanged?.Invoke(previous, current); + } private void OnMessageDeleted(MessageDeletedEventInternalDTO eventMessageDeleted) { @@ -1606,16 +1941,29 @@ var reaction } } + // Watcher lists are live presence. Replaying them would show people who left during the outage. + // The recovery query returns the current watcher set. private void OnUserWatchingStop(UserWatchingStopEventInternalDTO eventDto) { + if (IsApplyingHistorySync) + { + return; + } + if (_cache.Channels.TryGet(eventDto.Cid, out var streamChannel)) { streamChannel.InternalHandleUserWatchingStop(eventDto); } } + /// private void OnUserWatchingStart(UserWatchingStartEventInternalDTO eventDto) { + if (IsApplyingHistorySync) + { + return; + } + if (_cache.Channels.TryGet(eventDto.Cid, out var streamChannel)) { streamChannel.InternalHandleUserWatchingStartEvent(eventDto); @@ -1655,14 +2003,27 @@ private void OnUserPresenceChanged(UserPresenceChangedEventInternalDTO eventDto) private void OnTypingStopped(TypingStopEventInternalDTO eventDto) { + // Typing is live presence. A typing.start without its matching typing.stop would leave + // a user typing forever. Skip the whole event, including state. + if (IsApplyingHistorySync) + { + return; + } + if (_cache.Channels.TryGet(eventDto.Cid, out var streamChannel)) { streamChannel.InternalHandleTypingStopped(eventDto); } } + /// private void OnTypingStarted(TypingStartEventInternalDTO eventDto) { + if (IsApplyingHistorySync) + { + return; + } + if (_cache.Channels.TryGet(eventDto.Cid, out var streamChannel)) { streamChannel.InternalHandleTypingStarted(eventDto); diff --git a/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs new file mode 100644 index 00000000..5ff2bac4 --- /dev/null +++ b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs @@ -0,0 +1,818 @@ +#if STREAM_TESTS_ENABLED +using System; +using System.Collections; +using System.Collections.Generic; +using System.Linq; +using System.Reflection; +using System.Threading.Tasks; +using UnityEngine.TestTools; +using NSubstitute; +using NUnit.Framework; +using StreamChat.Core; +using StreamChat.Core.Configs; +using StreamChat.Core.InternalDTO.Models; +using StreamChat.Core.InternalDTO.Responses; +using StreamChat.Core.LowLevelClient; +using StreamChat.Core.Responses; +using StreamChat.Core.State.Caches; +using StreamChat.Core.StatefulModels; +using StreamChat.Libs.AppInfo; +using StreamChat.Libs.Auth; +using StreamChat.Libs.ChatInstanceRunner; +using StreamChat.Libs.Http; +using StreamChat.Libs.Logs; +using StreamChat.Libs.NetworkMonitors; +using StreamChat.Libs.Serialization; +using StreamChat.Libs.Time; +using StreamChat.Libs.Websockets; + +namespace StreamChat.Tests.StateSync.Unit +{ + /// + /// Unit tests for reconnect state recovery on , driven entirely + /// through mocked transports so connection state transitions can be sequenced exactly. + /// + internal class StateRecoveryClientTests + { + [SetUp] + public void Up() + { + _authCredentials = new AuthCredentials("api123", "user123", "token123"); + _mockWebsocketClient = Substitute.For(); + _mockHttpClient = Substitute.For(); + _mockTimeService = Substitute.For(); + _mockNetworkMonitor = Substitute.For(); + _mockApplicationInfo = Substitute.For(); + _mockLogs = Substitute.For(); + _config = new StreamClientConfig(); + + _mockWebsocketClient.ConnectAsync(Arg.Any()).Returns(Task.CompletedTask); + _mockWebsocketClient.DisconnectAsync(Arg.Any(), + Arg.Any()).Returns(Task.CompletedTask); + + _mockWebsocketClient.TryDequeueMessage(out Arg.Any()).Returns(arg => + { + if (_pendingWebsocketMessages.Count == 0) + { + return false; + } + + arg[0] = _pendingWebsocketMessages.Dequeue(); + return true; + }); + + RespondWith(SyncEndpoint, "{\"events\":[]}"); + RespondWith(QueryChannelsEndpoint, "{\"channels\":[]}"); + + _client = (StreamChatClient)StreamChatClient.CreateClientWithCustomDependencies(_mockWebsocketClient, + _mockHttpClient, new NewtonsoftJsonSerializer(), _mockTimeService, _mockNetworkMonitor, + _mockApplicationInfo, _mockLogs, _config); + } + + [TearDown] + public void TearDown() + { + _client.Dispose(); + _client = null; + _pendingWebsocketMessages.Clear(); + _recoveredEvents.Clear(); + } + + [Test] + public void when_first_connect_expect_no_recovery_and_no_state_recovered_event() + { + Connect(); + + Assert.AreEqual(0, _recoveredEvents.Count, + "A fresh login is not a recovery - firing StateRecovered here would make it useless as a signal."); + AssertQueryChannelsCallCount(0); + } + + [Test] + public void when_reconnected_expect_channels_requeried_even_though_sync_was_skipped() + { + Connect(); + WatchChannel("messaging:a"); + WatchChannel("messaging:b"); + + DropConnection(); + Reconnect(); + + // /sync is skipped here (no last_sync_at). Recovery must still re-query so the + // new socket watches the same channels again. + AssertQueryChannelsCallCount(1); + Assert.AreEqual(1, _recoveredEvents.Count); + _mockHttpClient.Received().SendHttpRequestAsync( + Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith(QueryChannelsEndpoint)), + Arg.Is(body => RequestHasJsonBool(body, "watch", true) + && RequestHasJsonBool(body, "state", true))); + } + + [Test] + public void when_connection_drops_expect_watches_released() + { + Connect(); + var channel = WatchChannel("messaging:a"); + + DropConnection(); + + Assert.IsFalse(channel.IsWatched); + Assert.AreEqual(0, _client.WatchedChannels.Count); + } + + [Test] + public void when_reconnect_attempt_fails_expect_channels_still_recovered() + { + Connect(); + WatchChannel("messaging:a"); + WatchChannel("messaging:b"); + + DropConnection(); + FailReconnectAttempt(); + FailReconnectAttempt(); + Reconnect(); + + // Failed Connecting -> Disconnected must not forget the channels that were + // watched before the outage. The attempt that succeeds still recovers them. + AssertQueryChannelsCallCount(1); + _mockHttpClient.Received().SendHttpRequestAsync( + Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith(QueryChannelsEndpoint)), + Arg.Is(body => RequestBodyContains(body, "messaging:a") + && RequestBodyContains(body, "messaging:b"))); + } + + [Test] + public void when_channels_cannot_be_recovered_expect_them_reported_as_unrecovered() + { + Connect(); + WatchChannel("messaging:a"); + + DropConnection(); + Reconnect(); + + Assert.AreEqual(1, _recoveredEvents.Count); + Assert.AreEqual(0, _recoveredEvents[0].Channels.Count); + Assert.AreEqual(new[] { "messaging:a" }, _recoveredEvents[0].UnrecoveredChannelCids.ToArray()); + Assert.IsFalse(_recoveredEvents[0].IsComplete); + } + + [Test] + public void when_recovery_query_fails_expect_remaining_chunks_still_queried() + { + Connect(); + for (var i = 0; i < 40; i++) + { + WatchChannel($"messaging:channel-{i:D2}"); + } + + var callCount = 0; + _mockHttpClient + .SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith(QueryChannelsEndpoint)), Arg.Any()) + .Returns(_ => + { + callCount++; + return callCount == 1 + ? new HttpResponse(false, 429, "{\"code\":9,\"message\":\"rate limited\"}", null, null) + : new HttpResponse(true, 200, "{\"channels\":[]}", null, null); + }); + + DropConnection(); + Reconnect(); + + Assert.AreEqual(2, callCount); + Assert.AreEqual(1, _recoveredEvents.Count); + Assert.AreEqual(40, _recoveredEvents[0].UnrecoveredChannelCids.Count); + Assert.IsFalse(_recoveredEvents[0].IsComplete); + } + + [Test] + public void when_recovery_set_exceeds_cap_expect_only_capped_channels_queried() + { + Connect(); + for (var i = 0; i < StreamChatClient.MaxRecoveredChannels + 25; i++) + { + WatchChannel($"messaging:channel-{i:D3}"); + } + + DropConnection(); + Reconnect(); + + AssertQueryChannelsCallCount(4); + } + + [Test] + public void when_strategy_is_disabled_expect_no_recovery_and_watches_left_untouched() + { + _config.StateRecoveryStrategy = StateRecoveryStrategy.Disabled; + + Connect(); + var channel = WatchChannel("messaging:a"); + + DropConnection(); + Reconnect(); + + AssertQueryChannelsCallCount(0); + Assert.AreEqual(0, _recoveredEvents.Count); + Assert.IsTrue(channel.IsWatched); + Assert.AreEqual(1, _client.WatchedChannels.Count); + } + + [Test] + public void when_user_disconnects_and_connects_again_expect_no_recovery_of_previous_session() + { + Connect(); + WatchChannel("messaging:a"); + + _client.DisconnectUserAsync().GetAwaiter().GetResult(); + DropConnection(); + + Connect(); + + AssertQueryChannelsCallCount(0); + Assert.AreEqual(0, _recoveredEvents.Count); + } + + [Test] + public void when_channel_unwatched_while_disconnected_expect_it_not_recovered() + { + Connect(); + var channel = WatchChannel("messaging:a"); + WatchChannel("messaging:b"); + + DropConnection(); + InvokeMarkChannelUnwatched(channel); + Reconnect(); + + _mockHttpClient.Received(1).SendHttpRequestAsync( + Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith(QueryChannelsEndpoint)), + Arg.Is(body => RequestBodyContains(body, "messaging:b") + && !RequestBodyContains(body, "messaging:a"))); + } + + [Test] + public void when_sync_returns_empty_events_with_inaccessible_cids_expect_those_cids_not_requeried() + { + Connect(); + WatchChannel("messaging:gone"); + var ok = WatchChannel("messaging:ok"); + + var now = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero); + _mockTimeService.Now.Returns(now); + + DropConnection(); + SetDisconnectionLastEventReceivedAt(_client.InternalLowLevelClient, now.AddHours(-1)); + + RespondWith(SyncEndpoint, "{\"events\":[],\"inaccessible_cids\":[\"messaging:gone\"]}"); + RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:ok", "ok")); + + Reconnect(); + + _mockHttpClient.Received(1).SendHttpRequestAsync( + Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith(QueryChannelsEndpoint)), + Arg.Is(body => RequestBodyContains(body, "messaging:ok") + && !RequestBodyContains(body, "messaging:gone"))); + + Assert.AreEqual(1, _recoveredEvents.Count); + Assert.AreSame(ok, _recoveredEvents[0].Channels.Single()); + Assert.AreEqual(new[] { "messaging:gone" }, _recoveredEvents[0].UnrecoveredChannelCids.ToArray()); + Assert.IsFalse(_recoveredEvents[0].IsComplete); + Assert.IsTrue(ok.IsWatched); + } + + [Test] + public void when_recovery_query_returns_channel_expect_watch_restored_and_state_recovered() + { + Connect(); + var channel = WatchChannel("messaging:a"); + + RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "restored")); + + DropConnection(); + Reconnect(); + + Assert.IsTrue(channel.IsWatched); + Assert.AreEqual(1, _client.WatchedChannels.Count); + Assert.AreSame(channel, _client.WatchedChannels.Single()); + Assert.AreEqual("restored", channel.Name); + Assert.AreEqual(1, _recoveredEvents.Count); + Assert.AreSame(channel, _recoveredEvents[0].Channels.Single()); + Assert.IsTrue(_recoveredEvents[0].IsComplete); + } + + [UnityTest] + public IEnumerator when_stale_recovery_query_completes_after_newer_recovery_expect_stale_result_not_applied() + { + Connect(); + var channel = WatchChannel("messaging:a"); + + var staleQuery = new TaskCompletionSource(); + var currentQuery = new TaskCompletionSource(); + HoldQueryChannelsResponses(staleQuery.Task, currentQuery.Task); + + DropConnection(); + Reconnect(); + + Assert.AreEqual(0, _recoveredEvents.Count, + "The first recovery must still be waiting on its query."); + Assert.IsFalse(channel.IsWatched); + Assert.AreEqual(0, _client.WatchedChannels.Count); + + DropConnection(); + Reconnect(); + + Assert.AreEqual(0, _recoveredEvents.Count); + + currentQuery.SetResult(QueryChannelsHttpResponse("messaging:a", "fresh-B")); + yield return WaitUntil(() => channel.Name == "fresh-B", + "Current-generation recovery did not apply the fresh query payload."); + + Assert.IsTrue(channel.IsWatched); + Assert.AreEqual(1, _client.WatchedChannels.Count); + Assert.AreSame(channel, _client.WatchedChannels.Single()); + Assert.AreEqual(1, _recoveredEvents.Count); + Assert.AreSame(channel, _recoveredEvents[0].Channels.Single()); + Assert.IsTrue(_recoveredEvents[0].IsComplete); + + staleQuery.SetResult(QueryChannelsHttpResponse("messaging:a", "stale-A")); + for (var i = 0; i < 5; i++) + { + Update(); + yield return null; + } + + Assert.AreEqual("fresh-B", channel.Name, + "A late query from a superseded recovery must not replace channel state."); + Assert.IsTrue(channel.IsWatched); + Assert.AreEqual(1, _client.WatchedChannels.Count); + Assert.AreEqual(1, _recoveredEvents.Count, + "The stale recovery must not raise StateRecovered after a newer one already did."); + } + + [Test] + public void when_silent_history_batch_trims_expect_messages_removed_from_cache_not_raised() + { + Connect(); + var channel = WatchChannel("messaging:a"); + channel.OverrideMessageCacheWindow(SmallWindow); + + var removedCount = 0; + channel.MessagesRemovedFromCache += (_, __) => removedCount++; + + _client.InternalLowLevelClient.ApplyHistoryEvents(MessageNewEvents("messaging:a", count: 7)); + + Assert.AreEqual(0, removedCount); + Assert.AreEqual(SmallWindow.MaxMessages - SmallWindow.DiscardBatchSize, channel.Messages.Count); + } + + [Test] + public void when_history_replay_trims_expect_messages_removed_from_cache_raised() + { + Connect(); + var channel = WatchChannel("messaging:a"); + channel.OverrideMessageCacheWindow(SmallWindow); + + var removedCount = 0; + channel.MessagesRemovedFromCache += (_, __) => removedCount++; + + _client.InternalLowLevelClient.ReplayHistoryEvents(MessageNewEvents("messaging:a", count: 7)); + + Assert.AreEqual(1, removedCount); + Assert.AreEqual(SmallWindow.MaxMessages - SmallWindow.DiscardBatchSize, channel.Messages.Count); + } + + [Test] + public void when_silent_history_batch_adds_thread_reply_expect_reply_received_not_raised() + { + Connect(); + WatchChannel("messaging:a"); + var thread = TrackThread("messaging:a", "parent-1"); + + var replyCount = 0; + thread.ReplyReceived += (_, __) => replyCount++; + + _client.InternalLowLevelClient.ApplyHistoryEvents(new[] + { + ThreadReplyJson("messaging:a", "parent-1", "reply-1"), + }); + + Assert.AreEqual(0, replyCount); + Assert.AreEqual(1, thread.LatestReplies.Count); + Assert.AreEqual("reply-1", thread.LatestReplies[0].Id); + } + + [Test] + public void when_history_replay_adds_thread_reply_expect_reply_received_raised() + { + Connect(); + WatchChannel("messaging:a"); + var thread = TrackThread("messaging:a", "parent-1"); + + var replyCount = 0; + thread.ReplyReceived += (_, __) => replyCount++; + + _client.InternalLowLevelClient.ReplayHistoryEvents(new[] + { + ThreadReplyJson("messaging:a", "parent-1", "reply-1"), + }); + + Assert.AreEqual(1, replyCount); + Assert.AreEqual(1, thread.LatestReplies.Count); + Assert.AreEqual("reply-1", thread.LatestReplies[0].Id); + } + + [Test] + public void when_silent_history_batch_changes_presence_expect_presence_changed_not_raised() + { + Connect(); + var user = TrackUser("other-user", online: false); + + var presenceCount = 0; + user.PresenceChanged += (_, __, ___) => presenceCount++; + + _client.InternalLowLevelClient.ApplyHistoryEvents(new[] + { + PresenceChangedJson("other-user", online: true), + }); + + Assert.AreEqual(0, presenceCount); + Assert.IsTrue(user.Online); + } + + [Test] + public void when_history_replay_changes_presence_expect_presence_changed_raised() + { + Connect(); + var user = TrackUser("other-user", online: false); + + var presenceCount = 0; + user.PresenceChanged += (_, __, ___) => presenceCount++; + + _client.InternalLowLevelClient.ReplayHistoryEvents(new[] + { + PresenceChangedJson("other-user", online: true), + }); + + Assert.AreEqual(1, presenceCount); + Assert.IsTrue(user.Online); + } + + [Test] + public void when_silent_history_batch_closes_poll_expect_closed_not_raised() + { + Connect(); + WatchChannel("messaging:a"); + var poll = TrackPoll("poll-1"); + + var closedCount = 0; + var updatedCount = 0; + poll.Closed += _ => closedCount++; + poll.Updated += _ => updatedCount++; + + _client.InternalLowLevelClient.ApplyHistoryEvents(new[] + { + PollClosedJson("messaging:a", "poll-1"), + }); + + Assert.AreEqual(0, closedCount); + Assert.AreEqual(0, updatedCount); + Assert.IsTrue(poll.IsClosed); + } + + [Test] + public void when_history_replay_closes_poll_expect_closed_raised() + { + Connect(); + WatchChannel("messaging:a"); + var poll = TrackPoll("poll-1"); + + var closedCount = 0; + poll.Closed += _ => closedCount++; + + _client.InternalLowLevelClient.ReplayHistoryEvents(new[] + { + PollClosedJson("messaging:a", "poll-1"), + }); + + Assert.AreEqual(1, closedCount); + Assert.IsTrue(poll.IsClosed); + } + + [Test] + public void when_silent_recovery_syncs_message_new_expect_channel_message_received_not_raised() + { + _config.StateRecoveryStrategy = StateRecoveryStrategy.BatchStateUpdate; + + Connect(); + var channel = WatchChannel("messaging:a"); + RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "a")); + + var receivedCount = 0; + channel.MessageReceived += (_, __) => receivedCount++; + + ReconnectWithSync(SyncEventsJson(MessageNewJson("messaging:a", "msg-offline", + new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero)))); + + Assert.AreEqual(0, receivedCount); + Assert.AreEqual(1, channel.Messages.Count); + Assert.AreEqual("msg-offline", channel.Messages[0].Id); + Assert.AreEqual(1, _recoveredEvents.Count); + } + + [Test] + public void when_replay_recovery_syncs_message_new_expect_channel_message_received_raised() + { + Connect(); + var channel = WatchChannel("messaging:a"); + RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "a")); + + var receivedCount = 0; + channel.MessageReceived += (_, __) => receivedCount++; + + ReconnectWithSync(SyncEventsJson(MessageNewJson("messaging:a", "msg-offline", + new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero)))); + + Assert.AreEqual(1, receivedCount); + Assert.AreEqual(1, channel.Messages.Count); + Assert.AreEqual("msg-offline", channel.Messages[0].Id); + Assert.AreEqual(1, _recoveredEvents.Count); + } + + [Test] + public void when_silent_recovery_syncs_custom_event_expect_channel_custom_event_received() + { + _config.StateRecoveryStrategy = StateRecoveryStrategy.BatchStateUpdate; + + Connect(); + var channel = WatchChannel("messaging:a"); + RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "a")); + + string receivedType = null; + channel.CustomEventReceived += (_, evt) => receivedType = evt.Type; + + ReconnectWithSync(SyncEventsJson(CustomEventJson("messaging:a", "game.state"))); + + Assert.AreEqual("game.state", receivedType); + } + + private const string SyncEndpoint = "/sync"; + private const string QueryChannelsEndpoint = "/channels"; + + // Same window as MessageCacheWindowTests: 7 messages exceed MaxMessages and trim down to 3. + private static readonly MessageCacheWindow SmallWindow = new MessageCacheWindow(6, 3); + + private static IEnumerable MessageNewEvents(string cid, int count) + { + var start = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero); + var events = new List(count); + for (var i = 0; i < count; i++) + { + events.Add(MessageNewJson(cid, $"msg-{i}", start.AddSeconds(i))); + } + + return events; + } + + private static string MessageNewJson(string cid, string messageId, DateTimeOffset createdAt) + => $"{{\"type\":\"message.new\",\"cid\":\"{cid}\",\"created_at\":\"{createdAt:O}\"," + + $"\"message\":{{\"id\":\"{messageId}\",\"text\":\"hi\",\"created_at\":\"{createdAt:O}\"," + + $"\"updated_at\":\"{createdAt:O}\",\"user\":{{\"id\":\"user-1\"}}}}}}"; + + private static string ThreadReplyJson(string cid, string parentId, string replyId) + { + var createdAt = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero); + return $"{{\"type\":\"message.new\",\"cid\":\"{cid}\",\"created_at\":\"{createdAt:O}\"," + + $"\"message\":{{\"id\":\"{replyId}\",\"parent_id\":\"{parentId}\",\"text\":\"reply\"," + + $"\"created_at\":\"{createdAt:O}\",\"updated_at\":\"{createdAt:O}\"," + + $"\"user\":{{\"id\":\"user-1\"}}}}}}"; + } + + private static string PresenceChangedJson(string userId, bool online) + { + var createdAt = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero); + var onlineJson = online ? "true" : "false"; + return $"{{\"type\":\"user.presence.changed\",\"created_at\":\"{createdAt:O}\"," + + $"\"user\":{{\"id\":\"{userId}\",\"online\":{onlineJson}}}}}"; + } + + private static string PollClosedJson(string cid, string pollId) + { + var createdAt = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero); + return $"{{\"type\":\"poll.closed\",\"cid\":\"{cid}\",\"created_at\":\"{createdAt:O}\"," + + $"\"poll\":{{\"id\":\"{pollId}\",\"name\":\"q\",\"is_closed\":true,\"vote_count\":0," + + $"\"voting_visibility\":\"public\"}}}}"; + } + + private static string CustomEventJson(string cid, string type) + { + var createdAt = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero); + return $"{{\"type\":\"{type}\",\"cid\":\"{cid}\",\"created_at\":\"{createdAt:O}\"," + + "\"user\":{\"id\":\"user-1\"}}"; + } + + private static string SyncEventsJson(params string[] events) + => "{\"events\":[" + string.Join(",", events) + "]}"; + + private void ReconnectWithSync(string syncJson) + { + var now = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero); + _mockTimeService.Now.Returns(now); + + DropConnection(); + SetDisconnectionLastEventReceivedAt(_client.InternalLowLevelClient, now.AddHours(-1)); + RespondWith(SyncEndpoint, syncJson); + Reconnect(); + } + + private IStreamThread TrackThread(string cid, string parentMessageId) + => _client.InternalCache.TryCreateOrUpdate(new ThreadStateInternalDTO + { + ParentMessageId = parentMessageId, + ChannelCid = cid, + ReplyCount = 0, + ParentMessage = new MessageInternalDTO + { + Id = parentMessageId, + Text = "parent", + User = new UserObjectInternalDTO { Id = "user-1" }, + }, + }); + + private IStreamUser TrackUser(string userId, bool online) + => _client.InternalCache.TryCreateOrUpdate(new UserObjectInternalDTO + { + Id = userId, + Online = online, + }); + + private IStreamPoll TrackPoll(string pollId) + => _client.InternalCache.TryCreateOrUpdate(new PollResponseDataInternalDTO + { + Id = pollId, + Name = "q", + IsClosed = false, + VoteCount = 0, + }); + + private void RespondWith(string endpointSuffix, string json) + { + _mockHttpClient + .SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith(endpointSuffix)), Arg.Any()) + .Returns(new HttpResponse(true, 200, json, null, null)); + } + + private void HoldQueryChannelsResponses(params Task[] responses) + { + var queue = new Queue>(responses); + _mockHttpClient + .SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith(QueryChannelsEndpoint)), Arg.Any()) + .Returns(_ => queue.Dequeue()); + } + + private static HttpResponse QueryChannelsHttpResponse(string cid, string name) + => new HttpResponse(true, 200, QueryChannelsJson(cid, name), null, null); + + private static string QueryChannelsJson(string cid, string name) + { + var separatorIndex = cid.IndexOf(':'); + var type = cid.Substring(0, separatorIndex); + var id = cid.Substring(separatorIndex + 1); + return "{\"channels\":[{\"channel\":{" + + $"\"cid\":\"{cid}\",\"id\":\"{id}\",\"type\":\"{type}\",\"name\":\"{name}\"" + + "},\"messages\":[],\"members\":[]}]}"; + } + + private IEnumerator WaitUntil(Func condition, string message, int maxFrames = 30) + { + var frames = 0; + while (!condition() && frames < maxFrames) + { + Update(); + yield return null; + frames++; + } + + Assert.IsTrue(condition(), message); + } + + private void AssertQueryChannelsCallCount(int expected) + { + _mockHttpClient.Received(expected).SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith(QueryChannelsEndpoint)), Arg.Any()); + } + + private void Connect() + { + _client.StateRecovered -= OnStateRecovered; + _client.StateRecovered += OnStateRecovered; + + var connectTask = _client.ConnectUserAsync(_authCredentials); + _pendingWebsocketMessages.Enqueue(HealthCheckJson); + Update(); + + Assert.IsTrue(connectTask.IsCompleted, "Expected the mocked health check to complete the connect."); + Assert.AreEqual(ConnectionState.Connected, _client.ConnectionState); + } + + private void Reconnect() + { + _client.InternalLowLevelClient.Connect(); + _pendingWebsocketMessages.Enqueue(HealthCheckJson); + Update(); + + Assert.AreEqual(ConnectionState.Connected, _client.ConnectionState); + } + + private void DropConnection() + { + _mockWebsocketClient.Disconnected += Raise.Event(); + Update(); + + Assert.AreEqual(ConnectionState.Disconnected, _client.ConnectionState); + } + + private void FailReconnectAttempt() + { + _client.InternalLowLevelClient.Connect(); + Assert.AreEqual(ConnectionState.Connecting, _client.ConnectionState); + + _mockWebsocketClient.ConnectionFailed += Raise.Event(); + Update(); + + Assert.AreEqual(ConnectionState.Disconnected, _client.ConnectionState); + } + + private void Update() => ((IStreamChatClientEventsListener)_client).Update(); + + private void OnStateRecovered(StreamStateRecoveredEventArgs args) => _recoveredEvents.Add(args); + + private IStreamChannel WatchChannel(string cid) + { + var separatorIndex = cid.IndexOf(':'); + var channel = _client.InternalCache.TryCreateOrUpdate(new ChannelResponseInternalDTO + { + Cid = cid, + Type = cid.Substring(0, separatorIndex), + Id = cid.Substring(separatorIndex + 1), + }); + + InvokePrivate("MarkChannelWatched", channel); + return channel; + } + + private void InvokeMarkChannelUnwatched(IStreamChannel channel) + => InvokePrivate("InternalMarkChannelUnwatched", channel); + + private void InvokePrivate(string methodName, object argument) + { + var method = typeof(StreamChatClient).GetMethod(methodName, + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.IsNotNull(method, $"Expected {methodName} to exist."); + method.Invoke(_client, new[] { argument }); + } + + private static void SetDisconnectionLastEventReceivedAt(StreamChatLowLevelClient client, DateTimeOffset value) + { + var field = typeof(StreamChatLowLevelClient).GetField("_disconnectionLastEventReceivedAt", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.IsNotNull(field, "Expected _disconnectionLastEventReceivedAt to exist."); + field.SetValue(client, (DateTimeOffset?)value); + } + + private static bool RequestBodyContains(object requestBody, string value) + { + var json = requestBody as string ?? requestBody?.ToString() ?? string.Empty; + return json.IndexOf(value, StringComparison.Ordinal) >= 0; + } + + private static bool RequestHasJsonBool(object requestBody, string property, bool value) + { + var json = requestBody as string ?? requestBody?.ToString() ?? string.Empty; + var needle = "\"" + property + "\":" + (value ? "true" : "false"); + return json.IndexOf(needle, StringComparison.Ordinal) >= 0; + } + + private const string HealthCheckJson = "{\"connection_id\":\"fakeId\",\"type\":\"health.check\"}"; + + private readonly Queue _pendingWebsocketMessages = new Queue(); + private readonly List _recoveredEvents = + new List(); + + private StreamChatClient _client; + private StreamClientConfig _config; + private AuthCredentials _authCredentials; + private IWebsocketClient _mockWebsocketClient; + private IApplicationInfo _mockApplicationInfo; + private ILogs _mockLogs; + private ITimeService _mockTimeService; + private INetworkMonitor _mockNetworkMonitor; + private IHttpClient _mockHttpClient; + } +} +#endif diff --git a/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs.meta b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs.meta new file mode 100644 index 00000000..7ae2eba4 --- /dev/null +++ b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs.meta @@ -0,0 +1,2 @@ +fileFormatVersion: 2 +guid: 533cd95f3d407dc42934ec9bb226a9e0 \ No newline at end of file diff --git a/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryLowLevelTests.cs b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryLowLevelTests.cs new file mode 100644 index 00000000..51b89097 --- /dev/null +++ b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryLowLevelTests.cs @@ -0,0 +1,315 @@ +#if STREAM_TESTS_ENABLED +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reflection; +using NSubstitute; +using NUnit.Framework; +using StreamChat.Core.Configs; +using StreamChat.Core.LowLevelClient; +using StreamChat.Libs.AppInfo; +using StreamChat.Libs.Auth; +using StreamChat.Libs.Http; +using StreamChat.Libs.Logs; +using StreamChat.Libs.NetworkMonitors; +using StreamChat.Libs.Serialization; +using StreamChat.Libs.Time; +using StreamChat.Libs.Websockets; + +namespace StreamChat.Tests.StateSync.Unit +{ + /// + /// Unit tests for the two history application modes on - + /// and + /// - and for the /sync request shape. + /// + internal class StateRecoveryLowLevelTests + { + [SetUp] + public void Up() + { + _authCredentials = new AuthCredentials("api123", "token123", "user123"); + _mockWebsocketClient = Substitute.For(); + _mockHttpClient = Substitute.For(); + _serializer = new NewtonsoftJsonSerializer(); + _mockTimeService = Substitute.For(); + _mockNetworkMonitor = Substitute.For(); + _mockApplicationInfo = Substitute.For(); + _mockLogs = Substitute.For(); + _mockStreamClientConfig = Substitute.For(); + + _lowLevelClient = CreateClient(); + _lowLevelClient.Update(0.1f); + + _mockHttpClient + .SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), Arg.Any(), Arg.Any()) + .Returns(new HttpResponse(true, 200, "{\"events\":[]}", null, null)); + } + + [TearDown] + public void TearDown() + { + for (var i = _clientsToDispose.Count - 1; i >= 0; i--) + { + _clientsToDispose[i].Dispose(); + } + + _clientsToDispose.Clear(); + _lowLevelClient = null; + } + + [Test] + public void when_sync_requested_with_more_than_100_cids_expect_only_100_sent() + { + var now = new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero); + _mockTimeService.Now.Returns(now); + SetDisconnectionLastEventReceivedAt(_lowLevelClient, now.AddHours(-1)); + + var cids = Enumerable.Range(0, 150).Select(i => $"messaging:channel-{i}").ToList(); + + _lowLevelClient.TrySyncHistoryAsync(cids).GetAwaiter().GetResult(); + + _mockHttpClient.Received(1).SendHttpRequestAsync( + Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith("/sync")), + Arg.Is(body => CountSyncCids(body) == StreamChatLowLevelClient.MaxSyncChannelCids)); + } + + [Test] + public void when_sync_requested_expect_inaccessible_cids_asked_for() + { + var now = new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero); + _mockTimeService.Now.Returns(now); + SetDisconnectionLastEventReceivedAt(_lowLevelClient, now.AddHours(-1)); + + _lowLevelClient.TrySyncHistoryAsync(new[] { "messaging:a" }).GetAwaiter().GetResult(); + + _mockHttpClient.Received(1).SendHttpRequestAsync( + Arg.Is(HttpMethodType.Post), + Arg.Is(uri => uri.AbsolutePath.EndsWith("/sync")), + Arg.Is(body => GetBoolMember(body, "WithInaccessibleCids") == true)); + } + + [Test] + public void when_history_batch_applied_expect_no_public_message_received() + { + var received = 0; + _lowLevelClient.MessageReceived += _ => received++; + + var result = _lowLevelClient.ApplyHistoryEvents(new List { MessageNewJson("msg-1", NewestCreatedAt) }); + + Assert.AreEqual(0, received); + Assert.AreEqual(0, result.FailedEventCount); + Assert.AreEqual(NewestCreatedAt, result.MaxAppliedCreatedAt); + } + + [Test] + public void when_history_batch_replayed_expect_public_message_received() + { + var received = 0; + _lowLevelClient.MessageReceived += _ => received++; + + _lowLevelClient.ReplayHistoryEvents(new List { MessageNewJson("msg-1", NewestCreatedAt) }); + + Assert.AreEqual(1, received); + } + + [Test] + public void when_history_batch_contains_custom_event_expect_it_delivered_per_event() + { + var received = new List(); + _lowLevelClient.CustomEventReceived += e => received.Add(e.Type); + + _lowLevelClient.ApplyHistoryEvents(new List + { + CustomEventJson("game.state", NewestCreatedAt), + }); + + Assert.AreEqual(new[] { "game.state" }, received.ToArray()); + } + + [Test] + public void when_history_batch_applied_expect_watermark_advanced_once_to_newest_applied_event() + { + var client = CreateClient(); + + client.ApplyHistoryEvents(new List + { + MessageNewJson("msg-1", NewestCreatedAt.AddMinutes(-10)), + MessageNewJson("msg-2", NewestCreatedAt), + MessageNewJson("msg-3", NewestCreatedAt.AddMinutes(-5)), + }); + + Assert.AreEqual(NewestCreatedAt, GetLastEventReceivedAt(client)); + } + + [Test] + public void when_history_batch_contains_malformed_event_expect_remaining_events_still_applied() + { + var client = CreateClient(); + var newest = NewestCreatedAt; + + var result = client.ApplyHistoryEvents(new List + { + MessageNewJson("msg-1", newest.AddMinutes(-10)), + $"{{\"type\":\"message.new\",\"cid\":\"messaging:test\",\"created_at\":\"{newest.AddMinutes(-1):O}\",\"message\":\"not-an-object\"}}", + MessageNewJson("msg-3", newest.AddMinutes(-5)), + }); + + Assert.AreEqual(1, result.FailedEventCount); + Assert.AreEqual(newest.AddMinutes(-5), result.MaxAppliedCreatedAt); + Assert.AreEqual(newest.AddMinutes(-5), GetLastEventReceivedAt(client)); + } + + [Test] + public void when_history_event_older_than_watermark_expect_watermark_not_regressed() + { + var client = CreateClient(); + SetLastEventReceivedAt(client, NewestCreatedAt); + + client.ApplyHistoryEvents(new List { MessageNewJson("msg-1", NewestCreatedAt.AddDays(-1)) }); + + Assert.AreEqual(NewestCreatedAt, GetLastEventReceivedAt(client)); + } + + [Test] + public void when_health_check_arrives_on_live_socket_expect_liveness_stamped() + { + var client = CreateClientWithMessages(HealthCheckJson()); + client.Connect(); + client.Update(0.2f); + + _mockTimeService.Time.Returns(12f); + EnqueueMessages(HealthCheckJson()); + client.Update(0.2f); + + Assert.AreEqual(12f, GetLastHealthCheckReceivedTime(client)); + } + + [Test] + public void when_health_check_arrives_from_history_replay_expect_liveness_not_stamped() + { + var client = CreateClientWithMessages(HealthCheckJson()); + client.Connect(); + client.Update(0.2f); + + var stampedOnConnect = GetLastHealthCheckReceivedTime(client); + + _mockTimeService.Time.Returns(99f); + client.ReplayHistoryEvents(new List { HealthCheckJson() }); + + Assert.AreEqual(stampedOnConnect, GetLastHealthCheckReceivedTime(client)); + } + + private static readonly DateTimeOffset NewestCreatedAt = + new DateTimeOffset(2026, 8, 10, 11, 0, 0, TimeSpan.Zero); + + private const string TestCid = "messaging:test"; + + private static string MessageNewJson(string messageId, DateTimeOffset createdAt) + => $"{{\"type\":\"message.new\",\"cid\":\"{TestCid}\",\"created_at\":\"{createdAt:O}\"," + + $"\"message\":{{\"id\":\"{messageId}\",\"text\":\"hi\",\"created_at\":\"{createdAt:O}\"," + + $"\"updated_at\":\"{createdAt:O}\",\"user\":{{\"id\":\"user-1\"}}}}}}"; + + private static string CustomEventJson(string type, DateTimeOffset createdAt) + => $"{{\"type\":\"{type}\",\"cid\":\"{TestCid}\",\"created_at\":\"{createdAt:O}\"," + + "\"user\":{\"id\":\"user-1\"}}"; + + private static string HealthCheckJson() + => "{\"connection_id\":\"fakeId\",\"type\":\"health.check\"}"; + + private static int CountSyncCids(object requestBody) + { + var list = GetMember(requestBody, "ChannelCids") as System.Collections.IList; + return list?.Count ?? -1; + } + + private static bool? GetBoolMember(object requestBody, string name) => GetMember(requestBody, name) as bool?; + + private static object GetMember(object requestBody, string name) + { + const BindingFlags flags = BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic; + + var property = requestBody.GetType().GetProperty(name, flags); + if (property != null) + { + return property.GetValue(requestBody); + } + + return requestBody.GetType().GetField(name, flags)?.GetValue(requestBody); + } + + private static void SetDisconnectionLastEventReceivedAt(StreamChatLowLevelClient client, DateTimeOffset value) + => GetPrivateField("_disconnectionLastEventReceivedAt").SetValue(client, (DateTimeOffset?)value); + + private static void SetLastEventReceivedAt(StreamChatLowLevelClient client, DateTimeOffset value) + => GetPrivateField("_lastEventReceivedAt").SetValue(client, (DateTimeOffset?)value); + + private static DateTimeOffset? GetLastEventReceivedAt(StreamChatLowLevelClient client) + => (DateTimeOffset?)GetPrivateField("_lastEventReceivedAt").GetValue(client); + + private static float GetLastHealthCheckReceivedTime(StreamChatLowLevelClient client) + => (float)GetPrivateField("_lastHealthCheckReceivedTime").GetValue(client); + + private static FieldInfo GetPrivateField(string name) + { + var field = typeof(StreamChatLowLevelClient).GetField(name, + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.IsNotNull(field, $"Expected {name} field to exist."); + return field; + } + + private StreamChatLowLevelClient CreateClient() + { + var client = new StreamChatLowLevelClient(_authCredentials, _mockWebsocketClient, _mockHttpClient, + _serializer, _mockTimeService, _mockNetworkMonitor, _mockApplicationInfo, _mockLogs, + _mockStreamClientConfig); + + _clientsToDispose.Add(client); + return client; + } + + private StreamChatLowLevelClient CreateClientWithMessages(params string[] websocketMessages) + { + var client = CreateClient(); + _mockWebsocketClient.ConnectAsync(Arg.Any()).Returns(System.Threading.Tasks.Task.CompletedTask); + + _mockWebsocketClient.TryDequeueMessage(out Arg.Any()).Returns(arg => + { + if (_pendingWebsocketMessages.Count == 0) + { + return false; + } + + arg[0] = _pendingWebsocketMessages.Dequeue(); + return true; + }); + + EnqueueMessages(websocketMessages); + return client; + } + + private void EnqueueMessages(params string[] websocketMessages) + { + foreach (var message in websocketMessages) + { + _pendingWebsocketMessages.Enqueue(message); + } + } + + private readonly List _clientsToDispose = new List(); + private readonly Queue _pendingWebsocketMessages = new Queue(); + + private StreamChatLowLevelClient _lowLevelClient; + private AuthCredentials _authCredentials; + private IWebsocketClient _mockWebsocketClient; + private IApplicationInfo _mockApplicationInfo; + private ILogs _mockLogs; + private ISerializer _serializer; + private ITimeService _mockTimeService; + private INetworkMonitor _mockNetworkMonitor; + private IHttpClient _mockHttpClient; + private IStreamClientConfig _mockStreamClientConfig; + } +} +#endif diff --git a/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryLowLevelTests.cs.meta b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryLowLevelTests.cs.meta new file mode 100644 index 00000000..f6c190db --- /dev/null +++ b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryLowLevelTests.cs.meta @@ -0,0 +1,2 @@ +fileFormatVersion: 2 +guid: b019eca0edff8ba44971000de05399a6 \ No newline at end of file