Rendezvous/tests/FinalFactory.Rendezvous.Tests/Browser/SessionStreamServiceTests.cs

using FinalFactory.Rendezvous.Contracts;
using FinalFactory.Rendezvous.Server.Browser;
using FinalFactory.Rendezvous.Server.State;
using FinalFactory.Rendezvous.Tests.State;

namespace FinalFactory.Rendezvous.Tests.Browser;

public sealed class SessionStreamServiceTests
{
    [Fact]
    public void SnapshotPlusUpdateMatchesFreshProjection()
    {
        using SessionBrowserFixture fixture = new();
        StoredListing listing = fixture.Add();
        BrowseSessionsRequest request = fixture.Request();
        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
        using SessionStreamSubscription subscription = AssertSuccess(
            fixture.Streams.Subscribe(request, snapshot.StreamCursor));

        StoreResult<StoredListing> updated = fixture.Store.UpdateListing(Update(
            listing,
            displayName: "Updated host",
            currentPlayers: 4));
        Assert.True(updated.Succeeded);
        SessionStreamEvent delta = Assert.Single(fixture.Streams.Read(subscription).Events);
        Assert.Equal(SessionStreamEventKind.SessionUpsert, delta.Kind);
        Assert.Equal("Updated host", delta.Session!.DisplayName);
        Assert.Equal(4, delta.Session.Capacity.CurrentPlayers);

        BrowseSessionsResponse fresh = AssertSuccess(fixture.Browser.Browse(request));
        SessionListing expected = Assert.Single(fresh.Items);
        Assert.Equal(expected.DisplayName, delta.Session.DisplayName);
        Assert.Equal(expected.Capacity.CurrentPlayers, delta.Session.Capacity.CurrentPlayers);
        Assert.DoesNotContain("lease", System.Text.Json.JsonSerializer.Serialize(delta, ContractJson.Options), StringComparison.OrdinalIgnoreCase);
    }

    [Fact]
    public void PresenceStalenessRecoveryAndRevocationProduceRemoveUpsertRemove()
    {
        using SessionBrowserFixture fixture = new();
        StoredListing listing = fixture.Add();
        BrowseSessionsRequest request = fixture.Request();
        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
        using SessionStreamSubscription subscription = AssertSuccess(
            fixture.Streams.Subscribe(request, snapshot.StreamCursor));

        fixture.Clock.Advance(TimeSpan.FromSeconds(21));
        AssertSuccess(fixture.Browser.Browse(request));
        SessionStreamEvent stale = Assert.Single(fixture.Streams.Read(subscription).Events);
        Assert.Equal(SessionStreamEventKind.SessionRemove, stale.Kind);
        Assert.Equal(listing.Definition.ListingId, stale.ListingId);

        StoreResult<StoredListing> rebound = fixture.Store.BindHostPresence(new(
            listing.Definition.HostPresenceHandle,
            listing.Definition.HostPresenceFingerprint,
            new ObservedEndpoint(AddressFamilyKind.Ipv4, "203.0.113.20", 40_020),
            null));
        Assert.True(rebound.Succeeded);
        Assert.Equal(
            SessionStreamEventKind.SessionUpsert,
            Assert.Single(fixture.Streams.Read(subscription).Events).Kind);

        Assert.True(fixture.Store.RevokeListing(listing.Definition.ListingId).Succeeded);
        Assert.Equal(
            SessionStreamEventKind.SessionRemove,
            Assert.Single(fixture.Streams.Read(subscription).Events).Kind);
    }

    [Fact]
    public void CreationAndLeaseExpiryProduceUpsertThenRemove()
    {
        using SessionBrowserFixture fixture = new();
        BrowseSessionsRequest request = fixture.Request();
        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
        using SessionStreamSubscription subscription = AssertSuccess(
            fixture.Streams.Subscribe(request, snapshot.StreamCursor));

        StoredListing listing = fixture.Add();
        SessionStreamEvent created = Assert.Single(fixture.Streams.Read(subscription).Events);
        Assert.Equal(SessionStreamEventKind.SessionUpsert, created.Kind);
        Assert.Equal(listing.Definition.ListingId, created.Session!.ListingId);

        for (int refresh = 0; refresh < 3; refresh++)
        {
            fixture.Clock.Advance(TimeSpan.FromSeconds(19));
            Assert.True(fixture.Store.BindHostPresence(new(
                listing.Definition.HostPresenceHandle,
                listing.Definition.HostPresenceFingerprint,
                new ObservedEndpoint(AddressFamilyKind.Ipv4, "203.0.113.20", 40_020),
                null)).Succeeded);
        }
        fixture.Clock.Advance(TimeSpan.FromSeconds(4));
        AssertSuccess(fixture.Browser.Browse(request));

        SessionStreamEvent expired = Assert.Single(fixture.Streams.Read(subscription).Events);
        Assert.Equal(SessionStreamEventKind.SessionRemove, expired.Kind);
        Assert.Equal(listing.Definition.ListingId, expired.ListingId);
        Assert.Equal(
            StoreResultCode.NotFound,
            fixture.Store.GetListing(listing.Definition.ListingId, requireFreshPresence: false).Code);
    }

    [Fact]
    public void ScopeProtocolRegionAndFullFiltersNeverLeak()
    {
        using SessionBrowserFixture fixture = new();
        BrowseSessionsRequest request = fixture.Request();
        request.ExcludeFull = true;
        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
        using SessionStreamSubscription subscription = AssertSuccess(
            fixture.Streams.Subscribe(request, snapshot.StreamCursor));

        fixture.Add(scope: new(new("other-game"), fixture.Scope.EnvironmentId));
        fixture.Add(protocolVersion: 99);
        fixture.Add(regionId: new("other-region"));
        fixture.Add(currentPlayers: 8, maximumPlayers: 8);
        Assert.Empty(fixture.Streams.Read(subscription).Events);
    }

    [Fact]
    public void VisibilityCompatibilityAndRegionChangesEnterAndLeaveTheFilter()
    {
        using SessionBrowserFixture fixture = new();
        StoredListing listing = fixture.Add();
        BrowseSessionsRequest request = fixture.Request();
        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
        using SessionStreamSubscription subscription = AssertSuccess(
            fixture.Streams.Subscribe(request, snapshot.StreamCursor));

        listing = fixture.Store.UpdateListing(Update(
            listing,
            listing.Definition.DisplayName,
            1,
            visibility: ListingVisibility.Unlisted)).Value!;
        Assert.Equal(SessionStreamEventKind.SessionRemove, SingleKind(fixture, subscription));
        listing = fixture.Store.UpdateListing(Update(
            listing,
            listing.Definition.DisplayName,
            1,
            visibility: ListingVisibility.Public)).Value!;
        Assert.Equal(SessionStreamEventKind.SessionUpsert, SingleKind(fixture, subscription));
        listing = fixture.Store.UpdateListing(Update(
            listing,
            listing.Definition.DisplayName,
            1,
            protocolVersion: 99)).Value!;
        Assert.Equal(SessionStreamEventKind.SessionRemove, SingleKind(fixture, subscription));
        listing = fixture.Store.UpdateListing(Update(
            listing,
            listing.Definition.DisplayName,
            1,
            protocolVersion: 7)).Value!;
        Assert.Equal(SessionStreamEventKind.SessionUpsert, SingleKind(fixture, subscription));
        listing = fixture.Store.UpdateListing(Update(
            listing,
            listing.Definition.DisplayName,
            1,
            regionId: new RegionId("other-region"))).Value!;
        Assert.Equal(SessionStreamEventKind.SessionRemove, SingleKind(fixture, subscription));
    }

    [Fact]
    public void ReplayGapAndForeignCursorForceResetAndSubscriberLimitFailsClosed()
    {
        ManualRendezvousClock clock = new();
        SessionChangeJournal changes = new(new SessionChangeJournalOptions
        {
            ReplayCapacity = 64,
            MaximumSubscribers = 1,
            MaximumSubscribersPerTenant = 1,
        });
        using SessionStreamCursorCodec cursors = new();
        SessionStreamService streams = new(changes, cursors, clock);
        EphemeralStateFixture state = new(changes: changes);
        StoredListing listing = state.CreateVisibleListing(out _);
        BrowseSessionsRequest request = new()
        {
            GameId = state.Scope.GameId,
            EnvironmentId = state.Scope.EnvironmentId,
            ProtocolVersion = listing.Definition.ProtocolVersion,
        };
        VisibleListingQuery query = new(state.Scope, listing.Definition.ProtocolVersion, null);
        string initial = cursors.Encode(query, changes.CurrentRevision, clock.UtcNow);
        using SessionStreamSubscription subscription = AssertSuccess(streams.Subscribe(request, initial));
        Assert.Equal(
            RendezvousErrorCode.CapacityExceeded,
            streams.Subscribe(request, initial).Error);

        for (int index = 0; index < 65; index++)
        {
            listing = state.Store.UpdateListing(Update(
                listing,
                displayName: $"Host {index}",
                currentPlayers: index % 8)).Value!;
        }
        Assert.True(streams.Read(subscription).RequiresReset);
        subscription.Dispose();

        using SessionStreamSubscription foreign = AssertSuccess(streams.Subscribe(
            request,
            "not-a-valid-cursor"));
        Assert.True(streams.Read(foreign).RequiresReset);
    }

    [Fact]
    public void BurstIsBoundedAndCoalescedWithoutLosingFinalState()
    {
        ManualRendezvousClock clock = new();
        SessionChangeJournal changes = new(new SessionChangeJournalOptions
        {
            ReplayCapacity = 1024,
            MaximumBatchSize = 128,
        });
        using SessionStreamCursorCodec cursors = new();
        SessionStreamService streams = new(changes, cursors, clock);
        EphemeralStateFixture state = new(changes: changes);
        StoredListing listing = state.CreateVisibleListing(out _);
        BrowseSessionsRequest request = new()
        {
            GameId = state.Scope.GameId,
            EnvironmentId = state.Scope.EnvironmentId,
            ProtocolVersion = listing.Definition.ProtocolVersion,
        };
        VisibleListingQuery query = new(state.Scope, listing.Definition.ProtocolVersion, null);
        using SessionStreamSubscription subscription = AssertSuccess(streams.Subscribe(
            request,
            cursors.Encode(query, changes.CurrentRevision, clock.UtcNow)));

        for (int index = 0; index < 1000; index++)
        {
            listing = state.Store.UpdateListing(Update(
                listing,
                displayName: $"Host {index}",
                currentPlayers: index % 8)).Value!;
        }

        List<SessionStreamEvent> emitted = [];
        while (subscription.Revision < changes.CurrentRevision)
        {
            SessionStreamReadResult read = streams.Read(subscription);
            Assert.False(read.RequiresReset);
            emitted.AddRange(read.Events);
        }

        Assert.Equal(8, emitted.Count);
        SessionStreamEvent final = emitted[^1];
        Assert.Equal(SessionStreamEventKind.SessionUpsert, final.Kind);
        Assert.Equal("Host 999", final.Session!.DisplayName);
        Assert.Equal(7, final.Session.Capacity.CurrentPlayers);
    }

    [Fact]
    public void SubscriberLimitIsEnforcedPerTenantAndReleasedOnDispose()
    {
        ManualRendezvousClock clock = new();
        SessionChangeJournal changes = new(new SessionChangeJournalOptions
        {
            MaximumSubscribers = 2,
            MaximumSubscribersPerTenant = 1,
        });
        using SessionStreamCursorCodec cursors = new();
        SessionStreamService streams = new(changes, cursors, clock);
        TenantScope firstScope = new(new("first-game"), new("production"));
        TenantScope secondScope = new(new("second-game"), new("production"));
        BrowseSessionsRequest firstRequest = Request(firstScope);
        BrowseSessionsRequest secondRequest = Request(secondScope);
        string firstCursor = cursors.Encode(
            new VisibleListingQuery(firstScope, 7, null),
            changes.CurrentRevision,
            clock.UtcNow);
        string secondCursor = cursors.Encode(
            new VisibleListingQuery(secondScope, 7, null),
            changes.CurrentRevision,
            clock.UtcNow);

        SessionStreamSubscription first = AssertSuccess(streams.Subscribe(firstRequest, firstCursor));
        Assert.Equal(
            RendezvousErrorCode.CapacityExceeded,
            streams.Subscribe(firstRequest, firstCursor).Error);
        using SessionStreamSubscription second = AssertSuccess(
            streams.Subscribe(secondRequest, secondCursor));
        first.Dispose();
        using SessionStreamSubscription replacement = AssertSuccess(
            streams.Subscribe(firstRequest, firstCursor));
    }

    private static BrowseSessionsRequest Request(TenantScope scope) => new()
    {
        GameId = scope.GameId,
        EnvironmentId = scope.EnvironmentId,
        ProtocolVersion = 7,
    };

    private static UpdateListingCommand Update(
        StoredListing listing,
        string displayName,
        int currentPlayers,
        RegionId? regionId = null,
        uint? protocolVersion = null,
        ListingVisibility? visibility = null) => new(
        listing.Definition.ListingId,
        listing.Definition.LeaseId,
        listing.Definition.LeaseFingerprint,
        listing.Definition.OwnerSubject,
        listing.Definition.BuildVersion,
        displayName,
        currentPlayers,
        listing.Definition.MaximumPlayers,
        listing.Definition.Metadata,
        listing.Definition.DedicatedFallback,
        regionId,
        protocolVersion,
        visibility);

    private static SessionStreamEventKind SingleKind(
        SessionBrowserFixture fixture,
        SessionStreamSubscription subscription) =>
        Assert.Single(fixture.Streams.Read(subscription).Events).Kind;

    private static T AssertSuccess<T>(BrowserServiceResult<T> result)
    {
        Assert.True(result.Succeeded, result.Error.ToString());
        return Assert.IsType<T>(result.Value);
    }
}