lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs

using System.Collections.Immutable;
using System.Diagnostics.CodeAnalysis;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.TestKit;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Cluster;
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.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;

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

public class CentralCommunicationActorTests : ControlPlaneActorTestBase
{
    private const string DeploymentsTopic = "deployments";

    /// <summary>Records every client the actor asks for, and hands back a probe in its place.</summary>
    private sealed class RecordingClientFactory : IMeshClusterClientFactory
    {
        private readonly Func<TestProbe> _newProbe;

        public RecordingClientFactory(Func<TestProbe> newProbe) => _newProbe = newProbe;

        public List<ImmutableHashSet<ActorPath>> Calls { get; } = [];

        public List<string> ClusterIds { get; } = [];

        public List<TestProbe> Probes { get; } = [];

        /// <summary>Probe created for a given cluster id, for per-cluster fan-out assertions.</summary>
        public Dictionary<string, TestProbe> ProbeByCluster { get; } = [];

        public IActorRef Create(ActorSystem system, string clusterId, ImmutableHashSet<ActorPath> contacts)
        {
            Calls.Add(contacts);
            ClusterIds.Add(clusterId);
            var probe = _newProbe();
            Probes.Add(probe);
            ProbeByCluster[clusterId] = probe;
            return probe.Ref;
        }
    }

    private static MeshTransportOptions ClusterClientMode(int refreshSeconds = 60) => new()
    {
        Mode = MeshTransportOptions.ModeClusterClient,
        CentralContactPoints = ["akka.tcp://otopcua@central-1:4053"],
        ContactRefreshSeconds = refreshSeconds,
    };

    private static MeshTransportOptions GrpcMode(int refreshSeconds = 60) => new()
    {
        Mode = MeshTransportOptions.ModeGrpc,
        ContactRefreshSeconds = refreshSeconds,
        DispatchTimeoutSeconds = 10,
    };

    private static void SeedCluster(
        IDbContextFactory<OtOpcUaConfigDbContext> factory,
        string clusterId,
        params (string NodeId, string Host, int AkkaPort, bool Enabled, bool Maintenance)[] nodes)
    {
        using var db = factory.CreateDbContext();
        db.ServerClusters.Add(new ServerCluster
        {
            ClusterId = clusterId,
            Name = clusterId,
            Enterprise = "ZB",
            Site = "Site",
            // Warm/Hot require NodeCount = 2 (CK_ServerCluster_RedundancyMode_NodeCount); these
            // fixtures only need the row to exist, so use the standalone shape.
            RedundancyMode = RedundancyMode.None,
            NodeCount = 1,
            CreatedBy = "test",
        });
        foreach (var n in nodes)
        {
            db.ClusterNodes.Add(new ClusterNode
            {
                NodeId = n.NodeId,
                ClusterId = clusterId,
                Host = n.Host,
                AkkaPort = n.AkkaPort,
                ApplicationUri = $"urn:{n.NodeId}",
                Enabled = n.Enabled,
                MaintenanceMode = n.Maintenance,
                CreatedBy = "test",
            });
        }

        db.SaveChanges();
    }

    /// <summary>
    ///     Seeds a cluster whose nodes carry a <c>GrpcPort</c> — the gRPC command plane's shape.
    ///     Kept separate from <see cref="SeedCluster"/> so the ClusterClient facts, which have no
    ///     opinion about the port, stay exactly as they were.
    /// </summary>
    private static void SeedGrpcCluster(
        IDbContextFactory<OtOpcUaConfigDbContext> factory,
        string clusterId,
        params (string NodeId, string Host, int? GrpcPort, bool Enabled, bool Maintenance)[] nodes)
    {
        using var db = factory.CreateDbContext();
        db.ServerClusters.Add(new ServerCluster
        {
            ClusterId = clusterId,
            Name = clusterId,
            Enterprise = "ZB",
            Site = "Site",
            RedundancyMode = RedundancyMode.None,
            NodeCount = 1,
            CreatedBy = "test",
        });
        foreach (var n in nodes)
        {
            db.ClusterNodes.Add(new ClusterNode
            {
                NodeId = n.NodeId,
                ClusterId = clusterId,
                Host = n.Host,
                AkkaPort = 4053,
                GrpcPort = n.GrpcPort,
                ApplicationUri = $"urn:{n.NodeId}",
                Enabled = n.Enabled,
                MaintenanceMode = n.Maintenance,
                CreatedBy = "test",
            });
        }

        db.SaveChanges();
    }

    private IActorRef Spawn(
        IDbContextFactory<OtOpcUaConfigDbContext> factory,
        MeshTransportOptions options,
        IMeshClusterClientFactory clientFactory,
        IActorRef? coordinator = null) =>
        Sys.ActorOf(CentralCommunicationActor.Props(factory, options, clientFactory, () => coordinator));

    [Fact]
    public void Dps_mode_publishes_on_the_carried_topic_and_creates_no_client()
    {
        var factory = NewInMemoryDbFactory();
        var clients = new RecordingClientFactory(() => CreateTestProbe());
        var mediatorProbe = CreateTestProbe();
        // Subscribe on the probe's behalf: SubscribeAck goes to the SENDER, so telling the mediator
        // without an explicit sender would ack to TestActor and this wait would hang.
        DistributedPubSub.Get(Sys).Mediator.Tell(
            new Subscribe(DeploymentsTopic, mediatorProbe.Ref), mediatorProbe.Ref);
        mediatorProbe.ExpectMsg<SubscribeAck>(TimeSpan.FromSeconds(5));

        var actor = Spawn(factory, new MeshTransportOptions(), clients);
        var dispatch = new DispatchDeployment(
            new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid()));

        actor.Tell(new MeshCommand(DeploymentsTopic, dispatch));

        mediatorProbe.ExpectMsg<DispatchDeployment>(TimeSpan.FromSeconds(5)).ShouldBe(dispatch);
        clients.Calls.ShouldBeEmpty();
    }

    [Fact]
    public void ClusterClient_mode_sends_to_all_not_to_one()
    {
        // THE assertion of this phase. ClusterClient.Send delivers to exactly ONE registered actor;
        // today's DPS publish reaches EVERY DriverHostActor, and the node side has no ClusterId or
        // node filter to compensate. Send would deploy to a single node while every other node
        // silently kept its old configuration.
        var factory = NewInMemoryDbFactory();
        SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false));
        var clients = new RecordingClientFactory(() => CreateTestProbe());

        var actor = Spawn(factory, ClusterClientMode(), clients);
        AwaitCondition(() => clients.Probes.Count == 1, TimeSpan.FromSeconds(5));

        var dispatch = new DispatchDeployment(
            new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid()));
        actor.Tell(new MeshCommand(DeploymentsTopic, dispatch));

        var sent = clients.Probes[0].ExpectMsg<ClusterClient.SendToAll>(TimeSpan.FromSeconds(5));
        sent.Path.ShouldBe(MeshPaths.NodeCommunication);
        sent.Message.ShouldBe(dispatch);
    }

    [Fact]
    public void One_client_per_cluster_is_created_for_a_multi_cluster_fleet()
    {
        // THE Phase 6 inversion. Each application Cluster is now its own Akka mesh, and a ClusterClient
        // reaches only the mesh whose receptionist it dialled. Central must therefore build one client
        // per cluster, each dialling only that cluster's nodes.
        var factory = NewInMemoryDbFactory();
        SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false));
        SeedCluster(factory, "C2", ("host-b:4053", "host-b", 4053, true, false));
        var clients = new RecordingClientFactory(() => CreateTestProbe());

        Spawn(factory, ClusterClientMode(), clients);
        AwaitCondition(() => clients.Calls.Count == 2, TimeSpan.FromSeconds(5));

        // Give a second refresh a chance to (wrongly) create more.
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
        clients.Calls.Count.ShouldBe(2);
        clients.ClusterIds.ShouldBe(new[] { "C1", "C2" }, ignoreOrder: true);

        // Each client dials ONLY its own cluster's node.
        clients.Calls[clients.ClusterIds.IndexOf("C1")].Single().ToString().ShouldContain("host-a");
        clients.Calls[clients.ClusterIds.IndexOf("C2")].Single().ToString().ShouldContain("host-b");
    }

    [Fact]
    public void A_command_fans_out_a_send_to_all_to_every_cluster_client()
    {
        // The other half of the inversion: one MeshCommand must reach EVERY cluster's mesh, so central
        // fans SendToAll across all per-cluster clients. A single Send/one-client shape would leave
        // every cluster but one silently on its old configuration.
        var factory = NewInMemoryDbFactory();
        SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false));
        SeedCluster(factory, "C2", ("host-b:4053", "host-b", 4053, true, false));
        var clients = new RecordingClientFactory(() => CreateTestProbe());

        var actor = Spawn(factory, ClusterClientMode(), clients);
        AwaitCondition(() => clients.Probes.Count == 2, TimeSpan.FromSeconds(5));

        var dispatch = new DispatchDeployment(
            new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid()));
        actor.Tell(new MeshCommand(DeploymentsTopic, dispatch));

        foreach (var (clusterId, probe) in clients.ProbeByCluster)
        {
            var sent = probe.ExpectMsg<ClusterClient.SendToAll>(
                TimeSpan.FromSeconds(5), $"cluster {clusterId} must receive the fanned SendToAll");
            sent.Path.ShouldBe(MeshPaths.NodeCommunication);
            sent.Message.ShouldBe(dispatch);
        }
    }

    [Fact]
    public void Disabled_and_maintenance_nodes_are_not_dialled()
    {
        var factory = NewInMemoryDbFactory();
        SeedCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4053, true, false),
            ("host-b:4053", "host-b", 4053, false, false),
            ("host-c:4053", "host-c", 4053, true, true));
        var clients = new RecordingClientFactory(() => CreateTestProbe());

        Spawn(factory, ClusterClientMode(), clients);
        AwaitCondition(() => clients.Calls.Count == 1, TimeSpan.FromSeconds(5));

        clients.Calls[0].Count.ShouldBe(1);
        clients.Calls[0].Single().ToString().ShouldContain("host-a");
    }

    [Fact]
    public void A_malformed_node_row_does_not_abort_the_refresh()
    {
        // The bad row is ordered FIRST on purpose: the failure this guards against is an exception
        // that aborts the loop and leaves the contact set half-built.
        var factory = NewInMemoryDbFactory();
        SeedCluster(
            factory,
            "C1",
            ("bad", "not a host!!", 4053, true, false),
            ("host-a:4053", "host-a", 4053, true, false));
        var clients = new RecordingClientFactory(() => CreateTestProbe());

        Spawn(factory, ClusterClientMode(), clients);
        AwaitCondition(() => clients.Calls.Count == 1, TimeSpan.FromSeconds(5));

        clients.Calls[0].ShouldContain(p => p.ToString().Contains("host-a"));
    }

    [Fact]
    public void A_command_with_no_client_is_dropped_with_a_warning()
    {
        // No ClusterNode rows at all: nothing to dial, so the command must be dropped loudly rather
        // than queued. buffer-size = 0 covers the case where a client exists but cannot connect;
        // this covers the case where there is no client at all.
        var factory = NewInMemoryDbFactory();
        var clients = new RecordingClientFactory(() => CreateTestProbe());
        var actor = Spawn(factory, ClusterClientMode(), clients);

        EventFilter.Warning(contains: "dropping").ExpectOne(TimeSpan.FromSeconds(5), () =>
            actor.Tell(new MeshCommand(DeploymentsTopic, "anything")));
    }

    [Fact]
    public void ApplyAck_is_forwarded_to_the_coordinator_preserving_the_sender()
    {
        var factory = NewInMemoryDbFactory();
        var clients = new RecordingClientFactory(() => CreateTestProbe());
        var coordinator = CreateTestProbe();
        var node = CreateTestProbe();
        var actor = Spawn(factory, ClusterClientMode(), clients, coordinator.Ref);

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

        actor.Tell(ack, node.Ref);

        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).ShouldBe(ack);
        // Forward, not Tell: the coordinator must see the node, not this relay.
        coordinator.LastSender.ShouldBe(node.Ref);
    }

    [Fact]
    public void An_unresolvable_coordinator_logs_rather_than_losing_the_ack_silently()
    {
        var factory = NewInMemoryDbFactory();
        var clients = new RecordingClientFactory(() => CreateTestProbe());
        var actor = Spawn(factory, ClusterClientMode(), clients, coordinator: null);

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

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

    [Fact]
    public void A_faulted_contact_load_warns_and_leaves_the_actor_alive()
    {
        var factory = NewInMemoryDbFactory();
        var clients = new RecordingClientFactory(() => CreateTestProbe());
        var actor = Spawn(factory, ClusterClientMode(), clients);

        // Filter on a phrase unique to the DB-failure warning. EventFilter matches case-INSENSITIVELY,
        // so a looser "was NOT" also caught this actor's other warning ("the mesh ClusterClient was
        // not created") and the assertion failed with two matches instead of one.
        EventFilter.Warning(contains: "contact set was NOT refreshed").ExpectOne(
            TimeSpan.FromSeconds(5),
            () => actor.Tell(new Status.Failure(new InvalidOperationException("db down"))));

        // Still serving: a transient DB outage must not take the command boundary down with it.
        actor.Tell(new MeshCommand(DeploymentsTopic, "anything"));
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
    }

    // ------------------------------------------------------------------ Mode = Grpc
    //
    // The third transport. The fan-out must reach EXACTLY the recipient set SendToAll reaches today
    // — every enabled, non-maintenance node in every cluster — because the node side still has no
    // ClusterId or node filter; scoping happens later, inside the artifact. A narrower fan would
    // leave nodes silently on their old configuration while the deployment sealed green.

    [Fact]
    public void Grpc_mode_fans_one_command_to_every_enabled_node_across_all_clusters()
    {
        // THE assertion of this task, and the direct counterpart of
        // A_command_fans_out_a_send_to_all_to_every_cluster_client.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, true, false));
        SeedGrpcCluster(factory, "C2", ("host-c:4053", "host-c", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 3, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls.Select(c => c.Endpoint).ShouldBe(
            new[] { "http://host-a:4056", "http://host-b:4056", "http://host-c:4056" }, ignoreOrder: true);

        // Each node is dialled with ITS OWN cluster's id — that is what selects the pre-shared key,
        // so getting it wrong turns every command into a PermissionDenied at the node.
        rig.Dialer.Calls.Single(c => c.Endpoint.Contains("host-c")).ClusterId.ShouldBe("C2");
        rig.Dialer.Calls.Where(c => !c.Endpoint.Contains("host-c")).ShouldAllBe(c => c.ClusterId == "C1");
    }

    [Fact]
    public void Grpc_mode_creates_no_ClusterClient_and_publishes_nothing_on_pubsub()
    {
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var clients = new RecordingClientFactory(() => CreateTestProbe());
        var mediatorProbe = CreateTestProbe();
        DistributedPubSub.Get(Sys).Mediator.Tell(
            new Subscribe(DeploymentsTopic, mediatorProbe.Ref), mediatorProbe.Ref);
        mediatorProbe.ExpectMsg<SubscribeAck>(TimeSpan.FromSeconds(5));

        var rig = SpawnGrpcRig(factory, clientFactory: clients);
        AwaitContactsLoaded(rig);

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
        clients.Calls.ShouldBeEmpty();
        mediatorProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
    }

    [Fact]
    public void Grpc_mode_does_not_dial_disabled_or_maintenance_nodes()
    {
        // Same recipient filter as the ClusterClient contact load, asserted independently so the two
        // transports cannot drift on who counts as a node.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, false, false),
            ("host-c:4053", "host-c", 4056, true, true));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls[0].Endpoint.ShouldContain("host-a");

        // And no late arrival from the two excluded rows.
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
        rig.Dialer.Calls.Count.ShouldBe(1);
    }

    [Fact]
    public void A_node_with_no_GrpcPort_is_skipped_with_a_warning_and_the_others_are_still_dialled()
    {
        // Mirrors TelemetryNodeSource: a row with no port exposes no endpoint yet. It must cost that
        // node its commands, never the whole fan.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", null, true, false),
            ("host-b:4053", "host-b", 4056, true, false));

        GrpcRig? rig = null;
        EventFilter.Warning(contains: "has no GrpcPort").ExpectOne(
            TimeSpan.FromSeconds(5), () => rig = SpawnGrpcRig(factory));

        rig.ShouldNotBeNull();
        AwaitContactsLoaded(rig);
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls[0].Endpoint.ShouldBe("http://host-b:4056");
    }

    [Fact]
    public void A_node_that_regains_then_loses_its_GrpcPort_warns_again()
    {
        // Warn-once is per continuous skip run, not per process. The suppression exists so a
        // permanently portless row does not repeat every minute; carried across a repair it would
        // silence the SECOND time a node loses its endpoint — the occurrence an operator most needs
        // to see, because it is a regression rather than a known gap.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", null, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        SetGrpcPort(factory, "host-a:4053", 4056);
        rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts());
        AwaitCondition(
            () => rig.Psk.KnownClusterPublications.Count >= 2, TimeSpan.FromSeconds(5));

        SetGrpcPort(factory, "host-a:4053", null);
        EventFilter.Warning(contains: "has no GrpcPort").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts()));
    }

    [Fact]
    public void One_failing_node_does_not_stop_the_fan_and_is_logged_by_node_id()
    {
        // Fire-and-forget: a failure is logged and DROPPED. No retry, no queue, and above all no
        // abort of the remaining nodes — the deploy path's apply deadline is the backstop.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, true, false));
        var dialer = new RecordingDialer
        {
            FailEndpointsContaining = "host-a",
            Fault = new Grpc.Core.RpcException(
                new Grpc.Core.Status(Grpc.Core.StatusCode.Unavailable, "node down")),
        };
        var rig = SpawnGrpcRig(factory, dialer);
        AwaitContactsLoaded(rig);

        // The node id, not the endpoint: an operator reading the log is looking at ClusterNode rows.
        EventFilter.Warning(contains: "host-a:4053").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));

        // Both were attempted; the healthy one is unaffected by its neighbour's failure.
        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));

        // Still serving after a per-node failure — no supervision restart, no state loss.
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
        AwaitCondition(() => rig.Dialer.Calls.Count == 4, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void A_psk_miss_costs_only_that_node_and_never_becomes_a_fan_level_abort()
    {
        // The dialer refuses a keyless cluster locally, so it arrives here as an ordinary per-node
        // failure. The fan-level assertion is that the other cluster is still dialled.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        SeedGrpcCluster(factory, "C2", ("host-b:4053", "host-b", 4056, true, false));
        var dialer = new RecordingDialer
        {
            FailClusters = ["C2"],
            Fault = new Grpc.Core.RpcException(
                new Grpc.Core.Status(Grpc.Core.StatusCode.Unauthenticated, "no key")),
        };
        var rig = SpawnGrpcRig(factory, dialer);
        AwaitContactsLoaded(rig);

        EventFilter.Warning(contains: "Unauthenticated").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));

        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void A_command_with_no_grpc_target_is_dropped_with_a_warning()
    {
        var factory = NewInMemoryDbFactory();
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        EventFilter.Warning(contains: "dropping").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));

        rig.Dialer.Calls.ShouldBeEmpty();
    }

    [Fact]
    public void A_payload_that_is_not_a_mesh_command_record_is_dropped_with_an_error()
    {
        // The mapper refuses to encode an unknown payload rather than shipping something the node
        // cannot decode. That refusal must cost the command, not the actor.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        EventFilter.Error(contains: "Cannot encode").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, "not a command record")));

        rig.Dialer.Calls.ShouldBeEmpty();

        // Still serving.
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void The_envelope_carries_the_commands_own_correlation_and_the_whole_fan_shares_it()
    {
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        var correlationId = new CorrelationId(Guid.NewGuid());
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, new DispatchDeployment(
            new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), correlationId)));

        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls.ShouldAllBe(c => c.Envelope.CorrelationId == correlationId.ToString());
        // One envelope object, encoded once and fanned — not re-encoded per node.
        rig.Dialer.Calls.Select(c => c.Envelope).Distinct().Count().ShouldBe(1);
    }

    [Fact]
    public void An_alarm_command_gets_one_minted_transport_correlation_shared_by_the_whole_fan()
    {
        // AlarmCommand is the one record with no correlation member of its own, so the envelope's is
        // the only handle an operator has on it. Minting per fan rather than per node is what makes
        // "this alarm command" one traceable event instead of N unrelated ones.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Actor.Tell(new MeshCommand(
            "alarms", new AlarmCommand("alarm-1", "Acknowledge", "operator", null, null)));

        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls.Select(c => c.Envelope.CorrelationId).Distinct().Count().ShouldBe(1);
        rig.Dialer.Calls[0].Envelope.CorrelationId.ShouldNotBeNullOrWhiteSpace();
        rig.Dialer.Calls[0].Envelope.AlarmCommand.AlarmId.ShouldBe("alarm-1");
    }

    [Fact]
    public void The_contact_refresh_publishes_the_known_cluster_set_and_drops_cached_psks()
    {
        // Phase 1A security register item 1, wired: the gate is only a control if something keeps it
        // current, and the refresh is the moment central's picture of the fleet is up to date. The
        // InvalidateAll on the same tick is what makes a rotated key take effect within one refresh
        // instead of requiring a central restart.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        SeedGrpcCluster(factory, "C2", ("host-b:4053", "host-b", null, true, false));
        var rig = SpawnGrpcRig(factory);

        AwaitContactsLoaded(rig);

        // C2 is included even though its only node has no GrpcPort: the gate is about which clusters
        // central KNOWS, not which it can currently dial. Excluding it would deny that cluster's acks.
        rig.Psk.KnownClusterPublications[0].ShouldBe(new[] { "C1", "C2" }, ignoreOrder: true);
        rig.Psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(1);
    }

    [Fact]
    public void A_slow_dial_does_not_park_the_actors_mailbox()
    {
        // The dialer's first act is a PSK lookup, which blocks on the secret store — bounded, but
        // seconds, and repeated after every refresh because the refresh drops the cache. An async
        // method runs synchronously to its first await, so dialing straight from the receive would
        // stall this actor for that long PER NODE, and deployment acks queue behind that mailbox.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var coordinator = CreateTestProbe();
        var dialer = new RecordingDialer { BlockBeforeFirstAwait = true };
        var rig = SpawnGrpcRig(factory, dialer, coordinator: coordinator.Ref);
        AwaitContactsLoaded(rig);

        try
        {
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

            // The ack must be handled while the dial is still parked.
            var ack = new ApplyAck(
                new DeploymentId(Guid.NewGuid()),
                new NodeId("host-a:4053"),
                ApplyAckOutcome.Applied,
                null,
                new CorrelationId(Guid.NewGuid()));
            rig.Actor.Tell(ack);

            coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).ShouldBe(ack);
        }
        finally
        {
            dialer.Release();
        }
    }

    [Fact]
    public void A_node_added_after_start_joins_the_fan_on_the_next_refresh()
    {
        // The refresh is not decoration: operators add and retire nodes, and central must pick them
        // up without a restart. Under Grpc that is the ONLY thing keeping the dispatch set current —
        // there is no receptionist to notice a new registration on its own.
        //
        // Driven by telling the actor its own RefreshContacts tick rather than by shortening
        // ContactRefreshSeconds: a one-second timer would run for the whole fact, adding DB load to
        // every test sharing the runner, and would make this assertion depend on wall-clock timing.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        using (var db = factory.CreateDbContext())
        {
            db.ClusterNodes.Add(new ClusterNode
            {
                NodeId = "host-b:4053",
                ClusterId = "C1",
                Host = "host-b",
                AkkaPort = 4053,
                GrpcPort = 4056,
                ApplicationUri = "urn:host-b",
                Enabled = true,
                CreatedBy = "test",
            });
            db.SaveChanges();
        }

        rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts());
        AwaitCondition(
            () => rig.Psk.KnownClusterPublications.Count >= 2, TimeSpan.FromSeconds(5));

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls.Select(c => c.Endpoint).ShouldBe(
            new[] { "http://host-a:4056", "http://host-b:4056" }, ignoreOrder: true);

        // Once PER REFRESH, not once at startup. A key can rotate without any membership change, so
        // an invalidation that only fired when the cluster set moved would serve the pre-rotation
        // key forever and a rotation would need a central restart.
        rig.Psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(2);
    }

    [Theory]
    [InlineData(MeshTransportOptions.ModeDps)]
    [InlineData(MeshTransportOptions.ModeClusterClient)]
    [InlineData(MeshTransportOptions.ModeGrpc)]
    public void The_known_cluster_gate_is_published_in_every_transport_mode(string mode)
    {
        // Replaces an earlier fact asserting the OPPOSITE, whose premise did not survive 1B: the
        // gate feeds CentralCommandAuthInterceptor, which is registered on every admin node, and the
        // ack service it protects is mapped by LISTENER PORT — neither consults MeshTransport:Mode.
        // Gating the publication on Grpc meant a dark-switch rollback to ClusterClient or Dps left
        // that endpoint serving with a permanently inactive gate, silently restoring both the timing
        // oracle and the pre-authentication store/audit flood. The set is central's cluster
        // inventory, not transport state.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var psk = new RecordingPskProvider();

        Sys.ActorOf(CentralCommunicationActor.Props(
            factory,
            new MeshTransportOptions { Mode = mode, ContactRefreshSeconds = 60 },
            new RecordingClientFactory(() => CreateTestProbe()),
            () => null,
            new RecordingDialer(),
            psk));

        AwaitCondition(
            () => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5));
        psk.KnownClusterPublications[0].ShouldBe(new[] { "C1" });
        psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(1);
    }

    [Fact]
    public void The_gate_set_includes_a_cluster_the_fan_set_excludes()
    {
        // The two sets answer different questions. The FAN asks "who does central talk to" — enabled,
        // non-maintenance, ported. The GATE asks "is this a cluster central has ever heard of", and
        // must say yes for a cluster whose every node is disabled or in maintenance: such a cluster
        // can still send a deployment ack, and refusing it would quietly turn maintenance mode into
        // an authentication switch.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        SeedGrpcCluster(factory, "C-DISABLED", ("host-b:4053", "host-b", 4056, false, false));
        SeedGrpcCluster(factory, "C-MAINT", ("host-c:4053", "host-c", 4056, true, true));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Psk.KnownClusterPublications[0].ShouldBe(
            new[] { "C1", "C-DISABLED", "C-MAINT" }, ignoreOrder: true);

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
        rig.Dialer.Calls.Count.ShouldBe(1);
        rig.Dialer.Calls[0].ClusterId.ShouldBe("C1");
    }

    [Fact]
    public void The_gate_is_published_after_the_dispatch_set_is_adopted_and_before_the_cache_is_dropped()
    {
        // Two orderings in one handler, both load-bearing. SetKnownClusters must precede
        // InvalidateAll, or the first lookup after a refresh re-reads the store under the PREVIOUS
        // membership — briefly resolving a cluster central has just retired. And the dispatch set is
        // adopted first so a command processed immediately after the publication already fans over
        // the current targets.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Psk.Sequence.ShouldBe(new[] { "SetKnownClusters", "InvalidateAll" });

        // Targets-before-gate, observed the only way it can be from outside the actor: the very next
        // message finds the dispatch set already populated.
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void Grpc_mode_with_no_dialer_wired_drops_the_command_with_an_error()
    {
        // Mis-wiring, not misconfiguration: Mode=Grpc validates fine but the DI graph did not supply
        // a dialer. Silence here would be a fleet-wide command outage with no symptom until a deploy
        // stalled at its apply deadline two minutes later.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var actor = Sys.ActorOf(CentralCommunicationActor.Props(
            factory,
            GrpcMode(),
            new RecordingClientFactory(() => CreateTestProbe()),
            () => null,
            dialer: null,
            pskProvider: null));

        // Filtered on the DISPATCH message's exact phrase: PreStart logs its own error about the same
        // mis-wiring, and a looser filter races it into a second match.
        EventFilter.Error(contains: "no command dialer is wired").ExpectOne(
            TimeSpan.FromSeconds(5),
            () => actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));
    }

    private static void SetGrpcPort(
        IDbContextFactory<OtOpcUaConfigDbContext> factory, string nodeId, int? grpcPort)
    {
        using var db = factory.CreateDbContext();
        var node = db.ClusterNodes.Single(n => n.NodeId == nodeId);
        node.GrpcPort = grpcPort;
        db.SaveChanges();
    }

    private static DispatchDeployment NewDispatch() => new(
        new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid()));

    /// <summary>A spawned gRPC-mode actor plus the two seams the tests observe it through.</summary>
    private sealed record GrpcRig(IActorRef Actor, RecordingDialer Dialer, RecordingPskProvider Psk);

    private GrpcRig SpawnGrpcRig(
        IDbContextFactory<OtOpcUaConfigDbContext> factory,
        RecordingDialer? dialer = null,
        IMeshClusterClientFactory? clientFactory = null,
        IActorRef? coordinator = null)
    {
        dialer ??= new RecordingDialer();
        var psk = new RecordingPskProvider();
        var actor = Sys.ActorOf(CentralCommunicationActor.Props(
            factory,
            GrpcMode(),
            clientFactory ?? new RecordingClientFactory(() => CreateTestProbe()),
            () => coordinator,
            dialer,
            psk));
        return new GrpcRig(actor, dialer, psk);
    }

    /// <summary>
    ///     Waits until the actor has applied a contact load. The known-cluster publication is the
    ///     signal because it happens in the same handler, immediately after the target set is
    ///     adopted — and unlike the target set itself it is observable from outside the actor. A
    ///     command told before this point would be dropped as "no targets" and never retried, which
    ///     is correct behaviour and a flaky test.
    /// </summary>
    private void AwaitContactsLoaded(GrpcRig rig) =>
        AwaitCondition(
            () => rig.Psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5));

    /// <summary>Records every dial and answers however the test asks; never opens a socket.</summary>
    private sealed class RecordingDialer : IMeshCommandDialer
    {
        private readonly object _gate = new();
        private readonly List<Dial> _calls = [];
        private readonly ManualResetEventSlim _released = new(false);

        /// <summary>
        ///     When true, every dial blocks its calling thread synchronously until
        ///     <see cref="Release"/>. Models the dialer's pre-await PSK lookup.
        /// </summary>
        public bool BlockBeforeFirstAwait { get; set; }

        /// <summary>Unblocks every dial parked by <see cref="BlockBeforeFirstAwait"/>.</summary>
        public void Release() => _released.Set();

        /// <summary>Every dial made. Copied under the lock — the fan runs on thread-pool threads.</summary>
        public IReadOnlyList<Dial> Calls
        {
            get { lock (_gate) return _calls.ToList(); }
        }

        /// <summary>Endpoints whose substring match makes the dial fault.</summary>
        public string? FailEndpointsContaining { get; set; }

        /// <summary>Cluster ids whose dials fault.</summary>
        public string[] FailClusters { get; set; } = [];

        /// <summary>
        ///     The fault raised for a matching dial. Fully qualified because this file's bare
        ///     <c>Status</c> is Akka's — an existing fact drives the actor's <c>Status.Failure</c>
        ///     handler, and importing Grpc.Core here would make that reference ambiguous.
        /// </summary>
        public Grpc.Core.RpcException? Fault { get; set; }

        public Task DispatchAsync(
            string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct)
        {
            lock (_gate) _calls.Add(new Dial(nodeEndpoint, envelope, clusterId));

            if (BlockBeforeFirstAwait)
            {
                // Blocks the CALLING thread, exactly as IMeshPskProvider.TryGetPsk does on a cold
                // cache: an async method runs synchronously to its first await, so a dial invoked
                // straight from the actor would park the mailbox here.
                _released.Wait(TimeSpan.FromSeconds(30));
            }

            var shouldFail =
                (FailEndpointsContaining is not null && nodeEndpoint.Contains(FailEndpointsContaining))
                || FailClusters.Contains(clusterId, StringComparer.OrdinalIgnoreCase);

            return shouldFail && Fault is not null
                ? Task.FromException(Fault)
                : Task.CompletedTask;
        }

        /// <summary>One recorded dial.</summary>
        public sealed record Dial(string Endpoint, CommandEnvelope Envelope, string ClusterId);
    }

    /// <summary>Records the gate and invalidation calls the contact refresh is supposed to make.</summary>
    private sealed class RecordingPskProvider : IMeshPskProvider
    {
        private readonly object _gate = new();
        private readonly List<IReadOnlyList<string>> _publications = [];
        private readonly List<string> _sequence = [];
        private int _invalidateAllCount;

        /// <summary>Names of the mutating calls made, in the order they arrived.</summary>
        public IReadOnlyList<string> Sequence
        {
            get { lock (_gate) return _sequence.ToList(); }
        }

        /// <summary>Every known-cluster set published, in order.</summary>
        public IReadOnlyList<IReadOnlyList<string>> KnownClusterPublications
        {
            get { lock (_gate) return _publications.ToList(); }
        }

        /// <summary>How many times the whole cache was dropped.</summary>
        public int InvalidateAllCount => Volatile.Read(ref _invalidateAllCount);

        public bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk)
        {
            psk = "psk";
            return true;
        }

        public bool Matches(string clusterId, string presented) => true;

        public void SetKnownClusters(IReadOnlyCollection<string> clusterIds)
        {
            lock (_gate)
            {
                _publications.Add(clusterIds.ToList());
                _sequence.Add(nameof(SetKnownClusters));
            }
        }

        public void Invalidate(string clusterId) { }

        public void InvalidateAll()
        {
            Interlocked.Increment(ref _invalidateAllCount);
            lock (_gate) _sequence.Add(nameof(InvalidateAll));
        }
    }
}