lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianAdapterActorTests.cs

using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

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

/// <summary>
/// TestKit coverage for <see cref="HistorianAdapterActor"/>'s Primary-only historization gate.
/// The actor caches this node's <see cref="RedundancyRole"/> from the <c>redundancy-state</c>
/// topic and SKIPS the sink enqueue when the local node is Secondary/Detached so a future
/// per-node feeder writes exactly once across the warm-redundant pair. Unknown/null role
/// default-writes (single-node deploys + the boot window must never silently drop historization).
/// </summary>
public sealed class HistorianAdapterActorTests : RuntimeActorTestBase
{
    /// <summary>The local node id the gating tests construct the adapter with.</summary>
    private static readonly NodeId LocalNode = new("node-A");

    /// <summary>
    ///     The window an <b>absence</b> assertion waits before concluding nothing arrived
    ///     (<c>ExpectNoMsg</c>). Its length is a calibration decision — long enough that a message which
    ///     was going to arrive would have — so it is deliberately NOT generous.
    /// </summary>
    private static readonly TimeSpan Settle = TimeSpan.FromMilliseconds(500);

    /// <summary>
    ///     The budget a <b>presence</b> assertion may take to become true (<c>AwaitAssert</c>).
    /// </summary>
    /// <remarks>
    ///     Separate from <see cref="Settle"/> on purpose, though they once shared its 500 ms. The two are
    ///     different quantities that merely had the same number: a presence budget is an upper bound before
    ///     giving up, and <c>AwaitAssert</c> returns the instant the condition holds, so a generous value
    ///     costs nothing in the passing case and can never make a genuinely failing assertion pass. An
    ///     absence window is the opposite — every millisecond is spent on every run. Conflating them meant
    ///     the enqueue assertions could only be given more headroom by slowing every <c>ExpectNoMsg</c> in
    ///     the class, so they kept a 500 ms budget that a fully parallel assembly run occasionally missed.
    /// </remarks>
    private static readonly TimeSpan AssertTimeout = TimeSpan.FromSeconds(5);

    /// <summary>Thread-safe fake sink that records every <see cref="EnqueueAsync"/> call.</summary>
    private sealed class RecordingSink : IAlarmHistorianSink
    {
        private readonly object _lock = new();
        private readonly List<AlarmHistorianEvent> _events = new();

        /// <summary>The number of <see cref="EnqueueAsync"/> calls observed so far.</summary>
        public int EnqueueCount { get { lock (_lock) { return _events.Count; } } }

        /// <summary>A snapshot of every event enqueued so far (in arrival order).</summary>
        public IReadOnlyList<AlarmHistorianEvent> Events
        {
            get { lock (_lock) { return _events.ToArray(); } }
        }

        /// <inheritdoc />
        public Task EnqueueAsync(AlarmHistorianEvent evt, CancellationToken cancellationToken)
        {
            lock (_lock)
            {
                _events.Add(evt);
            }

            return Task.CompletedTask;
        }

        /// <inheritdoc />
        public HistorianSinkStatus GetStatus() => new(
            QueueDepth: 0,
            DeadLetterDepth: 0,
            LastDrainUtc: null,
            LastSuccessUtc: null,
            LastError: null,
            DrainState: HistorianDrainState.Idle);
    }

    /// <summary>Tell <paramref name="actor"/> a <see cref="RedundancyStateChanged"/> snapshot marking
    /// <see cref="LocalNode"/> with <paramref name="role"/> so the gate observes the local role.</summary>
    private static void TellRedundancyRole(IActorRef actor, RedundancyRole role) =>
        actor.Tell(new RedundancyStateChanged(
            new[]
            {
                new NodeRedundancyState(
                    NodeId: LocalNode,
                    Role: role,
                    IsClusterLeader: role == RedundancyRole.Primary,
                    IsDriverPrimary: role == RedundancyRole.Primary,
                    AsOfUtc: DateTime.UtcNow),
            },
            CorrelationId.NewId()));

    /// <summary>Default-write (T1): before any redundancy snapshot — the boot window and the steady
    /// state for single-node deploys — the adapter MUST historize. Constructed WITH a localNode but
    /// no snapshot sent, so the cached role is unknown ⇒ default-write.</summary>
    [Fact]
    public void Default_before_redundancy_state_historizes()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));

        actor.Tell(SampleTransition());

        AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout, cancellationToken: TestContext.Current.CancellationToken);
    }

    /// <summary>Secondary suppression (T2): when the cached local role is Secondary, the adapter MUST
    /// NOT enqueue to the durable sink (the Primary writes the single copy).</summary>
    [Fact]
    public void Secondary_node_does_not_historize()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));

        TellRedundancyRole(actor, RedundancyRole.Secondary);
        actor.Tell(SampleTransition());

        // Give the (suppressed) fire-and-forget a stable window, then assert nothing landed.
        ExpectNoMsg(Settle, TestContext.Current.CancellationToken);
        sink.EnqueueCount.ShouldBe(0);
    }

    /// <summary>Detached suppression (T3): a Detached node likewise MUST NOT historize.</summary>
    [Fact]
    public void Detached_node_does_not_historize()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));

        TellRedundancyRole(actor, RedundancyRole.Detached);
        actor.Tell(SampleTransition());

        ExpectNoMsg(Settle, TestContext.Current.CancellationToken);
        sink.EnqueueCount.ShouldBe(0);
    }

    /// <summary>Primary writes (T4): when the cached local role is Primary, the adapter historizes as
    /// normal (this is the single copy the durable sink sees).</summary>
    [Fact]
    public void Primary_node_historizes()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));

        TellRedundancyRole(actor, RedundancyRole.Primary);
        actor.Tell(SampleTransition());

        AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout, cancellationToken: TestContext.Current.CancellationToken);
    }

    /// <summary>Absent-node default-historize (T5): a snapshot that mentions only a DIFFERENT node
    /// must NOT update the local cached role — the actor's own node is absent, so the role stays
    /// null/unknown and the default-historize path must fire. Partial/stale snapshots MUST NOT
    /// silently suppress historization for nodes not yet observed.</summary>
    [Fact]
    public void Redundancy_snapshot_without_local_node_leaves_role_unknown_and_historizes()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));

        // Send a snapshot that only describes a peer node — the local node is absent.
        actor.Tell(new RedundancyStateChanged(
            new[]
            {
                new NodeRedundancyState(
                    NodeId: new NodeId("some-other-node"),
                    Role: RedundancyRole.Secondary,
                    IsClusterLeader: false,
                    IsDriverPrimary: false,
                    AsOfUtc: DateTime.UtcNow),
            },
            CorrelationId.NewId()));

        actor.Tell(SampleTransition());

        // Local role is still unknown ⇒ default-historize path: sink must record exactly one enqueue.
        AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout, cancellationToken: TestContext.Current.CancellationToken);
    }

    /// <summary>Builds an <see cref="AlarmTransitionEvent"/> (the shape published on the <c>alerts</c>
    /// DPS topic) for the translate tests, with overridable severity / type / comment / kind.
    /// <paramref name="historizeToAveva"/> is <c>bool?</c> so tests can pass <c>null</c> to simulate the
    /// rolling-restart / cross-version case (missing field → CLR default null).</summary>
    private static AlarmTransitionEvent SampleTransition(
        int severity = 750,
        string alarmTypeName = "LimitAlarm",
        string? comment = "note",
        string transitionKind = "Activated",
        bool? historizeToAveva = true) => new(
        AlarmId: "alm-9",
        EquipmentPath: "Area/Line/Equip",
        AlarmName: "HiHi",
        TransitionKind: transitionKind,
        Severity: severity,
        Message: "level high",
        User: "operator1",
        TimestampUtc: DateTime.UtcNow,
        AlarmTypeName: alarmTypeName,
        Comment: comment,
        HistorizeToAveva: historizeToAveva);

    /// <summary>Alerts translate (T6): an <see cref="AlarmTransitionEvent"/> off the <c>alerts</c> topic
    /// is translated to an <see cref="AlarmHistorianEvent"/> and historized by default (unknown role).
    /// The translation must carry AlarmId, AlarmTypeName, EventKind (← TransitionKind), Severity bucket,
    /// and Comment through faithfully.</summary>
    [Fact]
    public void Alerts_transition_is_historized_by_default()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));

        actor.Tell(SampleTransition());

        AwaitAssert(
            () =>
            {
                sink.EnqueueCount.ShouldBe(1);
                var e = sink.Events.ShouldHaveSingleItem();
                e.AlarmId.ShouldBe("alm-9");
                e.AlarmTypeName.ShouldBe("LimitAlarm");
                e.EventKind.ShouldBe("Activated");
                e.Severity.ShouldBe(AlarmSeverity.High);
                e.Comment.ShouldBe("note");
            },
            AssertTimeout, cancellationToken: TestContext.Current.CancellationToken);
    }

    /// <summary>Secondary suppression for alerts (T7): a Secondary node must NOT historize a transition
    /// off the <c>alerts</c> topic — the Primary writes the single copy (DistributedPubSub fans the
    /// single publish to BOTH nodes' historian adapters, so the gate is what makes it exactly-once).</summary>
    [Fact]
    public void Secondary_node_does_not_historize_alerts_transition()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));

        TellRedundancyRole(actor, RedundancyRole.Secondary);
        actor.Tell(SampleTransition());

        ExpectNoMsg(Settle, TestContext.Current.CancellationToken);
        sink.EnqueueCount.ShouldBe(0);
    }

    /// <summary>Primary writes alerts (T8): a Primary node historizes a transition off the
    /// <c>alerts</c> topic (the single copy the durable sink sees).</summary>
    [Fact]
    public void Primary_node_historizes_alerts_transition()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));

        TellRedundancyRole(actor, RedundancyRole.Primary);
        actor.Tell(SampleTransition());

        AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout, cancellationToken: TestContext.Current.CancellationToken);
    }

    /// <summary>Per-alarm opt-out (T8b): a Primary node must NOT historize a transition whose
    /// <c>HistorizeToAveva</c> is <c>false</c> — that flag is a per-alarm opt-out of DURABLE
    /// historization only. The live <c>alerts</c> fan-out already happened upstream (the publish is NOT
    /// gated on this flag), so only the durable sink write is suppressed.</summary>
    [Fact]
    public void Primary_node_does_not_historize_when_opted_out()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));

        TellRedundancyRole(actor, RedundancyRole.Primary);
        actor.Tell(SampleTransition(historizeToAveva: false));

        ExpectNoMsg(Settle, TestContext.Current.CancellationToken);
        sink.EnqueueCount.ShouldBe(0);
    }

    /// <summary>Rolling-restart default-on (T8c): when <c>HistorizeToAveva</c> is <c>null</c> — the shape
    /// a cross-version / rolling-restart deserialize produces (old-format message missing the field maps to
    /// the CLR default <c>null</c> for <c>bool?</c>) — a Primary node MUST historize. <c>null</c> is the
    /// safe default-on posture: no audit row is dropped at a handover, matching the <c>AlarmTypeName</c>
    /// null-coalesce precedent in the same <c>HistorianAdapterActor.Translate</c>.</summary>
    [Fact]
    public void Primary_historizes_when_flag_is_null()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));

        TellRedundancyRole(actor, RedundancyRole.Primary);
        actor.Tell(SampleTransition(historizeToAveva: null));

        AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout, cancellationToken: TestContext.Current.CancellationToken);
    }

    /// <summary>Severity buckets (T9): the OPC UA 1–1000 numeric severity on the transition maps onto
    /// the coarse <see cref="AlarmSeverity"/> at the same ceilings <c>ScriptedAlarmHostActor.SeverityToInt</c>
    /// emits (Low≤250, Medium≤500, High≤750, Critical otherwise). Driven end-to-end through the enqueue.</summary>
    [Theory]
    [InlineData(250, AlarmSeverity.Low)]
    [InlineData(251, AlarmSeverity.Medium)]
    [InlineData(500, AlarmSeverity.Medium)]
    [InlineData(750, AlarmSeverity.High)]
    [InlineData(751, AlarmSeverity.Critical)]
    [InlineData(1000, AlarmSeverity.Critical)]
    public void Alerts_transition_severity_buckets(int severity, AlarmSeverity expected)
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));

        actor.Tell(SampleTransition(severity: severity));

        AwaitAssert(
            () => sink.Events.ShouldHaveSingleItem().Severity.ShouldBe(expected),
            AssertTimeout, cancellationToken: TestContext.Current.CancellationToken);
    }

    /// <summary>Rolling-restart null default (T10): an old-format transition deserialized by Akka's JSON
    /// serializer applies the CLR default (null) to <c>AlarmTypeName</c> rather than the record's
    /// "AlarmCondition" call-site default. The translation must null-coalesce that back to
    /// "AlarmCondition" so the historian never stores a null alarm type. Forced here by constructing the
    /// transition with <c>AlarmTypeName: null!</c> (simulating the post-deserialization shape).</summary>
    [Fact]
    public void Alerts_transition_with_missing_AlarmTypeName_defaults()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));

        actor.Tell(SampleTransition(alarmTypeName: null!));

        AwaitAssert(
            () => sink.Events.ShouldHaveSingleItem().AlarmTypeName.ShouldBe("AlarmCondition"),
            AssertTimeout, cancellationToken: TestContext.Current.CancellationToken);
    }

    /// <summary>
    ///     The direct <see cref="AlarmHistorianEvent"/> receive is gone (it had no production sender and was a latent
    ///     double-historize trap: a transition told directly AND published on <c>alerts</c> would be written twice). A
    ///     directly-told event is unhandled; the positive control — a transition after it — is historized exactly once.
    /// </summary>
    [Fact]
    public void A_directly_told_AlarmHistorianEvent_is_not_historized()
    {
        var sink = new RecordingSink();
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));

        actor.Tell(new AlarmHistorianEvent("alm-1", "Area/Line/Equip", "HiHi", "LimitAlarm", AlarmSeverity.High,
            "Activated", "level high", "system", null, DateTime.UtcNow));
        actor.Tell(SampleTransition());

        AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout, cancellationToken: TestContext.Current.CancellationToken);
        ExpectNoMsg(Settle, TestContext.Current.CancellationToken);
        sink.Events.ShouldHaveSingleItem().AlarmId.ShouldBe("alm-9");
    }

    /// <summary>A sink whose enqueue takes time and reports how many enqueues overlap.</summary>
    private sealed class SlowSink(TimeSpan delay) : IAlarmHistorianSink
    {
        private int _inFlight;
        private int _maxInFlight;
        private readonly object _lock = new();
        private readonly List<string> _order = new();

        public int MaxInFlight => Volatile.Read(ref _maxInFlight);
        public IReadOnlyList<string> Order { get { lock (_lock) return _order.ToArray(); } }

        public async Task EnqueueAsync(AlarmHistorianEvent evt, CancellationToken cancellationToken)
        {
            var now = Interlocked.Increment(ref _inFlight);
            int seen;
            while ((seen = Volatile.Read(ref _maxInFlight)) < now && Interlocked.CompareExchange(ref _maxInFlight, now, seen) != seen) { }
            await Task.Delay(delay, cancellationToken);
            lock (_lock) _order.Add(evt.Message ?? "");
            Interlocked.Decrement(ref _inFlight);
        }

        public HistorianSinkStatus GetStatus() => new(0, 0, null, null, null, HistorianDrainState.Idle);
    }

    /// <summary>
    ///     An alarm storm no longer fans into one concurrent sink insert per transition: the adapter drains a bounded
    ///     queue with one loop, so the sink sees one enqueue at a time, in arrival order, and every event lands.
    /// </summary>
    [Fact]
    public void An_alarm_storm_is_enqueued_one_at_a_time_in_order()
    {
        var sink = new SlowSink(TimeSpan.FromMilliseconds(5));
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));

        for (var i = 0; i < 40; i++) actor.Tell(SampleTransition() with { Message = $"m{i:00}" });

        AwaitAssert(() => sink.Order.Count.ShouldBe(40), TimeSpan.FromSeconds(10), cancellationToken: TestContext.Current.CancellationToken);
        sink.MaxInFlight.ShouldBe(1, "one loop drains the queue — never one concurrent insert per transition");
        sink.Order.ShouldBe(Enumerable.Range(0, 40).Select(i => $"m{i:00}"));
    }

    /// <summary>Stopping the adapter flushes the events still queued within the stop budget.</summary>
    [Fact]
    public void Stopping_flushes_the_queued_events()
    {
        var sink = new SlowSink(TimeSpan.FromMilliseconds(20));
        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));
        for (var i = 0; i < 10; i++) actor.Tell(SampleTransition() with { Message = $"m{i:00}" });

        // PoisonPill queues behind the transitions (Sys.Stop would overtake them in the mailbox): all ten are accepted
        // into the adapter's queue, most still waiting on the slow sink when the actor stops — and they are flushed.
        Watch(actor);
        actor.Tell(PoisonPill.Instance);
        ExpectTerminated(actor, TimeSpan.FromSeconds(10), cancellationToken: TestContext.Current.CancellationToken);
        sink.Order.Count.ShouldBe(10);
    }
}