CultLib/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using FluentAssertions;
using GameCult.Caching;
using GameCult.Networking;
using MessagePack;
using NUnit.Framework;

namespace GameCult.Mesh.Tests;

public sealed class CultMeshClientTests
{
    [Test]
    public void Construction_RequiresOneRendezvousEndpoint()
    {
        var create = () => new CultMeshClient(new CultMeshClientOptions());

        create.Should().Throw<ArgumentException>()
            .WithMessage("*rendezvous endpoint*");
    }

    [Test]
    public async Task Connect_ResolvesStableIdentityAndReusesOneSession()
    {
        var discoveryClients = 0;
        var connector = new RecordingConnector();
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions
            {
                CreateClient = () =>
                {
                    Interlocked.Increment(ref discoveryClients);
                    return new RendezvousClient();
                }
            },
            Connectors = new[] { connector }
        });

        var first = await mesh.ConnectAsync("aetheria", CultMeshProtocols.Documents);
        var second = await mesh.ConnectAsync("aetheria", CultMeshProtocols.Documents);

        first.Should().BeSameAs(second);
        first.EndpointId.Value.Should().Be("aetheria");
        first.State.Path!.Endpoint.Should().Be("rudp://aetheria:3081");
        discoveryClients.Should().Be(1);
        connector.ConnectCount.Should().Be(1);
    }

    [Test]
    public async Task Document_ReplicatesTypedRecordBehindOneIntentDeclaration()
    {
        var connector = new DocumentConnector();
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector }
        });

        using var firstLease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
        using var secondLease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
        var first = firstLease.Handle;
        var second = secondLease.Handle;

        first.Should().BeSameAs(second);
        (await first.LatestAsync()).Text.Should().Be("pilot surface 1");
        connector.SubscribeCount.Should().Be(1);
    }

    [Test]
    public async Task Document_ReplaysSubscriptionAndRefreshesSameHandleAfterReconnect()
    {
        var connector = new DocumentConnector();
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector }
        });
        using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
        var document = lease.Handle;

        connector.Clients[0].Fail(new InvalidOperationException("path lost"));
        await WaitUntilAsync(() => connector.Clients.Count == 2 && connector.Clients[1].SubscribeCount == 1);
        await WaitForTextAsync(document, "pilot surface 2");

        (await document.LatestAsync()).Text.Should().Be("pilot surface 2");
        connector.Clients[0].SubscribeCount.Should().Be(1);
        connector.Clients[1].SubscribeCount.Should().Be(1);
    }

    [Test]
    public async Task Document_InitialReadinessCanBeSatisfiedByReconnectReplay()
    {
        var connector = new DocumentConnector { SuppressFirstSnapshot = true };
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector }
        });

        var opening = mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
        await WaitUntilAsync(() => connector.Clients.Count == 1 && connector.Clients[0].SubscribeCount == 1);
        connector.Clients[0].Fail(new InvalidOperationException("path lost before initial snapshot"));

        using var lease = await opening;
        var document = lease.Handle;

        (await document.LatestAsync()).Text.Should().Be("pilot surface 2");
        connector.Clients.Should().HaveCount(2);
        connector.Clients[1].SubscribeCount.Should().Be(1);
    }

    [Test]
    public async Task Document_ReplaysUnansweredIntentOnSamePhysicalSession()
    {
        var connector = new DocumentConnector { SuppressFirstSubscriptionSnapshot = true };
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector },
            SubscriptionResponseTimeout = TimeSpan.FromMilliseconds(20)
        });

        using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
        var document = lease.Handle;

        (await document.LatestAsync()).Text.Should().Be("pilot surface 1");
        connector.Clients.Should().ContainSingle();
        connector.Clients[0].SubscribeCount.Should().Be(2);
    }

    [Test]
    public async Task Collection_ReplicatesAllDocumentsOfTypedSchema()
    {
        var connector = new DocumentConnector();
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector }
        });

        using var lease = await mesh.LeaseCollectionAsync<ClientTestDocument>("aetheria");
        var collection = lease.Handle;

        (await collection.LatestAsync()).Should().ContainSingle()
            .Which.Text.Should().Be("pilot surface 1");
        connector.SubscribeCount.Should().Be(1);
    }

    [Test]
    public async Task Collection_InitialReadinessCanBeSatisfiedByReconnectReplay()
    {
        var connector = new DocumentConnector { SuppressFirstSnapshot = true };
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector }
        });

        var opening = mesh.LeaseCollectionAsync<ClientTestDocument>("aetheria");
        await WaitUntilAsync(() => connector.Clients.Count == 1 && connector.Clients[0].SubscribeCount == 1);
        connector.Clients[0].Fail(new InvalidOperationException("path lost before initial collection snapshot"));

        using var lease = await opening;
        var collection = lease.Handle;

        (await collection.LatestAsync()).Should().ContainSingle()
            .Which.Text.Should().Be("pilot surface 2");
        connector.Clients.Should().HaveCount(2);
        connector.Clients[1].SubscribeCount.Should().Be(1);
    }

    [Test]
    public async Task Document_ResourceIsSharedAndReleasedWithLastLease()
    {
        var connector = new DocumentConnector();
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector }
        });

        var first = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
        var second = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
        mesh.ActiveDocumentResourceCount.Should().Be(1);

        first.Dispose();
        mesh.ActiveDocumentResourceCount.Should().Be(1);
        second.Dispose();
        mesh.ActiveDocumentResourceCount.Should().Be(0);

        using var replacement = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
        replacement.Handle.Should().NotBeSameAs(first.Handle);
        connector.SubscribeCount.Should().Be(2);
    }

    [Test]
    public async Task Document_OneHundredThousandDistinctLeaseCyclesLeaveNoResourcesOrCallbacks()
    {
        const int leaseCount = 100_000;
        var connector = new DocumentConnector();
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector }
        });

        for (var index = 0; index < leaseCount; index++)
        {
            var recordKey = "surface:churn:" + index;
            using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", recordKey);
            (await lease.Handle.LatestAsync()).Id.Should().Be(recordKey);
        }

        mesh.ActiveDocumentResourceCount.Should().Be(0);
        mesh.ActiveCollectionResourceCount.Should().Be(0);
        connector.Clients.Should().ContainSingle();
        connector.Clients[0].HandlerRegistrationCount.Should().Be(2,
            "one physical handler per message type must serve every logical lease");
    }

    [Test]
    public async Task SubmitDocument_SendsUncommittedInputThroughStableIdentitySession()
    {
        var connector = new DocumentConnector();
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector }
        });
        var request = new ClientTestDocument { Id = "command:1", Text = "move" };

        await mesh.SubmitDocumentAsync(
            "aetheria",
            request.Id,
            request,
            sourceRuntimeId: "pilot-one",
            sourceRole: "pilot");

        connector.Clients.Should().ContainSingle();
        var submitted = connector.Clients[0].Submitted.Should().ContainSingle().Subject;
        submitted.Document.RecordKey.Should().Be("command:1");
        submitted.Document.SourceRuntimeId.Should().Be("pilot-one");
        submitted.Document.SourceRole.Should().Be("pilot");
    }

    [Test]
    public async Task Dispose_TerminatesDocumentWaitingForInitialSnapshot()
    {
        var connector = new DocumentConnector { SuppressFirstSnapshot = true };
        var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Connectors = new[] { connector }
        });
        var opening = mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
        await WaitUntilAsync(() => connector.Clients.Count == 1 && connector.Clients[0].SubscribeCount == 1);

        mesh.Dispose();

        Func<Task> awaitOpening = async () => await opening;
        await awaitOpening.Should().ThrowAsync<ObjectDisposedException>();
    }

    private sealed class RendezvousClient : ICultNetSchemaClient
    {
        private readonly List<Action<CultMeshVerseCatalogResponseMessage>> _handlers = new();
        public bool Connected => true;
        public void Connect(string host, int port) { }
        public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage
        {
            var request = (CultMeshVerseCatalogRequestMessage)(object)message;
            var response = new CultMeshVerseCatalogResponseMessage
            {
                MessageId = request.MessageId,
                Verses = new[]
                {
                    new CultMeshVerseDescriptor(
                        "aetheria",
                        "Aetheria",
                        CultMeshVerseAuthorityModel.OperatorCluster,
                        new CultMeshVerseCompatibility("cultmesh.v0", "rules"),
                        new[] { "rudp://aetheria:3081" }).ToMessage()
                }
            };
            foreach (var handler in _handlers.ToArray()) handler(response);
        }
        public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
        {
            if (typeof(T) == typeof(CultMeshVerseCatalogResponseMessage))
                _handlers.Add(response => callback((T)(object)response));
        }
        public void Dispose() { }
    }

    private sealed class RecordingConnector : ICultMeshTransportConnector
    {
        private int _connectCount;
        public string ConnectorId => "test";
        public int Priority => 0;
        public int ConnectCount => Volatile.Read(ref _connectCount);
        public bool CanConnect(CultMeshTransportCandidate candidate) => true;
        public Task<ICultNetSchemaClient> ConnectAsync(
            CultMeshTransportCandidate candidate,
            CultMeshProtocolId protocol,
            CancellationToken cancellationToken = default)
        {
            Interlocked.Increment(ref _connectCount);
            return Task.FromResult<ICultNetSchemaClient>(new ConnectedClient());
        }
    }

    private sealed class ConnectedClient : ICultNetSchemaClient
    {
        public bool Connected => true;
        public void Connect(string host, int port) { }
        public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage { }
        public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage { }
        public void Dispose() { }
    }

    private sealed class DocumentConnector : ICultMeshTransportConnector
    {
        public List<DocumentClient> Clients { get; } = new();
        public bool SuppressFirstSnapshot { get; set; }
        public bool SuppressFirstSubscriptionSnapshot { get; set; }
        public string ConnectorId => "document-test";
        public int Priority => 0;
        public int SubscribeCount => Clients.Sum(client => client.SubscribeCount);
        public bool CanConnect(CultMeshTransportCandidate candidate) => true;
        public Task<ICultNetSchemaClient> ConnectAsync(
            CultMeshTransportCandidate candidate,
            CultMeshProtocolId protocol,
            CancellationToken cancellationToken = default)
        {
            var client = new DocumentClient(
                "pilot surface " + (Clients.Count + 1),
                snapshotsToSuppress: SuppressFirstSnapshot && Clients.Count == 0
                    ? int.MaxValue
                    : SuppressFirstSubscriptionSnapshot && Clients.Count == 0 ? 1 : 0);
            Clients.Add(client);
            return Task.FromResult<ICultNetSchemaClient>(client);
        }
    }

    private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth
    {
        private readonly List<Action<CultNetSnapshotResponseRawMessage>> _snapshots = new();
        private readonly CultNetDocumentRegistry _documents;
        private readonly string _text;
        private readonly TaskCompletionSource<Exception> _failure = new(TaskCreationOptions.RunContinuationsAsynchronously);
        private int _snapshotsToSuppress;
        public DocumentClient(string text, int snapshotsToSuppress = 0)
        {
            _text = text;
            _snapshotsToSuppress = snapshotsToSuppress;
            var cache = CultMesh.CreateCultCacheDocumentRegistry(typeof(ClientTestDocument));
            _documents = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(ClientTestDocument) }, cache);
        }
        public bool Connected => true;
        public Task<Exception> BackgroundFailure => _failure.Task;
        public int SubscribeCount { get; private set; }
        public int HandlerRegistrationCount { get; private set; }
        public List<CultNetDocumentPutRawMessage> Submitted { get; } = new();
        public void Connect(string host, int port) { }
        public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage
        {
            if (message is CultNetDocumentPutRawMessage submitted)
            {
                Submitted.Add(submitted);
                return;
            }
            if (message is not CultNetDatabaseSubscribeMessage request) return;
            SubscribeCount++;
            if (_snapshotsToSuppress > 0)
            {
                _snapshotsToSuppress--;
                return;
            }
            var recordKey = request.RecordKeys?.SingleOrDefault() ?? "surface:pilot";
            var put = _documents.CreateRawDocumentPutMessage(
                "test-put",
                new CultRecordHandle<ClientTestDocument>(new CultRecordKey(recordKey)),
                new ClientTestDocument { Id = recordKey, Text = _text });
            var response = new CultNetSnapshotResponseRawMessage
            {
                MessageId = request.MessageId,
                Documents = new[] { put.Document }
            };
            foreach (var handler in _snapshots.ToArray()) handler(response);
        }
        public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
        {
            HandlerRegistrationCount++;
            if (typeof(T) == typeof(CultNetSnapshotResponseRawMessage))
                _snapshots.Add(response => callback((T)(object)response));
        }
        public void Dispose() { }
        public void Fail(Exception error) => _failure.TrySetResult(error);
    }

    private static async Task WaitUntilAsync(Func<bool> condition)
    {
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(2);
        while (!condition() && DateTime.UtcNow < deadline) await Task.Delay(1);
        condition().Should().BeTrue();
    }

    private static async Task WaitForTextAsync(
        CultMeshDocumentHandle<ClientTestDocument> document,
        string expected)
    {
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(2);
        while (DateTime.UtcNow < deadline)
        {
            if (string.Equals((await document.LatestAsync()).Text, expected, StringComparison.Ordinal)) return;
            await Task.Delay(1);
        }
        (await document.LatestAsync()).Text.Should().Be(expected);
    }

    [CultDocument("tests.cultmesh_client_document", "tests.cultmesh_client_document.v1")]
    [MessagePackObject]
    public sealed class ClientTestDocument
    {
        [Key(0), CultName] public string Id { get; set; } = string.Empty;
        [Key(1)] public string Text { get; set; } = string.Empty;
    }
}