lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyAlarmFeedTests.cs

using System.Runtime.CompilerServices;
using Google.Protobuf.WellKnownTypes;
using MxGateway.Contracts.Proto;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;

namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;

/// <summary>
///     Pins <see cref="GatewayGalaxyAlarmFeed"/> — the session-less consumer of the
///     gateway's <c>StreamAlarms</c> feed. Synthetic <see cref="AlarmFeedMessage"/>s go
///     in through the stream-factory seam; the feed fires <c>OnAlarmTransition</c> with
///     decoded payloads and mapped severity buckets, drops malformed messages, and
///     re-opens the stream after a transport fault.
/// </summary>
public sealed class GatewayGalaxyAlarmFeedTests
{
    [Fact]
    public async Task Decodes_active_alarm_snapshot_then_live_transition()
    {
        var raise = new DateTime(2026, 5, 1, 12, 0, 0, DateTimeKind.Utc);
        var messages = new[]
        {
            SnapshotMessage("Tank01.Level.HiHi", AlarmConditionState.Active, severity: 750,
                lastTransition: raise),
            SnapshotMessage("Tank02.Level.HiHi", AlarmConditionState.ActiveAcked, severity: 500,
                lastTransition: raise, operatorUser: "alice", operatorComment: "investigating"),
            new AlarmFeedMessage { SnapshotComplete = true },
            TransitionMessage("Tank01.Level.HiHi", AlarmTransitionKind.Clear, severity: 750,
                transitionTime: raise.AddMinutes(5), originalRaise: raise),
        };

        var observed = new List<GalaxyAlarmTransition>();
        var got3 = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);

        await using var feed = new GatewayGalaxyAlarmFeed(
            (_, ct) => OpenStream(messages, ct), clientName: "FeedTest");
        feed.OnAlarmTransition += (_, t) =>
        {
            lock (observed)
            {
                observed.Add(t);
                if (observed.Count == 3) got3.TrySetResult(true);
            }
        };
        feed.Start();

        (await Task.WhenAny(got3.Task, Task.Delay(TimeSpan.FromSeconds(2))))
            .ShouldBe(got3.Task, "snapshot + transition should dispatch within 2s");

        observed.Count.ShouldBe(3);

        // Active snapshot entry → Raise.
        observed[0].AlarmFullReference.ShouldBe("Tank01.Level.HiHi");
        observed[0].TransitionKind.ShouldBe(GalaxyAlarmTransitionKind.Raise);
        observed[0].SeverityBucket.ShouldBe(AlarmSeverity.Critical);
        observed[0].RawMxAccessSeverity.ShouldBe(750);

        // Acknowledged snapshot entry → Acknowledge, operator fields preserved.
        observed[1].TransitionKind.ShouldBe(GalaxyAlarmTransitionKind.Acknowledge);
        observed[1].OperatorUser.ShouldBe("alice");
        observed[1].OperatorComment.ShouldBe("investigating");

        // Live transition after snapshot_complete.
        observed[2].TransitionKind.ShouldBe(GalaxyAlarmTransitionKind.Clear);
        observed[2].OriginalRaiseTimestampUtc.ShouldBe(raise);
    }

    [Fact]
    public async Task Drops_transition_with_unspecified_kind_and_empty_message()
    {
        var messages = new[]
        {
            TransitionMessage("Tank01.Level.HiHi", AlarmTransitionKind.Unspecified, severity: 100,
                transitionTime: DateTime.UtcNow),
            new AlarmFeedMessage(), // empty oneof — version skew
            TransitionMessage("Tank01.Level.HiHi", AlarmTransitionKind.Raise, severity: 600,
                transitionTime: DateTime.UtcNow),
        };

        var observed = new List<GalaxyAlarmTransition>();
        var gotOne = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);

        await using var feed = new GatewayGalaxyAlarmFeed(
            (_, ct) => OpenStream(messages, ct), clientName: "FeedTest");
        feed.OnAlarmTransition += (_, t) =>
        {
            lock (observed)
            {
                observed.Add(t);
                gotOne.TrySetResult(true);
            }
        };
        feed.Start();

        (await Task.WhenAny(gotOne.Task, Task.Delay(TimeSpan.FromSeconds(2))))
            .ShouldBe(gotOne.Task);

        // Only the well-formed Raise survives; the Unspecified + empty messages drop.
        observed.ShouldHaveSingleItem();
        observed[0].TransitionKind.ShouldBe(GalaxyAlarmTransitionKind.Raise);
        observed[0].SeverityBucket.ShouldBe(AlarmSeverity.High);
    }

    [Fact]
    public async Task Reopens_stream_after_a_transport_fault()
    {
        var calls = 0;
        var liveTransition = new[]
        {
            TransitionMessage("Tank01.Level.HiHi", AlarmTransitionKind.Raise, severity: 750,
                transitionTime: DateTime.UtcNow),
        };

        var observed = new List<GalaxyAlarmTransition>();
        var gotOne = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);

        await using var feed = new GatewayGalaxyAlarmFeed(
            (_, ct) =>
            {
                // First open faults; the feed must reconnect and succeed on the retry.
                if (Interlocked.Increment(ref calls) == 1)
                {
                    throw new InvalidOperationException("synthetic stream fault");
                }
                return OpenStream(liveTransition, ct);
            },
            clientName: "ReconnectTest",
            reconnectDelay: TimeSpan.FromMilliseconds(20));
        feed.OnAlarmTransition += (_, t) =>
        {
            observed.Add(t);
            gotOne.TrySetResult(true);
        };
        feed.Start();

        (await Task.WhenAny(gotOne.Task, Task.Delay(TimeSpan.FromSeconds(3))))
            .ShouldBe(gotOne.Task, "the feed should reopen the stream and deliver after a fault");

        calls.ShouldBeGreaterThanOrEqualTo(2);
        observed.ShouldHaveSingleItem();
        observed[0].TransitionKind.ShouldBe(GalaxyAlarmTransitionKind.Raise);
    }

    /// <summary>
    ///     Yields each message in order, then holds the stream open until the feed is
    ///     disposed — mirrors a live server-streaming RPC that does not complete on its
    ///     own.
    /// </summary>
    private static async IAsyncEnumerable<AlarmFeedMessage> OpenStream(
        IEnumerable<AlarmFeedMessage> messages,
        [EnumeratorCancellation] CancellationToken ct = default)
    {
        foreach (var message in messages)
        {
            ct.ThrowIfCancellationRequested();
            yield return message;
            await Task.Yield();
        }
        await Task.Delay(Timeout.Infinite, ct);
    }

    private static AlarmFeedMessage SnapshotMessage(
        string fullReference,
        AlarmConditionState state,
        int severity,
        DateTime lastTransition,
        string operatorUser = "",
        string operatorComment = "")
        => new()
        {
            ActiveAlarm = new ActiveAlarmSnapshot
            {
                AlarmFullReference = fullReference,
                SourceObjectReference = fullReference.Split('.')[0],
                AlarmTypeName = "AnalogLimitAlarm.HiHi",
                Severity = severity,
                CurrentState = state,
                Category = "Process",
                Description = "Tank high-high level",
                LastTransitionTimestamp = Timestamp.FromDateTime(lastTransition),
                OperatorUser = operatorUser,
                OperatorComment = operatorComment,
            },
        };

    private static AlarmFeedMessage TransitionMessage(
        string fullReference,
        AlarmTransitionKind kind,
        int severity,
        DateTime transitionTime,
        DateTime? originalRaise = null)
    {
        var body = new OnAlarmTransitionEvent
        {
            AlarmFullReference = fullReference,
            SourceObjectReference = fullReference.Split('.')[0],
            AlarmTypeName = "AnalogLimitAlarm.HiHi",
            TransitionKind = kind,
            Severity = severity,
            TransitionTimestamp = Timestamp.FromDateTime(transitionTime),
            Category = "Process",
            Description = "Tank high-high level",
        };
        if (originalRaise is { } orts)
        {
            body.OriginalRaiseTimestamp = Timestamp.FromDateTime(orts);
        }
        return new AlarmFeedMessage { Transition = body };
    }
}