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

using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.Communication;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

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

public class NodeCommunicationActorTests : RuntimeActorTestBase
{
    private static DispatchDeployment NewDispatch() => new(
        new DeploymentId(Guid.NewGuid()),
        new RevisionHash("rev-1"),
        new CorrelationId(Guid.NewGuid()));

    [Fact]
    public void DispatchDeployment_from_central_lands_on_the_node_local_event_stream()
    {
        var subscriber = CreateTestProbe();
        Sys.EventStream.Subscribe(subscriber.Ref, typeof(DispatchDeployment));
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));

        var dispatch = NewDispatch();
        actor.Tell(dispatch);

        subscriber.ExpectMsg<DispatchDeployment>(TimeSpan.FromSeconds(5)).ShouldBe(dispatch);
    }

    [Fact]
    public void A_command_is_published_exactly_once()
    {
        // The reason inbound commands do NOT go back onto their DPS topic: central SendToAll-s to
        // every node's comm actor, so a mesh-wide republish would deliver N copies to every
        // subscriber. One publish per inbound command is the whole point.
        var subscriber = CreateTestProbe();
        Sys.EventStream.Subscribe(subscriber.Ref, typeof(DispatchDeployment));
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));

        actor.Tell(NewDispatch());

        subscriber.ExpectMsg<DispatchDeployment>(TimeSpan.FromSeconds(5));
        subscriber.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
    }

    [Fact]
    public void RestartDriver_from_central_lands_on_the_event_stream()
    {
        var subscriber = CreateTestProbe();
        Sys.EventStream.Subscribe(subscriber.Ref, typeof(RestartDriver));
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));

        var msg = new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid());
        actor.Tell(msg);

        subscriber.ExpectMsg<RestartDriver>(TimeSpan.FromSeconds(5)).ShouldBe(msg);
    }

    [Fact]
    public void ReconnectDriver_from_central_lands_on_the_event_stream()
    {
        var subscriber = CreateTestProbe();
        Sys.EventStream.Subscribe(subscriber.Ref, typeof(ReconnectDriver));
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));

        var msg = new ReconnectDriver("C1", "driver-1", "alice", Guid.NewGuid());
        actor.Tell(msg);

        subscriber.ExpectMsg<ReconnectDriver>(TimeSpan.FromSeconds(5)).ShouldBe(msg);
    }

    [Fact]
    public void AlarmCommand_from_central_lands_on_the_event_stream()
    {
        // ScriptedAlarmHostActor is a CHILD of DriverHostActor with no registry key, so there is no
        // ref to hand this actor at wiring time — the EventStream is what makes the alarm-command
        // leg possible without a registration handshake.
        var subscriber = CreateTestProbe();
        Sys.EventStream.Subscribe(subscriber.Ref, typeof(AlarmCommand));
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));

        var msg = new AlarmCommand("alarm-1", "Acknowledge", "alice", "comment", null);
        actor.Tell(msg);

        subscriber.ExpectMsg<AlarmCommand>(TimeSpan.FromSeconds(5)).ShouldBe(msg);
    }

    [Fact]
    public void ApplyAck_goes_to_central_as_a_Send_not_a_SendToAll()
    {
        // Send, not SendToAll: the deploy coordinator is one singleton behind the central comm
        // actor's proxy. SendToAll would deliver one ack per central node and the coordinator would
        // count this node twice.
        var client = CreateTestProbe();
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref));

        var ack = new ApplyAck(
            new DeploymentId(Guid.NewGuid()),
            new NodeId("host-a:4053"),
            ApplyAckOutcome.Applied,
            null,
            new CorrelationId(Guid.NewGuid()));
        actor.Tell(ack);

        var sent = client.ExpectMsg<ClusterClient.Send>(TimeSpan.FromSeconds(5));
        sent.Path.ShouldBe(MeshPaths.CentralCommunication);
        sent.Message.ShouldBe(ack);
    }

    [Fact]
    public void An_ack_with_no_central_client_is_dropped_with_a_warning()
    {
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));

        var ack = new ApplyAck(
            new DeploymentId(Guid.NewGuid()),
            new NodeId("host-a:4053"),
            ApplyAckOutcome.Applied,
            null,
            new CorrelationId(Guid.NewGuid()));

        EventFilter.Warning(contains: "dropping ApplyAck").ExpectOne(
            TimeSpan.FromSeconds(5), () => actor.Tell(ack));
    }

    // ---- Mode=Grpc (ClusterClient→gRPC migration Task 1B.2) -------------------------------------

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

    [Fact]
    public void Under_Grpc_the_ack_goes_to_the_sender_exactly_once()
    {
        // The gRPC counterpart of the ClusterClient.Send fact above, and it carries the same
        // single-count obligation: one post per node per dispatch, because both central endpoints
        // route to the SAME coordinator singleton proxy. Where the ack goes among the endpoints is the
        // sender's business (GrpcApplyAckSenderTests); what this pins is that the actor hands it over
        // once and does not also fan it out itself.
        var sender = new RecordingAckSender();
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null, ackSender: sender));

        var ack = NewAck();
        actor.Tell(ack);

        AwaitAssert(() => sender.Sent.Count.ShouldBe(1), TimeSpan.FromSeconds(5));
        sender.Sent[0].ShouldBe(ack);
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
        sender.Sent.Count.ShouldBe(1);
    }

    [Fact]
    public void A_sender_wins_over_a_cluster_client_and_the_ack_is_not_sent_twice()
    {
        // Defence in depth. The registration arm builds exactly one transport — under Grpc no
        // ClusterClient is created at all — but if both were ever present, posting over both would
        // deliver two acks for one apply and the coordinator would count this node twice.
        var client = CreateTestProbe();
        var sender = new RecordingAckSender();
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref, sender));

        actor.Tell(NewAck());

        AwaitAssert(() => sender.Sent.Count.ShouldBe(1), TimeSpan.FromSeconds(5));
        client.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
    }

    [Fact]
    public void The_ClusterClient_branch_never_touches_the_sender()
    {
        // Null-safe by construction: under Dps and ClusterClient the sender is not built at all, so the
        // ApplyAck path must not depend on one existing.
        var client = CreateTestProbe();
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref));

        var ack = NewAck();
        actor.Tell(ack);

        client.ExpectMsg<ClusterClient.Send>(TimeSpan.FromSeconds(5)).Message.ShouldBe(ack);
    }

    [Fact]
    public void Inbound_commands_are_unaffected_by_the_presence_of_a_sender()
    {
        // The sender is outbound-only. A Grpc node still re-emits central's commands on its own
        // EventStream exactly as every other mode does.
        var subscriber = CreateTestProbe();
        Sys.EventStream.Subscribe(subscriber.Ref, typeof(DispatchDeployment));
        var sender = new RecordingAckSender();
        var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null, ackSender: sender));

        var dispatch = NewDispatch();
        actor.Tell(dispatch);

        subscriber.ExpectMsg<DispatchDeployment>(TimeSpan.FromSeconds(5)).ShouldBe(dispatch);
        sender.Sent.ShouldBeEmpty();
    }

    /// <summary>An <see cref="IApplyAckSender"/> double recording what the actor handed it.</summary>
    private sealed class RecordingAckSender : IApplyAckSender
    {
        private readonly Lock _gate = new();
        private readonly List<ApplyAck> _sent = [];

        /// <summary>Gets a snapshot of the acks handed to this sender.</summary>
        public IReadOnlyList<ApplyAck> Sent
        {
            get { lock (_gate) return [.. _sent]; }
        }

        /// <summary>Records <paramref name="ack"/>.</summary>
        /// <param name="ack">The ack the actor handed over.</param>
        /// <param name="cancellationToken">Unused.</param>
        /// <returns>A completed task — the real sender never throws either.</returns>
        public Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default)
        {
            lock (_gate) _sent.Add(ack);
            return Task.CompletedTask;
        }
    }
}