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
{
    private static CultMeshSessionTarget Target => new("aetheria", "aetheria-daemon");

    [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();
                }
            },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });

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

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

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

        Func<Task> connect = async () => await mesh.ConnectAsync(
            new CultMeshSessionTarget("aetheria", "forged-provider"),
            CultMeshProtocols.Documents);

        var failure = await connect.Should().ThrowAsync<CultMeshSessionException>();
        failure.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Resolution);
        connector.ConnectCount.Should().Be(0,
            "a valid Verse route must not authorize an unadvertised provider identity");
    }

    [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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });

        using var firstLease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
        using var secondLease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });
        using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });

        var opening = mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector },
            SubscriptionResponseTimeout = TimeSpan.FromMilliseconds(20)
        });

        using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });

        using var lease = await mesh.LeaseCollectionAsync<ClientTestDocument>(Target);
        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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });

        var opening = mesh.LeaseCollectionAsync<ClientTestDocument>(Target);
        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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });

        var first = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
        var second = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "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>(Target, "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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });

        for (var index = 0; index < leaseCount; index++)
        {
            var recordKey = "surface:churn:" + index;
            using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, 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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });
        var request = new ClientTestDocument { Id = "command:1", Text = "move" };

        await mesh.SubmitDocumentAsync(
            Target,
            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 InvokeOperation_CorrelatesProviderReceiptThroughStableIdentitySession()
    {
        var connector = new DocumentConnector();
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });

        var result = await mesh.InvokeAsync<ClientOperationRequest, ClientOperationReceipt>(
            Target,
            "tests.counter",
            "tests.counter.increment",
            "tests.counter.increment_request.v1",
            "tests.counter.increment_receipt.v1",
            new ClientOperationRequest { Amount = 2 },
            sourceRuntimeId: "pilot-one",
            idempotencyKey: "pilot-command-1");

        result.MessageId.Should().Be("pilot-command-1");
        result.Status.Should().Be("accepted");
        result.SourceRuntimeId.Should().Be("aetheria-daemon");
        result.Value.ReceiptId.Should().Be("receipt:pilot-command-1");
        result.Value.Count.Should().Be(2);
        connector.Clients.Should().ContainSingle();
        connector.Clients[0].Operations.Should().ContainSingle()
            .Which.SourceRuntimeId.Should().Be("pilot-one");
    }

    [Test]
    public async Task InvokeOperation_ReplaysSameIdempotencyKeyAfterRouteLossBeforeReceipt()
    {
        var connector = new DocumentConnector { FailFirstOperationBeforeResponse = true };
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector },
            OperationResponseTimeout = TimeSpan.FromSeconds(2)
        });

        var result = await mesh.InvokeAsync<ClientOperationRequest, ClientOperationReceipt>(
            Target,
            "tests.counter",
            "tests.counter.increment",
            "tests.counter.increment_request.v1",
            "tests.counter.increment_receipt.v1",
            new ClientOperationRequest { Amount = 2 },
            sourceRuntimeId: "pilot-one",
            idempotencyKey: "pilot-command-route-loss");

        result.Value.ReceiptId.Should().Be("receipt:pilot-command-route-loss");
        connector.Clients.Should().HaveCount(2);
        connector.Clients.SelectMany(client => client.Operations)
            .Select(operation => operation.MessageId)
            .Should().Equal("pilot-command-route-loss", "pilot-command-route-loss");
    }

    [Test]
    public async Task InvokeOperation_ThrowsCorrelatedFrameworkFailureWithoutTimingOut()
    {
        var connector = new DocumentConnector { RejectOperations = true };
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector },
            OperationResponseTimeout = TimeSpan.FromSeconds(5)
        });

        Func<Task> invoke = async () => await mesh.InvokeAsync<ClientOperationRequest, ClientOperationReceipt>(
            Target,
            "tests.counter",
            "tests.counter.increment",
            "wrong.request.v1",
            "tests.counter.increment_receipt.v1",
            new ClientOperationRequest { Amount = 2 },
            sourceRuntimeId: "pilot-one",
            idempotencyKey: "pilot-command-invalid");

        var failure = await invoke.Should().ThrowAsync<CultMeshRemoteOperationException>();
        failure.Which.Status.Should().Be("invalid");
        failure.Which.Code.Should().Be("request-schema-mismatch");
        failure.Which.Message.Should().Contain("expected payload schema");
        connector.Clients.Should().ContainSingle();
        connector.Clients[0].Operations.Should().ContainSingle();
    }

    [Test]
    public async Task InvokeOperation_RejectsReceiptFromAnotherRuntime()
    {
        var connector = new DocumentConnector { ResponseSourceRuntimeId = "intruder-daemon" };
        using var mesh = new CultMeshClient(new CultMeshClientOptions
        {
            RendezvousEndpoints = new[] { "rudp://odin:3076" },
            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });

        Func<Task> invoke = async () => await mesh.InvokeAsync<ClientOperationRequest, ClientOperationReceipt>(
            Target,
            "tests.counter",
            "tests.counter.increment",
            "tests.counter.increment_request.v1",
            "tests.counter.increment_receipt.v1",
            new ClientOperationRequest { Amount = 2 },
            sourceRuntimeId: "pilot-one",
            idempotencyKey: "pilot-command-wrong-source");

        var failure = await invoke.Should().ThrowAsync<CultMeshSessionException>();
        failure.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority);
    }

    [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() },
            Sessions = CultMeshTestTrust.LocalSessions,
            Connectors = new[] { connector }
        });
        var opening = mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "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://127.0.0.1:3081" },
                        new[] { "aetheria-daemon" }).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, ICultMeshVerifiedSchemaClient
    {
        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() { }
        public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
            verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon";
    }

    private sealed class DocumentConnector : ICultMeshTransportConnector
    {
        public List<DocumentClient> Clients { get; } = new();
        public bool SuppressFirstSnapshot { get; set; }
        public bool SuppressFirstSubscriptionSnapshot { get; set; }
        public bool RejectOperations { get; set; }
        public bool FailFirstOperationBeforeResponse { get; set; }
        public string ResponseSourceRuntimeId { get; set; } = "aetheria-daemon";
        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,
                rejectOperations: RejectOperations,
                failOperationBeforeResponse: FailFirstOperationBeforeResponse && Clients.Count == 0,
                responseSourceRuntimeId: ResponseSourceRuntimeId);
            Clients.Add(client);
            return Task.FromResult<ICultNetSchemaClient>(client);
        }
    }

    private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth, ICultMeshVerifiedSchemaClient
    {
        private readonly List<Action<CultNetSnapshotResponseRawMessage>> _snapshots = new();
        private readonly List<Action<CultNetOperationResponseMessage>> _operationResponses = new();
        private readonly CultNetDocumentRegistry _documents;
        private readonly string _text;
        private readonly TaskCompletionSource<Exception> _failure = new(TaskCreationOptions.RunContinuationsAsynchronously);
        private readonly bool _rejectOperations;
        private bool _failOperationBeforeResponse;
        private readonly string _responseSourceRuntimeId;
        private int _snapshotsToSuppress;
        public DocumentClient(
            string text,
            int snapshotsToSuppress = 0,
            bool rejectOperations = false,
            bool failOperationBeforeResponse = false,
            string responseSourceRuntimeId = "aetheria-daemon")
        {
            _text = text;
            _snapshotsToSuppress = snapshotsToSuppress;
            _rejectOperations = rejectOperations;
            _failOperationBeforeResponse = failOperationBeforeResponse;
            _responseSourceRuntimeId = responseSourceRuntimeId;
            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 List<CultNetOperationRequestMessage> Operations { 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 CultNetOperationRequestMessage operation)
            {
                operation.TargetRuntimeId.Should().Be("aetheria-daemon");
                Operations.Add(operation);
                if (_failOperationBeforeResponse)
                {
                    _failOperationBeforeResponse = false;
                    Fail(new InvalidOperationException("path lost before operation receipt"));
                    return;
                }
                if (_rejectOperations)
                {
                    var failure = new CultNetOperationFailure
                    {
                        Code = "request-schema-mismatch",
                        Message = "CultNet operation expected payload schema tests.counter.increment_request.v1."
                    };
                    var rejected = new CultNetOperationResponseMessage
                    {
                        MessageId = operation.MessageId,
                        ServiceId = operation.ServiceId,
                        Operation = operation.Operation,
                        Status = "invalid",
                        PayloadSchema = CultNetOperationServer.FailureSchemaId,
                        Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
                            failure,
                            CultNetSchemaMessageSerialization.Options)),
                        Diagnostics = new[] { failure.Message },
                        SourceRuntimeId = _responseSourceRuntimeId
                    };
                    foreach (var handler in _operationResponses.ToArray()) handler(rejected);
                    return;
                }
                var decodedRequest = MessagePackSerializer.Deserialize<ClientOperationRequest>(
                    Convert.FromBase64String(operation.Payload),
                    CultNetSchemaMessageSerialization.Options);
                var receipt = new ClientOperationReceipt
                {
                    ReceiptId = "receipt:" + operation.MessageId,
                    Count = decodedRequest.Amount
                };
                var operationResponse = new CultNetOperationResponseMessage
                {
                    MessageId = operation.MessageId,
                    ServiceId = operation.ServiceId,
                    Operation = operation.Operation,
                    Status = "accepted",
                    PayloadSchema = "tests.counter.increment_receipt.v1",
                    Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
                        receipt,
                        CultNetSchemaMessageSerialization.Options)),
                    SourceRuntimeId = _responseSourceRuntimeId
                };
                foreach (var handler in _operationResponses.ToArray()) handler(operationResponse);
                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));
            if (typeof(T) == typeof(CultNetOperationResponseMessage))
                _operationResponses.Add(response => callback((T)(object)response));
        }
        public void Dispose() { }
        public void Fail(Exception error) => _failure.TrySetResult(error);
        public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
            verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon";
    }

    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;
    }

    [MessagePackObject]
    public sealed class ClientOperationRequest
    {
        [Key("amount")] public int Amount { get; set; }
    }

    [MessagePackObject]
    public sealed class ClientOperationReceipt
    {
        [Key("receiptId")] public string ReceiptId { get; set; } = string.Empty;
        [Key("count")] public int Count { get; set; }
    }
}