lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAlarmWriterPipeliningTests.cs

using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;

/// <summary>
///     The gateway has no batch SendEvent RPC, so the alarm writer pipelines its per-event sends — at most
///     <see cref="GatewayAlarmHistorianWriter.MaxSendsInFlight"/> at once — instead of one round trip after another. Every
///     event still gets its own outcome, in input order.
/// </summary>
public sealed class GatewayAlarmWriterPipeliningTests
{
    private static CancellationToken Ct => TestContext.Current.CancellationToken;

    private static AlarmHistorianEvent Evt(int i) => new(
        $"alarm-{i}", "Area/Pump", "N", "LimitAlarm", AlarmSeverity.High, "Activated", $"m{i}", "u", null,
        new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).AddSeconds(i));

    [Fact]
    public async Task Sends_overlap_up_to_the_bound_and_outcomes_keep_input_order()
    {
        var inFlight = 0;
        var maxInFlight = 0;
        var fake = new FakeHistorianGatewayClient
        {
            SendEventSource = async (e, ct) =>
            {
                var now = Interlocked.Increment(ref inFlight);
                InterlockedMax(ref maxInFlight, now);
                await Task.Delay(20, ct);
                Interlocked.Decrement(ref inFlight);
                // Odd messages are soft failures: their outcome must land on their own slot.
                return e.Properties.TryGetValue("Message", out var m) && int.Parse(m[1..]) % 2 == 1
                    ? new WriteAck { Success = false }
                    : new WriteAck { Success = true };
            },
        };
        using var writer = new GatewayAlarmHistorianWriter(fake, NullLogger<GatewayAlarmHistorianWriter>.Instance);

        var outcomes = await writer.WriteBatchAsync(Enumerable.Range(0, 40).Select(Evt).ToList(), Ct);

        outcomes.Count.ShouldBe(40);
        for (var i = 0; i < 40; i++)
            outcomes[i].ShouldBe(i % 2 == 1 ? HistorianWriteOutcome.RetryPlease : HistorianWriteOutcome.Ack, $"event {i}");
        maxInFlight.ShouldBeGreaterThan(1, "the sends must overlap — a sequential drain is 40 round trips");
        maxInFlight.ShouldBeLessThanOrEqualTo(GatewayAlarmHistorianWriter.MaxSendsInFlight);
        fake.SendEventCallCount.ShouldBe(40);
    }

    [Fact]
    public async Task A_cancellation_mid_batch_sends_nothing_more_and_dead_letters_nothing()
    {
        using var cts = new CancellationTokenSource();
        var sent = 0;
        var fake = new FakeHistorianGatewayClient
        {
            SendEventSource = async (_, ct) =>
            {
                if (Interlocked.Increment(ref sent) == 3) await cts.CancelAsync();
                await Task.Delay(10, ct);
                return new WriteAck { Success = true };
            },
        };
        using var writer = new GatewayAlarmHistorianWriter(fake, NullLogger<GatewayAlarmHistorianWriter>.Instance);

        var outcomes = await writer.WriteBatchAsync(Enumerable.Range(0, 40).Select(Evt).ToList(), cts.Token);

        outcomes.Count.ShouldBe(40);
        outcomes.ShouldNotContain(HistorianWriteOutcome.PermanentFail);
        outcomes.Count(o => o == HistorianWriteOutcome.RetryPlease).ShouldBeGreaterThan(0);
        sent.ShouldBeLessThanOrEqualTo(GatewayAlarmHistorianWriter.MaxSendsInFlight + 2, "no new send starts after the cancel");
    }

    [Fact]
    public async Task A_poison_event_dead_letters_alone()
    {
        var fake = new FakeHistorianGatewayClient
        {
            SendEventSource = (e, _) => e.Properties.TryGetValue("Message", out var m) && m == "m5"
                ? Task.FromException<WriteAck>(new RpcException(new Status(StatusCode.InvalidArgument, "bad")))
                : Task.FromResult(new WriteAck { Success = true }),
        };
        using var writer = new GatewayAlarmHistorianWriter(fake, NullLogger<GatewayAlarmHistorianWriter>.Instance);

        var outcomes = await writer.WriteBatchAsync(Enumerable.Range(0, 10).Select(Evt).ToList(), Ct);

        outcomes[5].ShouldBe(HistorianWriteOutcome.PermanentFail);
        outcomes.Where((_, i) => i != 5).ShouldAllBe(o => o == HistorianWriteOutcome.Ack);
    }

    private static void InterlockedMax(ref int target, int value)
    {
        int current;
        while ((current = Volatile.Read(ref target)) < value && Interlocked.CompareExchange(ref target, value, current) != current) { }
    }
}