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

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

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

/// <summary>
///     The Gateway events read drains its stream past the cap (within a budget) instead of abandoning it — an abandoned
///     gateway stream cancels the RPC and the gateway evicts its pooled historian session — and a caller cancellation on
///     the events / native-AtTime arms is not a backend failure.
/// </summary>
public sealed class GatewayEventDrainTests
{
    private static readonly DateTime T0 = new(2026, 10, 22, 8, 0, 0, DateTimeKind.Utc);
    private static CancellationToken Ct => TestContext.Current.CancellationToken;

    [Fact]
    public async Task A_capped_read_drains_the_stream_to_its_end_and_still_flags_truncation()
    {
        var fake = new FakeHistorianGatewayClient { Events = Series(50) };
        var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);

        var r = await ds.ReadEventsAsync(null, T0, T0.AddHours(1), maxEvents: 10, Ct);

        r.Events.Select(e => e.EventId).ShouldBe(Enumerable.Range(0, 10).Select(i => $"E{i:000}"));
        r.ContinuationPoint.ShouldNotBeNull();
        fake.EventStreamsCompleted.ShouldBe(1, "the stream must be consumed to its end, never broken out of");
        fake.EventStreamsAbandoned.ShouldBe(0);
        fake.EventsYieldedCount.ShouldBe(50);
        ds.EventStreamsAbandoned.ShouldBe(0);
    }

    [Fact]
    public async Task Past_the_drain_budget_the_stream_is_abandoned_but_the_page_is_unchanged()
    {
        var fake = new FakeHistorianGatewayClient { Events = Series(100) };
        var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance, eventDrainBudget: 5);

        var r = await ds.ReadEventsAsync(null, T0, T0.AddHours(1), maxEvents: 10, Ct);

        r.Events.Count.ShouldBe(10);
        r.ContinuationPoint.ShouldNotBeNull();
        fake.EventStreamsAbandoned.ShouldBe(1);
        fake.EventsYieldedCount.ShouldBe(10 + 5 + 1, "cap + budget, then the row that overran it");
        ds.EventStreamsAbandoned.ShouldBe(1);
    }

    [Fact]
    public async Task Budget_zero_is_the_previous_abandon_at_the_cap()
    {
        var fake = new FakeHistorianGatewayClient { Events = Series(20) };
        var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance, eventDrainBudget: 0);

        var r = await ds.ReadEventsAsync(null, T0, T0.AddHours(1), maxEvents: 3, Ct);

        r.Events.Count.ShouldBe(3);
        r.ContinuationPoint.ShouldNotBeNull();
        fake.EventsYieldedCount.ShouldBe(4);
        fake.EventStreamsAbandoned.ShouldBe(1);
    }

    [Fact]
    public async Task Foreign_source_rows_past_the_cap_are_drained_but_do_not_signal_truncation()
    {
        // The page is exactly the matching events; everything after it belongs to another source (an unfiltered gateway).
        var events = Series(3, "Pump1").Concat(Series(5, "Pump2", offsetSeconds: 10)).ToArray();
        var fake = new FakeHistorianGatewayClient { Events = events };
        var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);

        var r = await ds.ReadEventsAsync("Pump1", T0, T0.AddHours(1), maxEvents: 3, Ct);

        r.Events.Count.ShouldBe(3);
        r.ContinuationPoint.ShouldBeNull("no further MATCHING event exists");
        fake.EventStreamsCompleted.ShouldBe(1);

        // Positive control: one more matching event at the end does signal truncation.
        fake.Events = events.Append(Ev(99, "Pump1")).ToArray();
        (await ds.ReadEventsAsync("Pump1", T0, T0.AddHours(1), maxEvents: 3, Ct)).ContinuationPoint.ShouldNotBeNull();
    }

    [Fact]
    public async Task A_cancelled_events_read_rethrows_and_is_not_a_backend_failure()
    {
        var fake = new FakeHistorianGatewayClient { Events = Series(5) };
        var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
        using var cts = new CancellationTokenSource();
        await cts.CancelAsync();

        await Should.ThrowAsync<OperationCanceledException>(() => ds.ReadEventsAsync(null, T0, T0.AddHours(1), 3, cts.Token));
        var h = ds.GetHealthSnapshot();
        h.TotalFailures.ShouldBe(0);
        h.ConsecutiveFailures.ShouldBe(0);

        // Positive control: a real fault on the same arm IS a failure.
        fake.ThrowOnRead = new InvalidOperationException("boom");
        await Should.ThrowAsync<InvalidOperationException>(() => ds.ReadEventsAsync(null, T0, T0.AddHours(1), 3, Ct));
        ds.GetHealthSnapshot().TotalFailures.ShouldBe(1);
    }

    [Fact]
    public async Task A_cancelled_native_at_time_read_rethrows_and_is_not_a_backend_failure()
    {
        using var cts = new CancellationTokenSource();
        await cts.CancelAsync();
        var fake = new FakeHistorianGatewayClient { ThrowOnRead = new OperationCanceledException(cts.Token) };
        var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance, atTimeFromRaw: false);

        await Should.ThrowAsync<OperationCanceledException>(() => ds.ReadAtTimeAsync("T", [T0], cts.Token));
        ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);

        // Positive control: the same arm counts a real fault.
        fake.ThrowOnRead = new InvalidOperationException("boom");
        await Should.ThrowAsync<InvalidOperationException>(() => ds.ReadAtTimeAsync("T", [T0], Ct));
        ds.GetHealthSnapshot().TotalFailures.ShouldBe(1);
    }

    /// <summary>
    ///     The paged-vs-unpaged fuzz for the drain: random event series (bursts, tie clusters, foreign-source rows), random
    ///     windows, caps and drain budgets, paged the way the server pages a forward events read (resume at the last event's
    ///     time inclusive, skip the trailing ties emitted, ask cap + skip; a resumed page stalled on a tie cluster
    ///     over-fetches the cluster and pages within it) against a gateway-faithful fake (ascending time, ties in store
    ///     order, inclusive end, optionally no server-side source filter). The concatenated pages must equal the one-shot
    ///     uncapped read exactly — boundary ties included — and every stream must be consumed to its end whenever the
    ///     budget covers the overshoot.
    /// </summary>
    [Fact]
    public async Task Paged_events_equal_the_unpaged_read_and_drain_within_budget()
    {
        var iterations = int.TryParse(Environment.GetEnvironmentVariable("OTOPCUA_PAGING_FUZZ_ITERATIONS"), out var n) && n > 0
            ? Math.Min(n, 4000) : 1500;
        var occurred = new Dictionary<string, int>();
        void Count(string k) => occurred[k] = occurred.GetValueOrDefault(k) + 1;

        for (var seed = 0; seed < iterations; seed++)
        {
            var rng = new Random(seed);
            var store = RandomStore(rng, out var source);
            var cap = rng.Next(6) switch { 0 => 1, 1 => 2, 2 => 3, 3 => rng.Next(4, 12), 4 => rng.Next(12, 60), _ => 500 };
            var budget = rng.Next(4) switch { 0 => 0, 1 => rng.Next(1, 8), _ => GatewayHistorianDataSource.DefaultEventDrainBudget };
            var (start, end) = RandomWindow(rng, store);
            var tag = $"seed {seed} cap {cap} budget {budget} source {source ?? "*"} [{start:O} .. {end:O}]";

            var truthFake = new FakeHistorianGatewayClient { EventsSource = (s, e) => Window(store, s, e) };
            var truthDs = new GatewayHistorianDataSource(truthFake, NullLogger<GatewayHistorianDataSource>.Instance);
            var truth = (await truthDs.ReadEventsAsync(source, start, end, 0, Ct)).Events;
            truth.ShouldAllBe(e => source == null || e.SourceName == source, tag);

            var fake = new FakeHistorianGatewayClient { EventsSource = (s, e) => Window(store, s, e) };
            var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance, eventDrainBudget: budget);
            var paged = await PageAsync(ds, source, start, end, cap, tag);

            paged.Select(Key).ShouldBe(truth.Select(Key), $"paged ≠ unpaged: {tag}");
            if (budget >= GatewayHistorianDataSource.DefaultEventDrainBudget)
            {
                fake.EventStreamsAbandoned.ShouldBe(0, $"a stream was abandoned within budget: {tag}");
                Count("drained");
            }
            if (fake.EventStreamsAbandoned > 0) Count("abandoned past budget");
            if (truth.Count == 0) Count("empty");
            if (truth.Count > cap) Count("paged");
            if (truth.GroupBy(e => e.EventTimeUtc).Any(g => g.Count() > cap)) Count("tie cluster over the cap");
            if (source is not null && store.Any(e => e.SourceName != source)) Count("foreign rows");
        }

        foreach (var k in new[] { "drained", "abandoned past budget", "empty", "paged", "tie cluster over the cap", "foreign rows" })
            occurred.GetValueOrDefault(k).ShouldBeGreaterThan(0, $"the fuzz never produced '{k}' — it is vacuous for that case");
    }

    /// <summary>The server's forward events paging, against the data source (see the fuzz summary).</summary>
    private static async Task<List<HistoricalEvent>> PageAsync(
        IHistorianDataSource ds, string? source, DateTime start, DateTime end, int cap, string tag)
    {
        var all = new List<HistoricalEvent>();
        var skip = 0;
        var resumed = false;
        for (var page = 0; ; page++)
        {
            page.ShouldBeLessThan(10_000, $"paging must terminate: {tag}");
            var r = await ds.ReadEventsAsync(source, start, end, cap + skip, Ct);
            var truncated = r.ContinuationPoint is not null;
            var events = r.Events.ToList();
            if (resumed && skip > 0)
            {
                var i = 0;
                while (i < events.Count && i < skip && events[i].EventTimeUtc == start) i++;
                events.RemoveRange(0, i);
            }
            if (events.Count > cap) { events.RemoveRange(cap, events.Count - cap); truncated = true; }

            if (truncated && resumed && (events.Count == 0 || events[^1].EventTimeUtc == start))
            {
                // Oversized tie cluster: read it whole ([T, T] on the gateway's inclusive end) and page within it.
                var cluster = (await ds.ReadEventsAsync(source, start, start, 100_000, Ct)).Events
                    .Where(e => e.EventTimeUtc == start).ToList();
                var slice = cluster.Skip(skip).Take(cap).ToList();
                all.AddRange(slice);
                if (skip + slice.Count < cluster.Count) { skip += slice.Count; continue; }
                start = start.AddTicks(1);
                skip = 0;
                if (start > end) return all;
                continue;
            }

            all.AddRange(events);
            if (!truncated || events.Count == 0) return all;
            var boundary = events[^1].EventTimeUtc;
            var ties = 0;
            for (var i = events.Count - 1; i >= 0 && events[i].EventTimeUtc == boundary; i--) ties++;
            // A resumed page landing on the same boundary carries its earlier skip forward.
            skip = resumed && boundary == start ? skip + ties : ties;
            start = boundary;
            resumed = true;
        }
    }

    private static string Key(HistoricalEvent e) => $"{e.EventTimeUtc.Ticks}:{e.EventId}";

    private static List<HistorianEvent> RandomStore(Random rng, out string? source)
    {
        var count = rng.Next(5) == 0 ? 0 : rng.Next(1, 120);
        var mixed = rng.Next(3) == 0;
        source = rng.Next(3) == 0 ? null : "Src";
        var t = T0;
        var store = new List<HistorianEvent>();
        for (var i = 0; i < count; i++)
        {
            // Bursts: often several events share one timestamp; sometimes a big cluster.
            var step = rng.Next(6) switch { 0 or 1 => 0, 2 => 1, 3 => rng.Next(1, 1000), _ => rng.Next(1, 10_000_000) };
            t = t.AddTicks(step);
            var src = mixed && rng.Next(3) == 0 ? "Other" : "Src";
            store.Add(new HistorianEvent
            {
                Id = $"E{i:0000}",
                SourceName = src,
                EventTime = Timestamp.FromDateTime(t),
                ReceivedTime = Timestamp.FromDateTime(t),
            });
            if (rng.Next(25) == 0)
                for (var k = 0; k < rng.Next(5, 40); k++)
                    store.Add(new HistorianEvent
                    {
                        Id = $"E{i:0000}-{k:00}",
                        SourceName = src,
                        EventTime = Timestamp.FromDateTime(t),
                        ReceivedTime = Timestamp.FromDateTime(t),
                    });
        }
        return store;
    }

    private static (DateTime Start, DateTime End) RandomWindow(Random rng, List<HistorianEvent> store)
    {
        if (store.Count == 0 || rng.Next(6) == 0) return (T0, T0.AddDays(2));
        DateTime Pick() => rng.Next(3) == 0
            ? T0.AddTicks(rng.NextInt64(0, 2_000_000_000))
            : store[rng.Next(store.Count)].EventTime.ToDateTime();
        var a = Pick();
        var b = Pick();
        return a <= b ? (a, b) : (b, a);
    }

    /// <summary>The gateway's ReadEvents window: ascending time, ties in store order, inclusive at both ends.</summary>
    private static IEnumerable<HistorianEvent> Window(List<HistorianEvent> store, DateTime start, DateTime end) =>
        store.Where(e => e.EventTime.ToDateTime() >= start && e.EventTime.ToDateTime() <= end)
            .Select((e, i) => (e, i)).OrderBy(x => x.e.EventTime.ToDateTime()).ThenBy(x => x.i).Select(x => x.e);

    private static HistorianEvent[] Series(int count, string source = "Pump1", int offsetSeconds = 0) =>
        Enumerable.Range(0, count).Select(i => Ev(i + offsetSeconds, source, $"E{i:000}")).ToArray();

    private static HistorianEvent Ev(int seconds, string source, string? id = null) => new()
    {
        Id = id ?? $"X{seconds}",
        SourceName = source,
        EventTime = Timestamp.FromDateTime(T0.AddSeconds(seconds)),
        ReceivedTime = Timestamp.FromDateTime(T0.AddSeconds(seconds)),
    };
}