lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs

using System.Collections.Concurrent;
using Shouldly;
using Xunit;
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.Runtime.Telemetry;

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

/// <summary>
/// Verifies the node-local telemetry hub (per-cluster mesh Phase 5): snapshot-replay of the two
/// snapshot-style channels (Health / Resilience), live-only forwarding of the append-style logs
/// (Alarm / Script), independent bounded per-subscriber fan-out, DropOldest backpressure, and
/// detach-on-dispose. The hub carries ONLY this node's own telemetry — it never touches DPS.
/// </summary>
public sealed class TelemetryLocalHubTests
{
    private static DriverHealthChanged Health(string instanceId, string state = "Healthy") =>
        new("cluster-a", instanceId, state, null, null, 0, DateTime.UtcNow);

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

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

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

    private static List<TelemetryItem> Drain(ITelemetrySubscription sub)
    {
        var items = new List<TelemetryItem>();
        while (sub.Reader.TryRead(out var item))
            items.Add(item);
        return items;
    }

    [Fact]
    public void Evict_drops_the_instances_Health_and_every_Resilience_snapshot_but_keeps_others()
    {
        var hub = new TelemetryLocalHub();
        hub.Emit(new TelemetryItem.Health(Health("drv-a")));
        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-a", "host-1")));
        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-a", "host-2")));
        hub.Emit(new TelemetryItem.Health(Health("drv-b")));
        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-b", "host-1")));

        hub.Evict("drv-a");
        hub.Evict("never-existed"); // unknown id is a no-op

        using var sub = hub.Subscribe(boundedCapacity: 16);
        var items = Drain(sub);

        // Positive control: drv-b still replays; drv-a does not.
        items.Count.ShouldBe(2);
        items.OfType<TelemetryItem.Health>().Single().E.DriverInstanceId.ShouldBe("drv-b");
        items.OfType<TelemetryItem.Resilience>().Single().E.DriverInstanceId.ShouldBe("drv-b");
    }

    /// <summary>
    ///     #609 — the host evicts a removed instance while its child is still stopping, and that child can
    ///     still publish (e.g. a health snapshot when an in-flight connect/subscribe completes). A snapshot for
    ///     an evicted instance must not re-prime the cache — nor reach live subscribers, where it would
    ///     re-populate central's view of a driver that no longer exists. A neighbour emitted at the same time is
    ///     the positive control.
    /// </summary>
    [Fact]
    public void A_late_snapshot_for_an_evicted_instance_is_neither_cached_nor_delivered()
    {
        var hub = new TelemetryLocalHub();
        hub.Emit(new TelemetryItem.Health(Health("drv-a")));
        hub.Evict("drv-a");
        using var live = hub.Subscribe(boundedCapacity: 16);

        hub.Emit(new TelemetryItem.Health(Health("drv-a")));
        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-a", "host-1")));
        hub.Emit(new TelemetryItem.Health(Health("drv-b")));

        Drain(live).Select(Id).ShouldBe(new[] { "drv-b" });
        using var primed = hub.Subscribe(boundedCapacity: 16);
        Drain(primed).Select(Id).ShouldBe(new[] { "drv-b" });
    }

    /// <summary>
    ///     #609 — eviction does not outlive the removal: the same instance id re-spawned in a later deployment
    ///     is <c>Readmit</c>ted and its snapshots are cached again.
    /// </summary>
    [Fact]
    public void Readmit_lets_an_evicted_instance_publish_again()
    {
        var hub = new TelemetryLocalHub();
        hub.Evict("drv-a");
        hub.Readmit("drv-a");

        hub.Emit(new TelemetryItem.Health(Health("drv-a")));
        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-a", "host-1")));

        using var sub = hub.Subscribe(boundedCapacity: 16);
        Drain(sub).Select(Id).ShouldBe(new[] { "drv-a", "drv-a" });
    }

    private static string Id(TelemetryItem item) => item switch
    {
        TelemetryItem.Health h => h.E.DriverInstanceId,
        TelemetryItem.Resilience r => r.E.DriverInstanceId,
        _ => "<other>",
    };

    [Fact]
    public void New_subscriber_is_primed_with_cached_Health_snapshot()
    {
        var hub = new TelemetryLocalHub();
        hub.Emit(new TelemetryItem.Health(Health("drv-1", "Faulted")));

        using var sub = hub.Subscribe(boundedCapacity: 16);

        var items = Drain(sub);
        items.ShouldHaveSingleItem();
        var health = items[0].ShouldBeOfType<TelemetryItem.Health>();
        health.E.DriverInstanceId.ShouldBe("drv-1");
        health.E.State.ShouldBe("Faulted");
    }

    [Fact]
    public void New_subscriber_is_primed_with_cached_Resilience_snapshot()
    {
        var hub = new TelemetryLocalHub();
        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-a")));

        using var sub = hub.Subscribe(boundedCapacity: 16);

        var items = Drain(sub);
        items.ShouldHaveSingleItem();
        var res = items[0].ShouldBeOfType<TelemetryItem.Resilience>();
        res.E.DriverInstanceId.ShouldBe("drv-1");
        res.E.HostName.ShouldBe("host-a");
    }

    [Fact]
    public void Alarm_and_Script_are_not_replayed_to_a_new_subscriber()
    {
        var hub = new TelemetryLocalHub();
        hub.Emit(new TelemetryItem.Alarm(Alarm("a-1")));
        hub.Emit(new TelemetryItem.Script(Script("s-1")));

        using var sub = hub.Subscribe(boundedCapacity: 16);

        Drain(sub).ShouldBeEmpty();
    }

    [Fact]
    public void Multiple_subscribers_each_receive_their_own_copy_of_a_post_subscribe_emit()
    {
        var hub = new TelemetryLocalHub();
        using var s1 = hub.Subscribe(boundedCapacity: 16);
        using var s2 = hub.Subscribe(boundedCapacity: 16);

        hub.Emit(new TelemetryItem.Alarm(Alarm("a-1")));

        Drain(s1).ShouldHaveSingleItem().ShouldBeOfType<TelemetryItem.Alarm>().E.AlarmId.ShouldBe("a-1");
        Drain(s2).ShouldHaveSingleItem().ShouldBeOfType<TelemetryItem.Alarm>().E.AlarmId.ShouldBe("a-1");
    }

    [Fact]
    public void Health_snapshot_keeps_only_the_latest_per_instance()
    {
        var hub = new TelemetryLocalHub();
        hub.Emit(new TelemetryItem.Health(Health("drv-1", "Reconnecting")));
        hub.Emit(new TelemetryItem.Health(Health("drv-1", "Healthy")));

        using var sub = hub.Subscribe(boundedCapacity: 16);

        var items = Drain(sub);
        items.ShouldHaveSingleItem();
        items[0].ShouldBeOfType<TelemetryItem.Health>().E.State.ShouldBe("Healthy");
    }

    [Fact]
    public void Health_snapshot_keeps_distinct_instances()
    {
        var hub = new TelemetryLocalHub();
        hub.Emit(new TelemetryItem.Health(Health("drv-1")));
        hub.Emit(new TelemetryItem.Health(Health("drv-2")));

        using var sub = hub.Subscribe(boundedCapacity: 16);

        var ids = Drain(sub)
            .OfType<TelemetryItem.Health>()
            .Select(h => h.E.DriverInstanceId)
            .OrderBy(x => x)
            .ToArray();
        ids.ShouldBe(new[] { "drv-1", "drv-2" });
    }

    [Fact]
    public void Resilience_snapshot_keeps_only_latest_per_instance_and_host()
    {
        var hub = new TelemetryLocalHub();
        // Same instance, two distinct hosts → two cache entries.
        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-a")));
        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-b")));
        // Overwrite (drv-1, host-a).
        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-a")));

        using var sub = hub.Subscribe(boundedCapacity: 16);

        var keys = Drain(sub)
            .OfType<TelemetryItem.Resilience>()
            .Select(r => (r.E.DriverInstanceId, r.E.HostName))
            .OrderBy(x => x.HostName)
            .ToArray();
        keys.ShouldBe(new[] { ("drv-1", "host-a"), ("drv-1", "host-b") });
    }

    [Fact]
    public void Bounded_channel_drops_oldest_not_newest_when_full()
    {
        var hub = new TelemetryLocalHub();
        using var sub = hub.Subscribe(boundedCapacity: 2);

        // Append-style (uncached) items so the snapshot doesn't consume capacity.
        hub.Emit(new TelemetryItem.Alarm(Alarm("a-1")));
        hub.Emit(new TelemetryItem.Alarm(Alarm("a-2")));
        hub.Emit(new TelemetryItem.Alarm(Alarm("a-3")));

        var ids = Drain(sub).OfType<TelemetryItem.Alarm>().Select(a => a.E.AlarmId).ToArray();
        // a-1 (oldest) evicted; the two newest survive in order.
        ids.ShouldBe(new[] { "a-2", "a-3" });
    }

    [Fact]
    public void Subscribe_with_non_positive_capacity_throws()
    {
        var hub = new TelemetryLocalHub();
        Should.Throw<ArgumentOutOfRangeException>(() => hub.Subscribe(0));
        Should.Throw<ArgumentOutOfRangeException>(() => hub.Subscribe(-1));
    }

    [Fact]
    public async Task Concurrent_emit_subscribe_dispose_is_safe_and_never_double_delivers_a_snapshot()
    {
        var hub = new TelemetryLocalHub();
        var stop = new CancellationTokenSource(TimeSpan.FromSeconds(3));
        var exceptions = new ConcurrentQueue<Exception>();

        // Writers: hammer Emit with a mix of all four kinds across a small key space so the snapshot
        // caches churn (same instance/host reused → last-value overwrites) while subscribers attach.
        var writers = Enumerable.Range(0, 4).Select(w => Task.Run(() =>
        {
            var rnd = new Random(w * 7919 + 1);
            try
            {
                var n = 0;
                while (!stop.IsCancellationRequested)
                {
                    var instance = "drv-" + rnd.Next(0, 4);
                    var host = "host-" + rnd.Next(0, 3);
                    TelemetryItem item = (n++ % 4) switch
                    {
                        0 => new TelemetryItem.Health(Health(instance, "s" + n)),
                        1 => new TelemetryItem.Resilience(Resilience(instance, host)),
                        2 => new TelemetryItem.Alarm(Alarm("a-" + n)),
                        _ => new TelemetryItem.Script(Script("s-" + n)),
                    };
                    hub.Emit(item);
                }
            }
            catch (Exception ex) { exceptions.Enqueue(ex); }
        })).ToArray();

        // Subscriber/disposer churn: attach, drain, verify the exactly-once-across-attach invariant, dispose.
        var churners = Enumerable.Range(0, 6).Select(_ => Task.Run(() =>
        {
            try
            {
                while (!stop.IsCancellationRequested)
                {
                    using var sub = hub.Subscribe(boundedCapacity: 4096);
                    // The snapshot prelude is fully written before Subscribe returns, so any Health/
                    // Resilience item read here that was NOT part of the prelude must be a live delta —
                    // i.e. it arrived strictly after attach. The invariant we assert: for one snapshot
                    // KEY, the hub never delivers the SAME snapshot instance both in the prelude and
                    // again live. We approximate "same snapshot" by reference identity: a live re-emit
                    // is always a freshly-allocated record, so a reference-equal duplicate would be a
                    // genuine double-delivery of one cached object across the boundary.
                    var seen = new HashSet<TelemetryItem>(ReferenceEqualityComparer.Instance);
                    while (sub.Reader.TryRead(out var item))
                    {
                        // (c) every drained item is a valid, non-null TelemetryItem of a known kind.
                        item.ShouldNotBeNull();
                        item.ShouldBeAssignableTo<TelemetryItem>();
                        // (b) no cached snapshot object delivered twice to THIS subscriber.
                        seen.Add(item).ShouldBeTrue();
                    }
                }
            }
            catch (Exception ex) { exceptions.Enqueue(ex); }
        })).ToArray();

        await Task.WhenAll(writers.Concat(churners));

        // (a) no thread threw.
        exceptions.ShouldBeEmpty();
    }

    [Fact]
    public void Dispose_detaches_the_subscriber_and_completes_its_channel()
    {
        var hub = new TelemetryLocalHub();
        var sub = hub.Subscribe(boundedCapacity: 16);

        sub.Dispose();

        // A further Emit must neither throw nor deliver to the detached subscriber.
        Should.NotThrow(() => hub.Emit(new TelemetryItem.Alarm(Alarm("a-1"))));
        sub.Reader.Completion.IsCompleted.ShouldBeTrue();
        sub.Reader.TryRead(out _).ShouldBeFalse();
    }
}