lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs

using Grpc.Core;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.Communication;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Communication;

/// <summary>
///     The node-side gRPC ack sender (mesh gRPC command plane, Task 1B.2). The client is a fake,
///     so what is under test is the sender's own logic: the ordered single-pass over
///     <c>CentralGrpcEndpoints</c>, the per-call credentials, and the drop that used to be silent.
/// </summary>
/// <remarks>
///     <para>
///         <b>The load-bearing fact is that ONE endpoint is dialled on success.</b> This replaces an
///         one ack per node per dispatch, single-target at-most-once — not a
///         <c>SendToAll</c>. Both central nodes route an ack to the same
///         <c>ConfigPublishCoordinator</c> singleton proxy, whose NodeId-keyed upsert absorbs a
///         duplicate idempotently — so the single-endpoint rule is defence in depth and a bound
///         on wasted dials, not what protects the seal.
///     </para>
///     <para>
///         The failure side matters just as much: an undeliverable ack is dropped in
///         silence, and the whole point of the migration is that the same drop is now a Warning naming
///         every endpoint and the status it answered with.
///     </para>
/// </remarks>
public sealed class GrpcApplyAckSenderTests
{
    private const string Psk = "psk-that-must-never-be-logged";

    private static ApplyAck Ack() => new(
        new DeploymentId(Guid.NewGuid()),
        new NodeId("site-a-1:4053"),
        ApplyAckOutcome.Applied,
        null,
        new CorrelationId(Guid.NewGuid()));

    private static GrpcApplyAckSender Sender(
        string[] endpoints,
        Func<string, CentralCommandService.CentralCommandServiceClient> factory,
        ILogger<GrpcApplyAckSender>? log = null,
        string[]? roles = null)
        => new(
            Options.Create(new MeshTransportOptions
            {
                Mode = MeshTransportOptions.ModeGrpc,
                CentralGrpcEndpoints = endpoints,
                GrpcPsk = Psk,
                AckTimeoutSeconds = 7,
            }),
            Options.Create(new AkkaClusterOptions { Roles = roles ?? ["driver", "cluster-SITE-A"] }),
            log ?? NullLogger<GrpcApplyAckSender>.Instance,
            factory);

    [Fact]
    public async Task A_successful_first_endpoint_is_the_ONLY_one_dialled()
    {
        // The single-count semantic. Two endpoints are configured for availability, not for fan-out:
        // an ack posted to both would reach the one coordinator singleton twice.
        var consulted = new List<string>();
        var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
        var factory = (string endpoint) =>
        {
            consulted.Add(endpoint);
            return (CentralCommandService.CentralCommandServiceClient)new RecordingClient(calls);
        };

        await Sender(["http://central-1:4055", "http://central-2:4055"], factory).SendAsync(Ack());

        consulted.ShouldBe(["http://central-1:4055"]);
        calls.Count.ShouldBe(1);
    }

    [Fact]
    public async Task The_wire_message_carries_the_whole_ack()
    {
        var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
        var ack = new ApplyAck(
            new DeploymentId(Guid.NewGuid()),
            new NodeId("site-a-2:4053"),
            ApplyAckOutcome.Failed,
            "disk full",
            new CorrelationId(Guid.NewGuid()));

        await Sender(["http://central-1:4055"], _ => new RecordingClient(calls)).SendAsync(ack);

        var sent = calls.Single().Request;
        sent.DeploymentId.ShouldBe(ack.DeploymentId.ToString());
        sent.NodeId.ShouldBe("site-a-2:4053");
        sent.Outcome.ShouldBe(ApplyOutcome.Failed);
        sent.FailureReason.ShouldBe("disk full");
        sent.CorrelationId.ShouldBe(ack.CorrelationId.ToString());
    }

    [Fact]
    public async Task Every_call_carries_the_bearer_the_cluster_header_and_a_deadline()
    {
        // Central's CentralCommandAuthInterceptor grades the bearer against the key of the cluster
        // named in x-otopcua-cluster, so an ack missing either header is PermissionDenied — and the
        // deadline is what stops a dead first endpoint from stalling the ack forever.
        var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
        var before = DateTime.UtcNow;

        await Sender(["http://central-1:4055"], _ => new RecordingClient(calls)).SendAsync(Ack());

        var options = calls.Single().Options;
        options.Headers.ShouldNotBeNull();
        options.Headers.GetValue("authorization").ShouldBe($"Bearer {Psk}");
        options.Headers.GetValue("x-otopcua-cluster").ShouldBe("SITE-A");
        options.Deadline.ShouldNotBeNull();
        // Bounded from BOTH sides against the configured 7s. A lower bound of `before` alone would
        // accept any positive timeout at all — the value could be ignored entirely and the fact would
        // still pass.
        options.Deadline!.Value.ShouldBeGreaterThan(before.AddSeconds(6));
        options.Deadline!.Value.ShouldBeLessThanOrEqualTo(before.AddSeconds(8));
    }

    [Fact]
    public async Task Each_endpoint_gets_a_FRESH_deadline()
    {
        // The failure this pins is silent and is the characteristic WAN one: a central that HANGS
        // rather than refusing. With one absolute deadline hoisted over the whole walk, endpoint 1
        // burns all of AckTimeoutSeconds, and endpoint 2 is then dialled with a deadline already in
        // the past — it fails instantly as DeadlineExceeded without a packet leaving the node, so the
        // surviving central is never really tried and failover is defeated while still LOOKING like it
        // ran. Both stamps come from real UtcNow, so a delay on the failing path plus a strict `>` is
        // the whole test.
        var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
        var factory = (string endpoint) => endpoint.Contains("central-1", StringComparison.Ordinal)
            ? (CentralCommandService.CentralCommandServiceClient)new SlowThrowingClient(calls, TimeSpan.FromMilliseconds(50))
            : new RecordingClient(calls);

        await Sender(["http://central-1:4055", "http://central-2:4055"], factory).SendAsync(Ack());

        calls.Count.ShouldBe(2);
        calls[1].Options.Deadline!.Value.ShouldBeGreaterThan(calls[0].Options.Deadline!.Value);
    }

    [Fact]
    public async Task An_unavailable_first_endpoint_fails_over_to_the_second()
    {
        var consulted = new List<string>();
        var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
        var factory = (string endpoint) =>
        {
            consulted.Add(endpoint);
            return endpoint.Contains("central-1", StringComparison.Ordinal)
                ? (CentralCommandService.CentralCommandServiceClient)
                    new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down")))
                : new RecordingClient(calls);
        };

        await Sender(["http://central-1:4055", "http://central-2:4055"], factory).SendAsync(Ack());

        consulted.ShouldBe(["http://central-1:4055", "http://central-2:4055"]);
        calls.Count.ShouldBe(1);
    }

    [Fact]
    public async Task When_every_endpoint_fails_the_ack_is_dropped_with_one_naming_Warning()
    {
        // The pre-migration behaviour was a silent drop. It is still a drop — there is no queue and no
        // retry — but the operator now gets the node, the deployment, the correlation id and what each
        // endpoint actually answered, which is everything needed to tell "central is down" from
        // "central rejected this node's key".
        var log = new CapturingLogger<GrpcApplyAckSender>();
        var attempts = 0;
        var factory = (string endpoint) =>
        {
            attempts++;
            return (CentralCommandService.CentralCommandServiceClient)new ThrowingClient(
                new RpcException(new Status(
                    endpoint.Contains("central-1", StringComparison.Ordinal)
                        ? StatusCode.Unavailable
                        : StatusCode.PermissionDenied,
                    "no")));
        };
        var ack = Ack();

        await Sender(["http://central-1:4055", "http://central-2:4055"], factory, log).SendAsync(ack);

        attempts.ShouldBe(2);
        var warning = log.Entries.Single(e => e.Level == LogLevel.Warning);
        warning.Message.ShouldContain(ack.NodeId.Value!);
        warning.Message.ShouldContain(ack.DeploymentId.ToString());
        warning.Message.ShouldContain(ack.CorrelationId.ToString());
        warning.Message.ShouldContain("Unavailable");
        warning.Message.ShouldContain("PermissionDenied");
    }

    [Fact]
    public async Task A_failed_pass_never_posts_the_ack_twice()
    {
        // The counterpart of the first fact: failover walks forward only. A retry of an
        // already-delivered endpoint could double-count this node at the coordinator, which is the one
        // outcome worse than the ack being lost.
        var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
        var consulted = new List<string>();
        var factory = (string endpoint) =>
        {
            consulted.Add(endpoint);
            return endpoint.Contains("c1", StringComparison.Ordinal)
                ? (CentralCommandService.CentralCommandServiceClient)
                    new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down")))
                : new RecordingClient(calls);
        };

        await Sender(["http://c1:4055", "http://c2:4055", "http://c3:4055"], factory).SendAsync(Ack());

        // c3 is never dialled: the pass stops at the first success, and it does not go back to c1.
        consulted.ShouldBe(["http://c1:4055", "http://c2:4055"]);
        calls.Count.ShouldBe(1);
    }

    [Fact]
    public async Task The_pre_shared_key_is_never_written_to_a_log()
    {
        // The bearer travels in the call headers on every attempt; the failure log names endpoints and
        // status codes only. A per-endpoint dump of the Metadata would put the cluster's PSK in the
        // node log of every deployment that could not reach central.
        var log = new CapturingLogger<GrpcApplyAckSender>();
        var factory = (string _) => (CentralCommandService.CentralCommandServiceClient)
            new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down")));

        await Sender(["http://central-1:4055"], factory, log).SendAsync(Ack());

        log.Entries.ShouldNotBeEmpty();
        log.Entries.Any(e => e.Message.Contains(Psk, StringComparison.Ordinal)).ShouldBeFalse();
    }

    [Fact]
    public async Task An_empty_endpoint_list_is_a_logged_drop_not_a_throw()
    {
        // The validator refuses to start a Grpc driver node with no endpoints, so this is reachable
        // only through a code path that built the sender for a node the validator never graded (a test
        // harness, an admin-only graph). It must still be a visible drop rather than an exception on a
        // task the actor discards, which would vanish entirely.
        var log = new CapturingLogger<GrpcApplyAckSender>();
        var dialled = false;
        var factory = (string _) =>
        {
            dialled = true;
            return (CentralCommandService.CentralCommandServiceClient)new RecordingClient([]);
        };

        await Sender([], factory, log).SendAsync(Ack());

        dialled.ShouldBeFalse();
        log.Entries.Single(e => e.Level == LogLevel.Warning).Message
            .ShouldContain("CentralGrpcEndpoints");
    }

    [Fact]
    public async Task A_node_with_no_cluster_role_still_posts_and_is_warned_about()
    {
        // No cluster-scoped role means no x-otopcua-cluster header, and central has no key to grade the
        // bearer against — every ack will come back PermissionDenied. Posting anyway (rather than
        // refusing locally) keeps the rejection where it can be read on both sides, and the local
        // Warning explains a PermissionDenied that would otherwise look like a key mismatch.
        var log = new CapturingLogger<GrpcApplyAckSender>();
        var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();

        await Sender(["http://central-1:4055"], _ => new RecordingClient(calls), log, roles: ["driver"])
            .SendAsync(Ack());

        calls.Single().Options.Headers!.Any(h =>
            h.Key.Equals("x-otopcua-cluster", StringComparison.Ordinal)).ShouldBeFalse();
        log.Entries.Any(e => e.Level == LogLevel.Warning
                             && e.Message.Contains("cluster-", StringComparison.Ordinal)).ShouldBeTrue();
    }

    [Fact]
    public async Task An_unexpected_client_fault_is_a_logged_drop_not_an_unobserved_task_fault()
    {
        // The caller is an actor that discards the returned Task, so anything escaping here would be
        // swallowed by the runtime and rediscovered as an unobserved exception at GC — the exact
        // silent drop this transport exists to remove.
        var log = new CapturingLogger<GrpcApplyAckSender>();
        var factory = (string _) => (CentralCommandService.CentralCommandServiceClient)
            new FaultingClient(new InvalidOperationException("channel disposed"));

        await Should.NotThrowAsync(() => Sender(["http://central-1:4055"], factory, log).SendAsync(Ack()));

        log.Entries.Any(e => e.Level == LogLevel.Warning).ShouldBeTrue();
    }

    [Fact]
    public async Task Dispose_releases_the_real_channel_cache_and_is_idempotent()
    {
        // Built with NO client factory, so the production DefaultClientFactory runs and a real
        // GrpcChannel lands in the cache — otherwise Dispose is walking an empty dictionary and this
        // test proves nothing about the thing it names. The address is loopback port 1: nothing
        // listens, the connection is refused immediately, and no traffic leaves the machine.
        var log = new CapturingLogger<GrpcApplyAckSender>();
        var sender = new GrpcApplyAckSender(
            Options.Create(new MeshTransportOptions
            {
                Mode = MeshTransportOptions.ModeGrpc,
                CentralGrpcEndpoints = ["http://127.0.0.1:1"],
                GrpcPsk = Psk,
                AckTimeoutSeconds = 5,
            }),
            Options.Create(new AkkaClusterOptions { Roles = ["driver", "cluster-SITE-A"] }),
            log);

        await sender.SendAsync(Ack());

        // The post must have failed (nothing is listening) — which is what put a channel in the cache.
        log.Entries.Any(e => e.Level == LogLevel.Warning).ShouldBeTrue();

        sender.Dispose();
        Should.NotThrow(sender.Dispose);
    }

    // ---- fakes --------------------------------------------------------------------------------

    private static AsyncUnaryCall<ReportAck> Ok() => new(
        Task.FromResult(new ReportAck()),
        Task.FromResult(new Metadata()),
        () => Status.DefaultSuccess,
        () => [],
        () => { });

    /// <summary>A generated client double recording the request and the call options it was given.</summary>
    private sealed class RecordingClient(List<(ApplyAckMessage Request, CallOptions Options)> calls)
        : CentralCommandService.CentralCommandServiceClient
    {
        public override AsyncUnaryCall<ReportAck> ReportApplyAsync(ApplyAckMessage request, CallOptions options)
        {
            calls.Add((request, options));
            return Ok();
        }
    }

    /// <summary>A generated client double whose call throws — an endpoint that is down or refuses.</summary>
    private sealed class ThrowingClient(RpcException toThrow) : CentralCommandService.CentralCommandServiceClient
    {
        public override AsyncUnaryCall<ReportAck> ReportApplyAsync(ApplyAckMessage request, CallOptions options)
            => throw toThrow;
    }

    /// <summary>
    ///     A generated client double modelling a HUNG endpoint: it accepts the call, records it, then
    ///     takes real wall-clock time before failing — the shape a shared deadline cannot survive.
    /// </summary>
    private sealed class SlowThrowingClient(
        List<(ApplyAckMessage Request, CallOptions Options)> calls, TimeSpan delay)
        : CentralCommandService.CentralCommandServiceClient
    {
        public override AsyncUnaryCall<ReportAck> ReportApplyAsync(ApplyAckMessage request, CallOptions options)
        {
            calls.Add((request, options));
            return new AsyncUnaryCall<ReportAck>(
                HangThenFail(delay),
                Task.FromResult(new Metadata()),
                () => new Status(StatusCode.DeadlineExceeded, "hung"),
                () => [],
                () => { });
        }

        private static async Task<ReportAck> HangThenFail(TimeSpan delay)
        {
            await Task.Delay(delay);
            throw new RpcException(new Status(StatusCode.DeadlineExceeded, "hung"));
        }
    }

    /// <summary>A generated client double throwing something that is NOT an <see cref="RpcException"/>.</summary>
    private sealed class FaultingClient(Exception toThrow) : CentralCommandService.CentralCommandServiceClient
    {
        public override AsyncUnaryCall<ReportAck> ReportApplyAsync(ApplyAckMessage request, CallOptions options)
            => throw toThrow;
    }

    /// <summary>Test logger that captures all log entries.</summary>
    /// <typeparam name="T">The category type.</typeparam>
    private sealed class CapturingLogger<T> : ILogger<T>
    {
        /// <summary>Gets the captured log entries.</summary>
        public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = [];

        /// <summary>Begins a scope (stub implementation).</summary>
        /// <typeparam name="TState">The type of the scope state.</typeparam>
        /// <param name="state">The scope state.</param>
        /// <returns>A no-op scope.</returns>
        public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;

        /// <summary>Checks if logging is enabled (always true).</summary>
        /// <param name="logLevel">The log level to check.</param>
        /// <returns><c>true</c>.</returns>
        public bool IsEnabled(LogLevel logLevel) => true;

        /// <summary>Logs an entry and captures it.</summary>
        /// <typeparam name="TState">The type of the log state.</typeparam>
        /// <param name="logLevel">The log level.</param>
        /// <param name="eventId">The event ID.</param>
        /// <param name="state">The log state.</param>
        /// <param name="exception">The exception, if any.</param>
        /// <param name="formatter">The log message formatter.</param>
        public void Log<TState>(
            LogLevel logLevel, EventId eventId, TState state, Exception? exception,
            Func<TState, Exception?, string> formatter)
        {
            Entries.Add((logLevel, formatter(state, exception), exception));
        }

        private sealed class NullScope : IDisposable
        {
            /// <summary>Gets the singleton instance.</summary>
            public static NullScope Instance { get; } = new();

            /// <summary>No-op.</summary>
            public void Dispose() { }
        }
    }
}