using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Fleet;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;

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

public sealed class FleetStatusBroadcasterTests : ControlPlaneActorTestBase
{
    [Fact]
    public void Self_member_up_lands_in_snapshot()
    {
        var probe = CreateTestProbe();
        Sys.ActorOf(FleetStatusBroadcaster.Props(broadcast: msg => probe.Ref.Tell(msg)));

        // The cluster self-joined in the harness — InitialStateAsEvents replays MemberUp for self.
        var snapshot = probe.ExpectMsg<FleetStatusChanged>(TimeSpan.FromSeconds(5));
        snapshot.Nodes.Count.ShouldBeGreaterThanOrEqualTo(1);
        snapshot.Nodes.Any(n => n.Health == FleetNodeHealth.Healthy).ShouldBeTrue();
    }

    [Fact]
    public void Heartbeat_updates_revision_in_next_snapshot()
    {
        var probe = CreateTestProbe();
        var actor = Sys.ActorOf(FleetStatusBroadcaster.Props(broadcast: msg => probe.Ref.Tell(msg)));

        // Drain the initial MemberUp snapshot.
        probe.ExpectMsg<FleetStatusChanged>(TimeSpan.FromSeconds(5));

        // Send a heartbeat for a node we know about (the self node from the cluster).
        var selfHost = Akka.Cluster.Cluster.Get(Sys).SelfAddress.Host!;
        var rev = RevisionHash.Parse(new string('c', 64));
        actor.Tell(new FleetStatusBroadcaster.DriverHostStatusHeartbeat(NodeId.Parse(selfHost), rev));

        // Wait for next periodic broadcast and assert the revision propagated.
        var next = probe.FishForMessage<FleetStatusChanged>(
            isMessage: m => m.Nodes.Any(n => n.CurrentRevision == rev),
            max: TimeSpan.FromSeconds(10));
        next.Nodes.Single(n => n.NodeId.Value == selfHost).CurrentRevision.ShouldBe(rev);
    }
}