lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialSupervisorTests.cs

using System.Collections.Concurrent;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
using ZB.MOM.WW.OtOpcUa.AdminUI.Telemetry;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;

namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Telemetry;

/// <summary>
///     Unit tests for <see cref="TelemetryDialSupervisor"/> — the central-side dial supervisor that
///     discovers driver nodes, keeps one reconnecting dialer each, and feeds the four AdminUI
///     in-process sinks. All seams are faked: no real DB, no real gRPC.
/// </summary>
public sealed class TelemetryDialSupervisorTests : TestKit
{
    /// <summary>Explicit AwaitCondition poll interval: Akka's default is max / 10 of the dilated budget, i.e.
    /// seconds under CI's timefactor (lmxopcua#775).</summary>
    private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(20);

    private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5);

    [Fact]
    public void Discovery_starts_one_dialer_per_target_with_distinct_correlation_ids()
    {
        var dial = new FakeDialLoop();
        var targets = new[]
        {
            new TelemetryDialTarget("node-1", "http://h1:5300"),
            new TelemetryDialTarget("node-2", "http://h2:5300"),
        };

        Spawn(dial, () => targets);

        var invocations = dial.WaitForCount(2, Timeout);
        invocations.Select(i => i.Target.NodeId).ShouldBe(new[] { "node-1", "node-2" }, ignoreOrder: true);
        invocations.ShouldContain(i => i.CorrelationId == "central-node-1-1");
        invocations.ShouldContain(i => i.CorrelationId == "central-node-2-1");
        invocations.ShouldAllBe(i => i.Ct.CanBeCanceled);
    }

    [Fact]
    public void Refresh_removes_departed_nodes_and_adds_new_ones()
    {
        var dial = new FakeDialLoop();
        var current = new List<TelemetryDialTarget>
        {
            new("node-1", "http://h1:5300"),
            new("node-2", "http://h2:5300"),
        };

        var actor = Spawn(dial, () => current.ToArray());
        dial.WaitForCount(2, Timeout);

        var node1 = dial.Invocations.Single(i => i.Target.NodeId == "node-1");

        // node-1 leaves, node-3 joins.
        current.Clear();
        current.Add(new TelemetryDialTarget("node-2", "http://h2:5300"));
        current.Add(new TelemetryDialTarget("node-3", "http://h3:5300"));
        actor.Tell(new TelemetryDialSupervisor.RefreshNodes());

        // node-1's dialer is cancelled and dropped.
        AwaitCondition(() => node1.Ct.IsCancellationRequested, Timeout, PollInterval);
        // node-3 gets a fresh dialer.
        var invocations = dial.WaitForCount(3, Timeout);
        invocations.ShouldContain(i => i.Target.NodeId == "node-3" && i.CorrelationId == "central-node-3-1");

        // node-2 is present in BOTH refreshes with an unchanged endpoint: it must NOT be restarted, or
        // its live stream would be dropped on every refresh. Exactly one dial invocation for it.
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
        dial.Invocations.Count(i => i.Target.NodeId == "node-2").ShouldBe(1);
    }

    [Fact]
    public void Refresh_redials_a_node_whose_endpoint_changed_but_not_an_unchanged_one()
    {
        var dial = new FakeDialLoop();
        var current = new List<TelemetryDialTarget>
        {
            new("node-1", "http://old-host:5300"),
            new("node-2", "http://h2:5300"),
        };

        var actor = Spawn(dial, () => current.ToArray());
        dial.WaitForCount(2, Timeout);
        var node1Original = dial.Invocations.Single(i => i.Target.NodeId == "node-1");

        // node-1 re-provisioned to a new host; node-2 unchanged.
        current.Clear();
        current.Add(new TelemetryDialTarget("node-1", "http://new-host:5300"));
        current.Add(new TelemetryDialTarget("node-2", "http://h2:5300"));
        actor.Tell(new TelemetryDialSupervisor.RefreshNodes());

        // The old node-1 stream is torn down and a fresh dialer opens at the NEW endpoint (gen 2).
        AwaitCondition(() => node1Original.Ct.IsCancellationRequested, Timeout, PollInterval);
        var invocations = dial.WaitForCount(3, Timeout);
        invocations.ShouldContain(i =>
            i.Target.NodeId == "node-1"
            && i.Target.Endpoint == "http://new-host:5300"
            && i.CorrelationId == "central-node-1-2");

        // node-2's endpoint did not change → still exactly one dial invocation (no churn).
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
        dial.Invocations.Count(i => i.Target.NodeId == "node-2").ShouldBe(1);
    }

    [Fact]
    public void Each_record_type_lands_on_its_sink()
    {
        var dial = new FakeDialLoop();
        var sinks = new Sinks();
        Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }, sinks);

        var inv = dial.WaitForCount(1, Timeout).Single();

        inv.OnMapped(Alarm("a1"));
        inv.OnMapped(Script("s1"));
        inv.OnMapped(Health("d1"));
        inv.OnMapped(Resilience("d1"));

        AwaitAssert(
            () =>
            {
                sinks.Alarms.ShouldHaveSingleItem().AlarmId.ShouldBe("a1");
                sinks.Scripts.ShouldHaveSingleItem().ScriptId.ShouldBe("s1");
                sinks.Health.ShouldHaveSingleItem().DriverInstanceId.ShouldBe("d1");
                sinks.Resilience.ShouldHaveSingleItem().DriverInstanceId.ShouldBe("d1");
            },
            Timeout);
    }

    [Fact]
    public void Stream_failure_reconnects_with_an_incremented_generation()
    {
        var dial = new FakeDialLoop();
        Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") });

        var first = dial.WaitForCount(1, Timeout).Single();
        first.CorrelationId.ShouldBe("central-node-1-1");

        // Transport failure → the client would end the stream after onError, so end the fake loop too.
        first.OnError(new InvalidOperationException("boom"));
        first.Complete();

        // First retry is immediate; generation is bumped.
        var invocations = dial.WaitForCount(2, Timeout);
        invocations[1].CorrelationId.ShouldBe("central-node-1-2");
    }

    [Fact]
    public void Late_event_from_a_superseded_generation_is_dropped()
    {
        var dial = new FakeDialLoop();
        var sinks = new Sinks();
        Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }, sinks);

        var first = dial.WaitForCount(1, Timeout).Single();
        first.OnError(new InvalidOperationException("boom"));
        first.Complete();
        dial.WaitForCount(2, Timeout); // gen-2 dialer is live.

        // A late event arriving on the OLD (gen-1) stream must be dropped, not routed.
        first.OnMapped(Alarm("stale"));

        // Give the message a chance to be (wrongly) routed, then assert it was not.
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
        sinks.Alarms.ShouldBeEmpty();
    }

    [Fact]
    public void Late_failure_from_a_superseded_generation_does_not_trigger_a_second_reconnect()
    {
        var dial = new FakeDialLoop();
        Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") });

        var first = dial.WaitForCount(1, Timeout).Single();
        first.OnError(new InvalidOperationException("boom"));
        first.Complete();
        dial.WaitForCount(2, Timeout); // gen-2 dialer.

        // A duplicate/late failure from gen-1 must be ignored (no third dial).
        first.OnError(new InvalidOperationException("late duplicate"));

        ExpectNoMsg(TimeSpan.FromMilliseconds(400));
        dial.Invocations.Count.ShouldBe(2);
    }

    [Fact]
    public void Pill_flips_true_on_first_connection_and_false_only_when_all_down()
    {
        var dial = new FakeDialLoop();
        var sinks = new Sinks();
        Spawn(
            dial,
            () => new[]
            {
                new TelemetryDialTarget("node-1", "http://h1:5300"),
                new TelemetryDialTarget("node-2", "http://h2:5300"),
            },
            sinks);

        var inv = dial.WaitForCount(2, Timeout);
        var node1 = inv.Single(i => i.Target.NodeId == "node-1");
        var node2 = inv.Single(i => i.Target.NodeId == "node-2");

        // First node connects → both broadcasters flip to connected.
        node1.OnMapped(Alarm("a1"));
        AwaitAssert(() => sinks.PillStates.ToArray().ShouldBe(new[] { true }), Timeout);

        // Second node connects → still connected, no extra transition.
        node2.OnMapped(Alarm("a2"));
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
        sinks.PillStates.ToArray().ShouldBe(new[] { true });

        // node-1 drops but node-2 is still up → pill stays true.
        node1.OnError(new InvalidOperationException("down"));
        node1.Complete();
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
        sinks.PillStates.ToArray().ShouldBe(new[] { true });

        // node-2 drops too → all down → pill flips false.
        node2.OnError(new InvalidOperationException("down"));
        node2.Complete();
        AwaitAssert(() => sinks.PillStates.ToArray().ShouldBe(new[] { true, false }), Timeout);
    }

    [Fact]
    public void PostStop_cancels_every_dialer()
    {
        var dial = new FakeDialLoop();
        var actor = Spawn(
            dial,
            () => new[]
            {
                new TelemetryDialTarget("node-1", "http://h1:5300"),
                new TelemetryDialTarget("node-2", "http://h2:5300"),
            });

        var inv = dial.WaitForCount(2, Timeout);

        Sys.Stop(actor);

        AwaitCondition(() => inv.All(i => i.Ct.IsCancellationRequested), Timeout, PollInterval);
    }

    [Fact]
    public async Task Production_node_source_skips_null_grpc_port_and_maintenance_rows()
    {
        var factory = new InMemoryFactory(Guid.NewGuid().ToString());
        await using (var db = factory.CreateDbContext())
        {
            db.ClusterNodes.AddRange(
                Node("has-port", "h1", grpcPort: 5300, enabled: true, maintenance: false),
                Node("null-port", "h2", grpcPort: null, enabled: true, maintenance: false),
                Node("maintenance", "h3", grpcPort: 5300, enabled: true, maintenance: true),
                Node("disabled", "h4", grpcPort: 5300, enabled: false, maintenance: false));
            await db.SaveChangesAsync();
        }

        var source = TelemetryNodeSource.Create(factory, NullLogger.Instance);
        var targets = await source();

        targets.ShouldHaveSingleItem();
        targets[0].NodeId.ShouldBe("has-port");
        targets[0].Endpoint.ShouldBe("http://h1:5300");
    }

    private IActorRef Spawn(
        FakeDialLoop dial,
        Func<TelemetryDialTarget[]> nodeSource,
        Sinks? sinks = null)
    {
        sinks ??= new Sinks();
        var options = new TelemetryDialOptions { ContactRefreshSeconds = 3600 };
        return Sys.ActorOf(TelemetryDialSupervisor.Props(
            () => Task.FromResult<IReadOnlyList<TelemetryDialTarget>>(nodeSource()),
            dial.Loop,
            sinks.AlarmBroadcaster,
            sinks.ScriptBroadcaster,
            sinks.HealthStore,
            sinks.ResilienceStore,
            options));
    }

    private static AlarmTransitionEvent Alarm(string id) =>
        new(id, "Area/Line/Eq", "hi", "Activated", 500, "msg", "system", DateTime.UtcNow);

    private static ScriptLogEntry Script(string id) =>
        new(id, "Info", "msg", DateTime.UtcNow, null, null, null);

    private static DriverHealthChanged Health(string id) =>
        new("C1", id, "Healthy", DateTime.UtcNow, null, 0, DateTime.UtcNow);

    private static DriverResilienceStatusChanged Resilience(string id) =>
        new(id, "host", false, 0, 0, null, DateTime.UtcNow, DateTime.UtcNow);

    private static ClusterNode Node(string id, string host, int? grpcPort, bool enabled, bool maintenance) =>
        new()
        {
            NodeId = id,
            ClusterId = "C1",
            Host = host,
            GrpcPort = grpcPort,
            Enabled = enabled,
            MaintenanceMode = maintenance,
            ApplicationUri = $"urn:{id}",
            CreatedBy = "test",
        };

    /// <summary>A captured dial-loop invocation with the callbacks the test drives by hand.</summary>
    private sealed class DialInvocation
    {
        private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);

        public required TelemetryDialTarget Target { get; init; }
        public required string CorrelationId { get; init; }
        public required Action<object> OnMapped { get; init; }
        public required Action<Exception> OnError { get; init; }
        public required CancellationToken Ct { get; init; }

        public Task Completion => _completion.Task;

        /// <summary>Ends the fake loop (as the client returns after a server-ended or post-onError stream).</summary>
        public void Complete() => _completion.TrySetResult();
    }

    /// <summary>A fake <see cref="TelemetryDialLoop"/> that records each invocation and blocks until told to end.</summary>
    private sealed class FakeDialLoop
    {
        private readonly ConcurrentQueue<DialInvocation> _invocations = new();

        public IReadOnlyList<DialInvocation> Invocations => _invocations.ToArray();

        public TelemetryDialLoop Loop => async (target, correlationId, onMapped, onError, ct) =>
        {
            var invocation = new DialInvocation
            {
                Target = target,
                CorrelationId = correlationId,
                OnMapped = onMapped,
                OnError = onError,
                Ct = ct,
            };
            _invocations.Enqueue(invocation);

            // Complete when the test ends the loop OR the supervisor cancels us.
            await using var reg = ct.Register(() => invocation.Complete());
            await invocation.Completion;
        };

        public IReadOnlyList<DialInvocation> WaitForCount(int count, TimeSpan timeout)
        {
            var deadline = DateTime.UtcNow + timeout;
            while (DateTime.UtcNow < deadline)
            {
                var snapshot = _invocations.ToArray();
                if (snapshot.Length >= count)
                {
                    return snapshot;
                }

                Thread.Sleep(10);
            }

            throw new Xunit.Sdk.XunitException(
                $"Expected at least {count} dial invocations within {timeout}, saw {_invocations.Count}.");
        }
    }

    /// <summary>The four fake sinks plus the pill-transition record.</summary>
    private sealed class Sinks
    {
        public RecordingBroadcaster<AlarmTransitionEvent> AlarmBroadcaster { get; }
        public RecordingBroadcaster<ScriptLogEntry> ScriptBroadcaster { get; }
        public RecordingHealthStore HealthStore { get; } = new();
        public RecordingResilienceStore ResilienceStore { get; } = new();

        public ConcurrentQueue<AlarmTransitionEvent> Alarms { get; } = new();
        public ConcurrentQueue<ScriptLogEntry> Scripts { get; } = new();
        public ConcurrentQueue<bool> PillStates { get; } = new();

        public ConcurrentQueue<DriverHealthChanged> Health => HealthStore.Upserts;
        public ConcurrentQueue<DriverResilienceStatusChanged> Resilience => ResilienceStore.Upserts;

        public Sinks()
        {
            // Both broadcasters share the ONE pill-transition record so a test can assert the alarm and
            // script pills move together (matching the supervisor calling SetConnected on both).
            AlarmBroadcaster = new RecordingBroadcaster<AlarmTransitionEvent>(Alarms, PillStates);
            ScriptBroadcaster = new RecordingBroadcaster<ScriptLogEntry>(Scripts, null);
        }
    }

    private sealed class RecordingBroadcaster<T>(ConcurrentQueue<T> published, ConcurrentQueue<bool>? pill)
        : IInProcessBroadcaster<T>
    {
        public event Action<T>? Received;
        public event Action<bool>? ConnectionStateChanged;

        public bool IsConnected { get; private set; }

        public void Publish(T item)
        {
            published.Enqueue(item);
            Received?.Invoke(item);
        }

        public void SetConnected(bool connected)
        {
            IsConnected = connected;
            pill?.Enqueue(connected);
            ConnectionStateChanged?.Invoke(connected);
        }
    }

    private sealed class RecordingHealthStore : IDriverStatusSnapshotStore
    {
        public ConcurrentQueue<DriverHealthChanged> Upserts { get; } = new();

        public event Action<DriverHealthChanged>? SnapshotChanged;

        public void Upsert(DriverHealthChanged snapshot)
        {
            Upserts.Enqueue(snapshot);
            SnapshotChanged?.Invoke(snapshot);
        }

        public bool TryGet(string driverInstanceId, out DriverHealthChanged snapshot)
        {
            snapshot = null!;
            return false;
        }

        public IReadOnlyCollection<DriverHealthChanged> GetAll() => Upserts.ToArray();
        public int EvictUnconfiguredStale(IReadOnlySet<string> configuredInstanceIds, DateTime olderThanUtc) => 0;
    }

    private sealed class RecordingResilienceStore : IDriverResilienceStatusStore
    {
        public ConcurrentQueue<DriverResilienceStatusChanged> Upserts { get; } = new();

        public event Action<DriverResilienceStatusChanged>? SnapshotChanged;

        public void Upsert(DriverResilienceStatusChanged snapshot)
        {
            Upserts.Enqueue(snapshot);
            SnapshotChanged?.Invoke(snapshot);
        }

        public IReadOnlyList<DriverResilienceStatusChanged> GetForInstance(string driverInstanceId) =>
            Upserts.ToArray();

        public IReadOnlyCollection<DriverResilienceStatusChanged> GetAll() => Upserts.ToArray();
        public int EvictUnconfiguredStale(IReadOnlySet<string> configuredInstanceIds, DateTime olderThanUtc) => 0;
    }

    private sealed class InMemoryFactory(string name) : IDbContextFactory<OtOpcUaConfigDbContext>
    {
        public OtOpcUaConfigDbContext CreateDbContext() =>
            new(new DbContextOptionsBuilder<OtOpcUaConfigDbContext>()
                .UseInMemoryDatabase(name)
                .Options);

        public Task<OtOpcUaConfigDbContext> CreateDbContextAsync(CancellationToken cancellationToken = default) =>
            Task.FromResult(CreateDbContext());
    }
}