ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs

using System.Collections.Concurrent;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;

namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;

/// <summary>
/// Tests for <see cref="SiteAlarmAggregatorActor"/> (plan #10, Task 4) — the transient,
/// in-memory, per-site live alarm aggregator. Covers the tricky bits: seed-then-stream
/// ordering (a delta arriving before the seed completes is neither lost nor double-applied),
/// dedup by AlarmKey, placeholder seeded-but-not-on-stream coherence, NodeA↔NodeB reconnect
/// re-seed, and periodic reconcile authoritative-replace (drift correction).
/// </summary>
public class SiteAlarmAggregatorActorTests : TestKit
{
    private const string SiteId = "site-alpha";
    private const string GrpcNodeA = "http://localhost:5100";
    private const string GrpcNodeB = "http://localhost:5200";
    private const string Instance = "Site1.Pump01";

    public SiteAlarmAggregatorActorTests() : base(@"akka.loglevel = WARNING")
    {
        SiteAlarmAggregatorActor.ReconnectDelay = TimeSpan.FromMilliseconds(50);
        SiteAlarmAggregatorActor.StabilityWindow = TimeSpan.FromSeconds(30);
    }

    // ── Test doubles ────────────────────────────────────────────────────────────

    /// <summary>
    /// Controllable seed fan-out: each invocation returns a fresh Task the test completes
    /// on demand (so seed/reconcile timing relative to buffered deltas is deterministic).
    /// </summary>
    private sealed class SeedStub
    {
        private readonly ConcurrentQueue<TaskCompletionSource<IReadOnlyList<AlarmStateChanged>>> _pending = new();
        private int _callCount;

        public int CallCount => Volatile.Read(ref _callCount);

        public Task<IReadOnlyList<AlarmStateChanged>> Seed(CancellationToken ct)
        {
            Interlocked.Increment(ref _callCount);
            var tcs = new TaskCompletionSource<IReadOnlyList<AlarmStateChanged>>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            ct.Register(() => tcs.TrySetCanceled());
            _pending.Enqueue(tcs);
            return tcs.Task;
        }

        /// <summary>Completes the oldest not-yet-completed seed call with the given rows.</summary>
        public void CompleteNext(params AlarmStateChanged[] rows)
        {
            SpinWait.SpinUntil(() => _pending.TryPeek(out _), TimeSpan.FromSeconds(3));
            if (_pending.TryDequeue(out var tcs))
                tcs.TrySetResult(rows);
            else
                throw new InvalidOperationException("No pending seed to complete.");
        }

        /// <summary>Faults the oldest not-yet-completed seed call (simulates a whole-fan-out failure).</summary>
        public void FaultNext()
        {
            SpinWait.SpinUntil(() => _pending.TryPeek(out _), TimeSpan.FromSeconds(3));
            if (_pending.TryDequeue(out var tcs))
                tcs.TrySetException(new InvalidOperationException("seed boom"));
            else
                throw new InvalidOperationException("No pending seed to fault.");
        }
    }

    private sealed class PublishSink
    {
        private readonly object _lock = new();
        public List<IReadOnlyList<AlarmStateChanged>> Snapshots { get; } = new();

        public void Publish(IReadOnlyList<AlarmStateChanged> snapshot)
        {
            lock (_lock) { Snapshots.Add(snapshot); }
        }

        public IReadOnlyList<AlarmStateChanged>? Latest
        {
            get { lock (_lock) { return Snapshots.Count == 0 ? null : Snapshots[^1]; } }
        }

        public int Count { get { lock (_lock) { return Snapshots.Count; } } }
    }

    private sealed record SiteSub(
        string CorrelationId, Action<AlarmStateChanged> OnAlarm, Action<Exception> OnError, CancellationToken Ct);

    private sealed class MockSiteAlarmStreamClient : SiteStreamGrpcClient
    {
        private readonly object _lock = new();
        private readonly List<SiteSub> _subs = new();
        private readonly List<string> _unsubscribed = new();

        public List<SiteSub> Subs { get { lock (_lock) { return _subs.ToList(); } } }
        public List<string> Unsubscribed { get { lock (_lock) { return _unsubscribed.ToList(); } } }

        public MockSiteAlarmStreamClient() : base() { }

        public override Task SubscribeSiteAsync(
            string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> onError, CancellationToken ct)
        {
            lock (_lock) { _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, ct)); }
            var tcs = new TaskCompletionSource();
            ct.Register(() => tcs.TrySetResult());
            return tcs.Task; // never completes until cancelled (simulates a live stream)
        }

        public override void Unsubscribe(string correlationId)
        {
            lock (_lock) { _unsubscribed.Add(correlationId); }
        }
    }

    private sealed class MockSiteAlarmStreamClientFactory : SiteStreamGrpcClientFactory
    {
        private readonly ConcurrentDictionary<string, MockSiteAlarmStreamClient> _byEndpoint = new();

        public MockSiteAlarmStreamClientFactory()
            : base(Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance) { }

        public MockSiteAlarmStreamClient ClientFor(string endpoint) =>
            _byEndpoint.GetOrAdd(endpoint, _ => new MockSiteAlarmStreamClient());

        public override SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint)
            => ClientFor(grpcEndpoint);

        public override SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint)
            => _byEndpoint.TryGetValue(grpcEndpoint, out var c) ? c : null;
    }

    private (IActorRef Actor, SeedStub Seed, PublishSink Sink, MockSiteAlarmStreamClientFactory Factory) CreateActor(
        TimeSpan? reconcileInterval = null)
    {
        var seed = new SeedStub();
        var sink = new PublishSink();
        var factory = new MockSiteAlarmStreamClientFactory();

        var props = Props.Create(() => new SiteAlarmAggregatorActor(
            SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB,
            reconcileInterval ?? TimeSpan.FromMinutes(10)));

        var actor = Sys.ActorOf(props);
        return (actor, seed, sink, factory);
    }

    private static AlarmStateChanged Alarm(
        string alarmName, string sourceRef, int priority, DateTimeOffset ts,
        bool placeholder = false, string instance = Instance) =>
        new(instance, alarmName, AlarmState.Active, priority, ts)
        {
            SourceReference = sourceRef,
            IsConfiguredPlaceholder = placeholder,
            Kind = placeholder ? AlarmKind.NativeOpcUa : AlarmKind.Computed
        };

    private static AlarmStateChanged? Find(IReadOnlyList<AlarmStateChanged>? rows, string alarmName, string sourceRef) =>
        rows?.FirstOrDefault(r => r.AlarmName == alarmName && r.SourceReference == sourceRef);

    // ── Tests ───────────────────────────────────────────────────────────────────

    [Fact]
    public void Delta_Arriving_Before_Seed_Completes_Is_Buffered_And_Applied_Exactly_Once()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        // Live delta arrives DURING the seed window (new key, not in the seed).
        var t1 = DateTimeOffset.UtcNow;
        actor.Tell(Alarm("Overheat", "", 900, t1));

        // Still buffering — nothing published yet.
        Assert.Equal(0, sink.Count);

        // Seed completes with a DIFFERENT alarm.
        seed.CompleteNext(Alarm("PumpFault", "", 500, t1.AddSeconds(-1)));

        AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3));
        var latest = sink.Latest!;
        Assert.NotNull(Find(latest, "PumpFault", "")); // seeded
        var overheat = Find(latest, "Overheat", "");   // buffered gap-window delta survived
        Assert.NotNull(overheat);
        Assert.Single(latest, r => r.AlarmName == "Overheat"); // exactly once
    }

    [Fact]
    public void Buffered_Delta_Older_Than_Seed_Entry_Is_Dropped()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var tSeed = DateTimeOffset.UtcNow;
        // Buffered delta for the SAME key but OLDER than the seed entry.
        actor.Tell(Alarm("PumpFault", "", 100, tSeed.AddSeconds(-1)));
        seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed));

        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));
        // Give a beat to ensure no late (dropped) delta sneaks in.
        Thread.Sleep(150);
        var latest = sink.Latest!;
        var pumpFault = Find(latest, "PumpFault", "");
        Assert.NotNull(pumpFault);
        Assert.Equal(500, pumpFault!.Priority); // seed value kept; older buffered delta dropped
    }

    [Fact]
    public void Buffered_Delta_Newer_Than_Seed_Entry_Is_Applied()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var tSeed = DateTimeOffset.UtcNow;
        actor.Tell(Alarm("PumpFault", "", 900, tSeed.AddSeconds(1))); // newer than seed
        seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed));

        AwaitCondition(() => sink.Latest is { } l && Find(l, "PumpFault", "")?.Priority == 900,
            TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void Dedup_By_AlarmKey_Distinguishes_SourceReference()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext(); // empty seed
        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        // Same instance + alarm name, different source reference → two distinct rows.
        actor.Tell(Alarm("LevelAlarm", "Tank01.Level.HiHi", 700, t));
        actor.Tell(Alarm("LevelAlarm", "Tank01.Level.LoLo", 300, t));

        AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3));
        var latest = sink.Latest!;
        Assert.NotNull(Find(latest, "LevelAlarm", "Tank01.Level.HiHi"));
        Assert.NotNull(Find(latest, "LevelAlarm", "Tank01.Level.LoLo"));
    }

    [Fact]
    public void Same_Key_Live_Update_Replaces_In_Place()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();
        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        actor.Tell(Alarm("PumpFault", "", 500, t));
        AwaitCondition(() => Find(sink.Latest, "PumpFault", "")?.Priority == 500, TimeSpan.FromSeconds(3));

        // Newer update for the same key replaces (not duplicates).
        actor.Tell(Alarm("PumpFault", "", 800, t.AddSeconds(1)));
        AwaitCondition(() => Find(sink.Latest, "PumpFault", "")?.Priority == 800, TimeSpan.FromSeconds(3));
        Assert.Single(sink.Latest!); // still exactly one row for the key
    }

    [Fact]
    public void Seeded_Native_Alarm_And_Its_Live_Delta_Collapse_Onto_One_Key()
    {
        // Identity parity across the two paths (snapshot seed vs live stream): a native
        // alarm seeded with a non-empty SourceReference and a later live delta for the
        // SAME (instance, alarmName, sourceRef) must collapse to ONE in-place-updated row,
        // not a ghost duplicate. Both paths key on SourceReference, so they must agree.
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        seed.CompleteNext(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 300, t));
        AwaitCondition(() => Find(sink.Latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi") is not null,
            TimeSpan.FromSeconds(3));

        // Live delta for the exact same identity, newer → updates in place.
        actor.Tell(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 800, t.AddSeconds(1)));

        AwaitCondition(
            () => Find(sink.Latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi")?.Priority == 800,
            TimeSpan.FromSeconds(3));
        Assert.Single(sink.Latest!); // exactly one row — no seed/live ghost duplicate
    }

    [Fact]
    public void Placeholder_Seeded_Coexists_With_Live_Real_Alarm_And_Is_Not_Wiped()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        // Seed carries a configured-placeholder row (binding with no active conditions).
        var placeholder = Alarm("Motor1.MotorAlarms", "", 0, t, placeholder: true);
        seed.CompleteNext(placeholder);
        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));

        // A live REAL condition for the same instance arrives (distinct AlarmKey).
        actor.Tell(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 800, t.AddSeconds(1)));

        AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3));
        var latest = sink.Latest!;
        // Placeholder still present (not wiped by the real alarm under a different key).
        var ph = Find(latest, "Motor1.MotorAlarms", "");
        Assert.NotNull(ph);
        Assert.True(ph!.IsConfiguredPlaceholder);
        Assert.NotNull(Find(latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi"));
    }

    [Fact]
    public void GrpcError_Flips_Node_Reconnects_And_ReSeeds()
    {
        var (_, seed, sink, factory) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext(); // initial seed done
        // Wait for the seed to actually be APPLIED (published) so the re-seed on error is
        // not skipped as "fan-out already in flight".
        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));

        // Fail the NodeA stream.
        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("NodeA down"));

        // Reconnect must reach NodeB, AND a re-seed fan-out must have been triggered.
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5));
        // The failed NodeA stream was unsubscribed (relay released, not left zombie).
        Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).Unsubscribed);
    }

    [Fact]
    public void Periodic_Reconcile_Authoritatively_Replaces_Stale_Rows()
    {
        var (_, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(250));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        seed.CompleteNext(Alarm("PumpFault", "", 500, t)); // seed has a row
        AwaitCondition(() => Find(sink.Latest, "PumpFault", "") is not null, TimeSpan.FromSeconds(3));

        // Reconcile tick fires → seed #2 (e.g. instance was disabled) returns EMPTY.
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
        seed.CompleteNext(); // empty

        // The stale PumpFault row must disappear (authoritative replace, not merge).
        AwaitCondition(() => sink.Latest is { } l && l.Count == 0, TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void Failed_Initial_Seed_Does_Not_Publish_But_A_Later_Reconcile_Recovers()
    {
        var (_, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(250));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        // The initial seed fan-out throws as a whole.
        seed.FaultNext();

        // A failed INITIAL seed must not publish (page keeps its poll fallback — not live yet).
        Thread.Sleep(200);
        Assert.Equal(0, sink.Count);

        // The periodic reconcile is the backstop: it re-runs the fan-out (flag was cleared
        // on failure), completes, and the actor publishes → now live.
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
        seed.CompleteNext(Alarm("PumpFault", "", 500, DateTimeOffset.UtcNow));
        AwaitCondition(() => Find(sink.Latest, "PumpFault", "") is not null, TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void Stream_GivenUp_After_MaxRetries_Reopens_On_Reconcile_Tick()
    {
        // After the retry budget is exhausted the live stream is left down, but the
        // aggregator is NOT stopped — the next reconcile tick self-heals it (resets the
        // budget + reopens) so a sustained site outage never permanently kills the feed.
        var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(300));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));

        // Drive initial + 3 retries (each flips node) to exhaust the budget.
        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("1"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeB).Subs[0].OnError(new Exception("2"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5));

        int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count;
        var before = TotalSubs(); // 4

        // Fourth error exceeds MaxRetries → stream given up (no immediate reopen).
        factory.ClientFor(GrpcNodeB).Subs[1].OnError(new Exception("4"));

        // The reconcile tick reopens the stream (self-heal).
        AwaitCondition(() => TotalSubs() > before, TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void Stop_Message_TearsDown_Grpc_And_Stops_Actor()
    {
        var (actor, seed, _, factory) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();

        Watch(actor);
        actor.Tell(new StopSiteAlarmAggregator());

        ExpectTerminated(actor, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Unsubscribed.Contains("corr-1"),
            TimeSpan.FromSeconds(3));
    }
}