Skip to content

Advance _lastEventReceivedAt only by newer events. #234

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sierpinskid merged 1 commit into from
Aug 19, 2026
Merged

Advance _lastEventReceivedAt only by newer events. #234

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 23 additions & 4 deletions Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -996,7 +996,7 @@ private void RegisterEventType<TDto, TEvent>(string key,
#endif
var eventObj = DeserializeEvent<TDto, TEvent>(serializedContent, out var dto);
postprocess?.Invoke(dto);
_lastEventReceivedAt = eventObj.CreatedAt;
TryAdvanceLastEventReceivedAt(eventObj.CreatedAt, key);
handler?.Invoke(eventObj, dto);
internalHandler?.Invoke(dto);
}
Expand Down Expand Up @@ -1054,7 +1054,7 @@ private void HandleNewWebsocketMessage(string msg)

if (!_eventKeyToHandler.TryGetValue(type, out var handler))
{
if (TryHandleCustomChannelEvent(msg))
if (TryHandleCustomChannelEvent(msg, type))
{
return;
}
Expand All @@ -1070,7 +1070,7 @@ private void HandleNewWebsocketMessage(string msg)
handler(msg);
}

private bool TryHandleCustomChannelEvent(string serializedContent)
private bool TryHandleCustomChannelEvent(string serializedContent, string eventType)
{
if (!_serializer.TryPeekValue<string>(serializedContent, "cid", out var cid)
|| string.IsNullOrEmpty(cid))
Expand All @@ -1081,7 +1081,7 @@ private bool TryHandleCustomChannelEvent(string serializedContent)
try
{
var dto = _serializer.Deserialize<CustomEventInternalDTO>(serializedContent);
_lastEventReceivedAt = dto.CreatedAt;
TryAdvanceLastEventReceivedAt(dto.CreatedAt, eventType);

var evt = new EventCustom();
((ILoadableFrom<CustomEventInternalDTO, EventCustom>)evt).LoadFromDto(dto);
Expand Down Expand Up @@ -1141,6 +1141,25 @@ private void HandleHealthCheckEvent(EventHealthCheck healthCheckEvent, HealthChe
}
}

private void TryAdvanceLastEventReceivedAt(DateTimeOffset createdAt, string eventType)
{
if (createdAt == DateTimeOffset.MinValue)
{
if (_config.LogLevel.IsDebugEnabled())
{
_logs.Warning(
$"WebSocket event `{eventType}` has no valid `created_at`; the /sync watermark was not advanced.");
}

return;
}

if (!_lastEventReceivedAt.HasValue || createdAt > _lastEventReceivedAt.Value)
{
_lastEventReceivedAt = createdAt;
}
}

private static bool IsUserIdValid(string userId)
{
var r = new Regex("^[a-zA-Z0-9@_-]+$");
Expand Down
56 changes: 49 additions & 7 deletions Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
using System.Collections.Generic;
using System.Linq;
using System.Net.WebSockets;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using NSubstitute;
Expand Down Expand Up @@ -347,6 +348,27 @@ public void when_connection_state_changed_subscriber_throws_expect_remaining_sub
Assert.AreNotEqual(ConnectionState.Connected, lastStateSeenByLateSubscriber);
}

[Test]
public void when_event_without_created_at_expect_last_event_watermark_not_set()
{
var client = CreateConnectedClient();

Assert.IsNull(GetLastEventReceivedAt(client));
}

[Test]
public void when_event_with_created_at_expect_last_event_watermark_set()
{
var createdAt = new DateTimeOffset(2026, 8, 18, 13, 58, 59, TimeSpan.Zero);
var client = CreateClientWithMessages(logs: null,
$"{{\"connection_id\":\"fakeId\", \"type\":\"health.check\", \"created_at\":\"{createdAt:O}\"}}");

client.Connect();
client.Update(deltaTime: 0.2f);

Assert.AreEqual(createdAt, GetLastEventReceivedAt(client));
}

private readonly List<IDisposable> _resourcesToDispose = new List<IDisposable>();

private IStreamChatLowLevelClient _lowLevelClient;
Expand All @@ -362,6 +384,17 @@ public void when_connection_state_changed_subscriber_throws_expect_remaining_sub
private IStreamClientConfig _mockStreamClientConfig;

private StreamChatLowLevelClient CreateConnectedClient(ILogs logs = null)
{
var client = CreateClientWithMessages(logs, "{\"connection_id\":\"fakeId\", \"type\":\"health.check\"}");
client.Connect();
client.Update(deltaTime: 0.2f);

Assert.IsTrue(client.ConnectionState == ConnectionState.Connected);

return client;
}

private StreamChatLowLevelClient CreateClientWithMessages(ILogs logs, params string[] websocketMessages)
{
var client = new StreamChatLowLevelClient(_authCredentials, _mockWebsocketClient, _mockHttpClient,
new NewtonsoftJsonSerializer(), _mockTimeService, _mockNetworkMonitor, _mockApplicationInfo,
Expand All @@ -370,19 +403,28 @@ private StreamChatLowLevelClient CreateConnectedClient(ILogs logs = null)

_mockWebsocketClient.ConnectAsync(Arg.Any<Uri>()).Returns(Task.CompletedTask);

var messages = new Queue<string>(websocketMessages);
_mockWebsocketClient.TryDequeueMessage(out Arg.Any<string>()).Returns(arg =>
{
arg[0] = "{\"connection_id\":\"fakeId\", \"type\":\"health.check\"}";
return true;
}, arg => false);

client.Connect();
client.Update(deltaTime: 0.2f);
if (messages.Count == 0)
{
return false;
}

Assert.IsTrue(client.ConnectionState == ConnectionState.Connected);
arg[0] = messages.Dequeue();
return true;
});

return client;
}

private static DateTimeOffset? GetLastEventReceivedAt(StreamChatLowLevelClient client)
{
var field = typeof(StreamChatLowLevelClient).GetField("_lastEventReceivedAt",
BindingFlags.Instance | BindingFlags.NonPublic);
Assert.IsNotNull(field, "Expected _lastEventReceivedAt field to exist.");
return (DateTimeOffset?)field.GetValue(client);
}
}
}
#endif
63 changes: 60 additions & 3 deletions Assets/Plugins/StreamChat/Tests/StateSync/StateSyncCatchUpTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,44 @@ public void when_disconnection_timestamp_missing_expect_sync_not_called()
AssertSyncNotCalled();
}

/// <summary>
/// /sync replays historical events through the same handlers as live events. Because the catch-up runs
/// asynchronously after reconnect, live events can already have advanced the watermark by the time the
/// replay is processed. Rewinding it here would make the next disconnect sync from a stale point.
/// </summary>
[Test]
public void when_sync_replays_events_older_than_watermark_expect_watermark_not_regressed()
{
var now = new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero);
_mockTimeService.Now.Returns(now);

SetLastEventReceivedAt(now);
SetDisconnectionLastEventReceivedAt(now.AddHours(-1));
StubSyncResponseWithMessageEvent(now.AddHours(-1));

_lowLevelClient.FetchAndProcessEventsSinceLastReceivedEvent(new[] { TestChannelCid }).GetAwaiter()
.GetResult();

Assert.AreEqual(now, GetLastEventReceivedAt());
}

[Test]
public void when_sync_replays_events_newer_than_watermark_expect_watermark_advanced()
{
var now = new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero);
_mockTimeService.Now.Returns(now);

var replayedEventCreatedAt = now.AddHours(-1);
SetLastEventReceivedAt(now.AddHours(-2));
SetDisconnectionLastEventReceivedAt(now.AddHours(-2));
StubSyncResponseWithMessageEvent(replayedEventCreatedAt);

_lowLevelClient.FetchAndProcessEventsSinceLastReceivedEvent(new[] { TestChannelCid }).GetAwaiter()
.GetResult();

Assert.AreEqual(replayedEventCreatedAt, GetLastEventReceivedAt());
}

private const string TestChannelCid = "messaging:test-channel";

private StreamChatLowLevelClient _lowLevelClient;
Expand All @@ -120,12 +158,31 @@ private void AssertSyncNotCalled()
Arg.Is<Uri>(uri => uri.AbsolutePath.EndsWith("/sync")),
Arg.Any<object>());

private void StubSyncResponseWithMessageEvent(DateTimeOffset eventCreatedAt)
{
var body =
$"{{\"events\":[{{\"type\":\"message.new\",\"cid\":\"{TestChannelCid}\",\"created_at\":\"{eventCreatedAt:O}\"}}]}}";

_mockHttpClient
.SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), Arg.Any<Uri>(), Arg.Any<object>())
.Returns(new HttpResponse(true, 200, body, null, null));
}

private void SetDisconnectionLastEventReceivedAt(DateTimeOffset value)
=> GetPrivateField("_disconnectionLastEventReceivedAt").SetValue(_lowLevelClient, (DateTimeOffset?)value);

private void SetLastEventReceivedAt(DateTimeOffset value)
=> GetPrivateField("_lastEventReceivedAt").SetValue(_lowLevelClient, (DateTimeOffset?)value);

private DateTimeOffset? GetLastEventReceivedAt()
=> (DateTimeOffset?)GetPrivateField("_lastEventReceivedAt").GetValue(_lowLevelClient);

private static FieldInfo GetPrivateField(string name)
{
var field = typeof(StreamChatLowLevelClient).GetField("_disconnectionLastEventReceivedAt",
var field = typeof(StreamChatLowLevelClient).GetField(name,
BindingFlags.Instance | BindingFlags.NonPublic);
Assert.IsNotNull(field, "Expected _disconnectionLastEventReceivedAt field to exist.");
field.SetValue(_lowLevelClient, (DateTimeOffset?)value);
Assert.IsNotNull(field, $"Expected {name} field to exist.");
return field;
}
}
}
Expand Down
Loading