lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/TelemetryStreamGrpcServiceTests.cs

using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
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.Commons.Protos.Telemetry.V1;
using ZB.MOM.WW.OtOpcUa.Host.Grpc;
using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;

namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Grpc;

/// <summary>
///     Verifies the node-side telemetry streaming service (per-cluster mesh Phase 5): the hub is
///     fanned to a connected caller as <see cref="TelemetryEvent"/> envelopes, every domain field is
///     transcribed onto its proto field, DateTime→Timestamp mapping is Kind-defensive, a cancelled
///     context ends the stream cleanly, and the process-wide concurrency cap sheds excess dials with
///     <see cref="StatusCode.ResourceExhausted"/>.
/// </summary>
public sealed class TelemetryStreamGrpcServiceTests
{
    private static TelemetryStreamGrpcService NewService(ITelemetryLocalHub hub) =>
        new(hub, NullLogger<TelemetryStreamGrpcService>.Instance);

    private static ServerCallContext Ctx(CancellationToken token) => new FakeServerCallContext(token);

    private static async Task WaitUntilAsync(Func<bool> condition, string because)
    {
        var deadline = DateTime.UtcNow.AddSeconds(5);
        while (DateTime.UtcNow < deadline)
        {
            if (condition())
                return;
            await Task.Delay(15);
        }

        throw new TimeoutException($"Timed out waiting: {because}");
    }

    [Fact]
    public async Task Streams_one_of_each_kind_with_fields_transcribed()
    {
        var hub = new TelemetryLocalHub();
        var service = NewService(hub);
        var writer = new CapturingStreamWriter();
        using var cts = new CancellationTokenSource();
        var request = new TelemetryStreamRequest { CorrelationId = "corr-1" };

        var publishedHealth = new DateTime(2026, 7, 23, 10, 0, 0, DateTimeKind.Utc);
        var health = new DriverHealthChanged(
            "cluster-a", "drv-1", "Healthy", LastSuccessfulReadUtc: null, LastError: null,
            ErrorCount5Min: 3, PublishedUtc: publishedHealth);
        var resilience = new DriverResilienceStatusChanged(
            "drv-1", "host-x", BreakerOpen: true, ConsecutiveFailures: 4, CurrentInFlight: 2,
            LastBreakerOpenUtc: null, LastSampledUtc: DateTime.UtcNow, PublishedUtc: DateTime.UtcNow);

        // Snapshot-style items are cached, so emitting them BEFORE the service attaches guarantees they
        // are replayed to the fresh subscription — a deterministic "the pump is draining" signal.
        hub.Emit(new TelemetryItem.Health(health));
        hub.Emit(new TelemetryItem.Resilience(resilience));

        var pump = service.Subscribe(request, writer, Ctx(cts.Token));

        await WaitUntilAsync(() => writer.Count >= 2, "cached snapshots to be replayed");

        // Append-style items are live-only: emit them once the subscription is proven attached.
        var alarm = new AlarmTransitionEvent(
            "raw/dev/Speed", "Area/Line/Equip", "HighSpeed", "Activated", 700, "over limit", "operator1",
            new DateTime(2026, 7, 23, 11, 0, 0, DateTimeKind.Utc), "LimitAlarm",
            Comment: "manual ack", HistorizeToAveva: true,
            ReferencingEquipmentPaths: ["Area/Line/EquipA", "Area/Line/EquipB"]);
        var script = new ScriptLogEntry(
            "script-9", "Warning", "threshold near", new DateTime(2026, 7, 23, 12, 0, 0, DateTimeKind.Utc),
            VirtualTagId: "vt-1", AlarmId: null, EquipmentId: "equip-1");

        hub.Emit(new TelemetryItem.Alarm(alarm));
        hub.Emit(new TelemetryItem.Script(script));

        await WaitUntilAsync(() => writer.Count >= 4, "all four telemetry events to be streamed");

        cts.Cancel();
        await pump; // cancellation must be swallowed — Subscribe returns cleanly.

        var events = writer.Snapshot();
        events.ShouldAllBe(e => e.CorrelationId == "corr-1");

        var healthEvt = events.Single(e => e.EventCase == TelemetryEvent.EventOneofCase.DriverHealth);
        healthEvt.DriverHealth.ClusterId.ShouldBe("cluster-a");
        healthEvt.DriverHealth.State.ShouldBe("Healthy");
        healthEvt.DriverHealth.ErrorCount5Min.ShouldBe(3);
        healthEvt.DriverHealth.PublishedUtc.ToDateTime().ShouldBe(publishedHealth);
        healthEvt.DriverHealth.LastSuccessfulReadUtc.ShouldBeNull(); // nullable DateTime absent
        healthEvt.DriverHealth.HasLastError.ShouldBeFalse();          // nullable string absent

        var resilienceEvt = events.Single(e => e.EventCase == TelemetryEvent.EventOneofCase.DriverResilience);
        resilienceEvt.DriverResilience.DriverInstanceId.ShouldBe("drv-1");
        resilienceEvt.DriverResilience.BreakerOpen.ShouldBeTrue();
        resilienceEvt.DriverResilience.LastBreakerOpenUtc.ShouldBeNull();

        var alarmEvt = events.Single(e => e.EventCase == TelemetryEvent.EventOneofCase.AlarmTransition);
        alarmEvt.AlarmTransition.AlarmId.ShouldBe("raw/dev/Speed");
        alarmEvt.AlarmTransition.TransitionKind.ShouldBe("Activated");
        alarmEvt.AlarmTransition.Severity.ShouldBe(700);
        alarmEvt.AlarmTransition.Comment.ShouldBe("manual ack");
        alarmEvt.AlarmTransition.HistorizeToAveva.ShouldBeTrue();
        alarmEvt.AlarmTransition.ReferencingEquipmentPaths.ShouldBe(["Area/Line/EquipA", "Area/Line/EquipB"]);

        var scriptEvt = events.Single(e => e.EventCase == TelemetryEvent.EventOneofCase.ScriptLog);
        scriptEvt.ScriptLog.ScriptId.ShouldBe("script-9");
        scriptEvt.ScriptLog.Level.ShouldBe("Warning");
        scriptEvt.ScriptLog.VirtualTagId.ShouldBe("vt-1");
        scriptEvt.ScriptLog.HasAlarmId.ShouldBeFalse(); // nullable string absent
        scriptEvt.ScriptLog.EquipmentId.ShouldBe("equip-1");
    }

    [Fact]
    public async Task Cancelled_context_ends_stream_cleanly()
    {
        var hub = new TelemetryLocalHub();
        var service = NewService(hub);
        var writer = new CapturingStreamWriter();
        using var cts = new CancellationTokenSource();
        var request = new TelemetryStreamRequest { CorrelationId = "corr-cancel" };

        var pump = service.Subscribe(request, writer, Ctx(cts.Token));

        cts.Cancel();
        await Should.NotThrowAsync(() => pump); // client disconnect is normal, not an error.
    }

    [Theory]
    [InlineData("")]
    [InlineData("   ")]
    public async Task Empty_correlation_id_is_invalid_argument(string correlationId)
    {
        var service = NewService(new TelemetryLocalHub());
        using var cts = new CancellationTokenSource();

        var ex = await Should.ThrowAsync<RpcException>(() => service.Subscribe(
            new TelemetryStreamRequest { CorrelationId = correlationId },
            new CapturingStreamWriter(), Ctx(cts.Token)));

        ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
    }

    [Fact]
    public async Task Over_long_correlation_id_is_invalid_argument()
    {
        var service = NewService(new TelemetryLocalHub());
        using var cts = new CancellationTokenSource();

        var ex = await Should.ThrowAsync<RpcException>(() => service.Subscribe(
            new TelemetryStreamRequest { CorrelationId = new string('x', 257) },
            new CapturingStreamWriter(), Ctx(cts.Token)));

        ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
    }

    [Fact]
    public async Task Concurrency_cap_sheds_the_over_the_limit_dial()
    {
        // Serial assembly (see AssemblyInfo) => the static active-stream counter starts at 0 here.
        var hub = new TelemetryLocalHub();
        var service = NewService(hub);
        using var cts = new CancellationTokenSource();

        // Open exactly the cap. Each stream blocks in the pump (hub emits nothing and is never completed),
        // so all 100 stay open, holding the counter at 100.
        var open = new List<Task>();
        for (var i = 0; i < 100; i++)
            open.Add(service.Subscribe(
                new TelemetryStreamRequest { CorrelationId = $"cap-{i}" },
                new CapturingStreamWriter(), Ctx(cts.Token)));

        // The 101st dial must be rejected.
        var ex = await Should.ThrowAsync<RpcException>(() => service.Subscribe(
            new TelemetryStreamRequest { CorrelationId = "cap-overflow" },
            new CapturingStreamWriter(), Ctx(cts.Token)));
        ex.StatusCode.ShouldBe(StatusCode.ResourceExhausted);

        // Drain the 100 open streams; the counter returns to 0 for the next test.
        cts.Cancel();
        await Task.WhenAll(open);

        // With the counter back at 0, a fresh single dial is accepted (proves no leak / no double-count).
        using var cts2 = new CancellationTokenSource();
        var writer = new CapturingStreamWriter();
        var pump = service.Subscribe(
            new TelemetryStreamRequest { CorrelationId = "cap-after" }, writer, Ctx(cts2.Token));
        cts2.Cancel();
        await Should.NotThrowAsync(() => pump);
    }

    [Fact]
    public void ToProto_health_with_unspecified_kind_datetime_does_not_throw_and_roundtrips()
    {
        // Producers use UtcNow, but a Kind=Unspecified value must never crash the stream. The mapper
        // assumes an Unspecified value is already-UTC, so the instant round-trips on any machine.
        var wall = new DateTime(2026, 7, 23, 13, 30, 0, DateTimeKind.Unspecified);
        var expectedUtc = DateTime.SpecifyKind(wall, DateTimeKind.Utc);
        var health = new DriverHealthChanged(
            "cluster-a", "drv-1", "Healthy", LastSuccessfulReadUtc: wall, LastError: null,
            ErrorCount5Min: 0, PublishedUtc: wall);

        TelemetryEvent evt = null!;
        Should.NotThrow(() => evt = TelemetryProtoMapNode.ToProto(new TelemetryItem.Health(health), "c"));

        evt.DriverHealth.PublishedUtc.ToDateTime().ShouldBe(expectedUtc);
        evt.DriverHealth.LastSuccessfulReadUtc.ToDateTime().ShouldBe(expectedUtc);
    }

    [Fact]
    public async Task Client_disconnect_mid_stream_ends_cleanly_without_leaking_a_slot()
    {
        // A broken pipe surfaces as IOException out of WriteAsync. It must be swallowed as a routine
        // disconnect (not escape), and the finally must still decrement the counter. Looping well past
        // the cap proves no leak: a single leaked slot per pass would trip ResourceExhausted by pass 101.
        var hub = new TelemetryLocalHub();
        var service = NewService(hub);

        // Cached snapshot ⇒ every fresh subscription is deterministically handed one item to write.
        hub.Emit(new TelemetryItem.Health(new DriverHealthChanged(
            "cluster-a", "drv-1", "Healthy", null, null, 0, DateTime.UtcNow)));

        for (var i = 0; i < 150; i++)
        {
            using var cts = new CancellationTokenSource();
            var pump = service.Subscribe(
                new TelemetryStreamRequest { CorrelationId = $"disc-{i}" },
                new ThrowingStreamWriter(new IOException("connection reset")),
                Ctx(cts.Token));

            // No exception escapes Subscribe, and no ResourceExhausted ever fires (proves the slot is freed).
            await Should.NotThrowAsync(() => pump);
        }

        // Counter is back at its prior value: a normal stream still opens and pumps.
        using var okCts = new CancellationTokenSource();
        var writer = new CapturingStreamWriter();
        var okPump = service.Subscribe(
            new TelemetryStreamRequest { CorrelationId = "disc-after" }, writer, Ctx(okCts.Token));
        await WaitUntilAsync(() => writer.Count >= 1, "the cached snapshot to stream after the disconnect loop");
        okCts.Cancel();
        await okPump;
    }

    [Fact]
    public void ToProto_null_required_string_does_not_throw_and_maps_to_empty()
    {
        // Required proto string setters throw on null; the domain records carry no runtime guard.
        var alarm = new AlarmTransitionEvent(
            AlarmId: null!, EquipmentPath: null!, AlarmName: null!, TransitionKind: null!,
            Severity: 100, Message: null!, User: null!, TimestampUtc: DateTime.UtcNow,
            AlarmTypeName: null!, Comment: null, HistorizeToAveva: null, ReferencingEquipmentPaths: null);

        AlarmTransition msg = null!;
        Should.NotThrow(() => msg = TelemetryProtoMapNode.ToProto(new TelemetryItem.Alarm(alarm), "c").AlarmTransition);

        msg.AlarmId.ShouldBe("");
        msg.EquipmentPath.ShouldBe("");
        msg.AlarmName.ShouldBe("");
        msg.TransitionKind.ShouldBe("");
        msg.Message.ShouldBe("");
        msg.User.ShouldBe("");
        msg.AlarmTypeName.ShouldBe("");
    }

    [Fact]
    public void ToProto_alarm_omits_absent_nullables()
    {
        var alarm = new AlarmTransitionEvent(
            "id", "Area/Line/Equip", "Name", "Cleared", 100, "msg", "system",
            DateTime.UtcNow, Comment: null, HistorizeToAveva: null, ReferencingEquipmentPaths: null);

        var msg = TelemetryProtoMapNode.ToProto(new TelemetryItem.Alarm(alarm), "c").AlarmTransition;

        msg.HasComment.ShouldBeFalse();
        msg.HasHistorizeToAveva.ShouldBeFalse();
        msg.ReferencingEquipmentPaths.ShouldBeEmpty();
    }

    /// <summary>Minimal <see cref="ServerCallContext"/> exposing a caller-controlled cancellation token.</summary>
    private sealed class FakeServerCallContext(CancellationToken token) : ServerCallContext
    {
        protected override string MethodCore => "/telemetry.v1.TelemetryStreamService/Subscribe";
        protected override string HostCore => "localhost";
        protected override string PeerCore => "ipv4:127.0.0.1:0";
        protected override DateTime DeadlineCore => DateTime.MaxValue;
        protected override Metadata RequestHeadersCore => [];
        protected override CancellationToken CancellationTokenCore => token;
        protected override Metadata ResponseTrailersCore { get; } = [];
        protected override Status StatusCore { get; set; }
        protected override WriteOptions? WriteOptionsCore { get; set; }
        protected override AuthContext AuthContextCore => new(null, new Dictionary<string, List<AuthProperty>>());

        protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) =>
            throw new NotSupportedException();

        protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask;
    }

    /// <summary>
    ///     Captures every streamed <see cref="TelemetryEvent"/>. Implements the cancellable
    ///     <c>WriteAsync</c> overload explicitly — otherwise the interface's default method throws
    ///     <see cref="NotSupportedException"/> for the pump's cancellable lifetime token.
    /// </summary>
    private sealed class CapturingStreamWriter : IServerStreamWriter<TelemetryEvent>
    {
        private readonly List<TelemetryEvent> _items = [];
        private readonly Lock _gate = new();

        public WriteOptions? WriteOptions { get; set; }

        public int Count
        {
            get
            {
                lock (_gate)
                    return _items.Count;
            }
        }

        public Task WriteAsync(TelemetryEvent message)
        {
            lock (_gate)
                _items.Add(message);
            return Task.CompletedTask;
        }

        public Task WriteAsync(TelemetryEvent message, CancellationToken cancellationToken)
        {
            cancellationToken.ThrowIfCancellationRequested();
            return WriteAsync(message);
        }

        public List<TelemetryEvent> Snapshot()
        {
            lock (_gate)
                return [.. _items];
        }
    }

    /// <summary>A stream writer that always throws the given exception — models a broken client pipe.</summary>
    private sealed class ThrowingStreamWriter(Exception toThrow) : IServerStreamWriter<TelemetryEvent>
    {
        public WriteOptions? WriteOptions { get; set; }

        public Task WriteAsync(TelemetryEvent message) => throw toThrow;

        public Task WriteAsync(TelemetryEvent message, CancellationToken cancellationToken)
        {
            cancellationToken.ThrowIfCancellationRequested();
            return WriteAsync(message);
        }
    }
}