lmxopcua/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/PollGroupEngineBatchTests.cs

using System.Collections.Concurrent;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests;

/// <summary>
///     Guards for <c>PollGroupEngine</c>'s optional per-cycle batch callback: one raise per poll
///     cycle carrying exactly the items that changed in that cycle, raised after the per-tag
///     callbacks of the same cycle and only when non-empty. Drivers forward it to
///     <see cref="IBatchSubscribable.OnDataChangeBatch"/>.
/// </summary>
[Trait("Category", "Unit")]
public sealed class PollGroupEngineBatchTests
{
    private static readonly TimeSpan Interval = TimeSpan.FromMilliseconds(30);
    private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(5);

    /// <summary>
    ///     A reader over a mutable value map. Snapshots are returned in the requested reference
    ///     order (the engine's reader contract); <see cref="PollCount"/> lets a test await poll
    ///     cycles instead of sleeping a fixed wall-clock span.
    /// </summary>
    private sealed class FakeReader
    {
        private readonly Lock _gate = new();
        private readonly Dictionary<string, object?> _values = new(StringComparer.Ordinal);
        private readonly Dictionary<string, uint> _statusCodes = new(StringComparer.Ordinal);
        private int _pollCount;

        public FakeReader(params (string Reference, object? Value)[] seed)
        {
            foreach (var (reference, value) in seed) _values[reference] = value;
        }

        /// <summary>Gets the number of completed reader invocations.</summary>
        public int PollCount => Volatile.Read(ref _pollCount);

        /// <summary>
        ///     Overwrites the values a subsequent poll will observe. Applied under the same lock the
        ///     reader holds, so a multi-reference change is observed by ONE poll cycle rather than
        ///     being torn across two.
        /// </summary>
        public void Set(params (string Reference, object? Value)[] changes)
        {
            lock (_gate)
            {
                foreach (var (reference, value) in changes) _values[reference] = value;
            }
        }

        /// <summary>
        ///     Overwrites the status code a subsequent poll will report for a reference, leaving its
        ///     value untouched — the shape a device fault takes on the wire (#519).
        /// </summary>
        public void SetStatus(string reference, uint statusCode)
        {
            lock (_gate) _statusCodes[reference] = statusCode;
        }

        /// <summary>The reader delegate handed to the engine.</summary>
        public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(IReadOnlyList<string> references, CancellationToken ct)
        {
            var now = DateTime.UtcNow;
            var snapshots = new List<DataValueSnapshot>(references.Count);
            lock (_gate)
            {
                foreach (var reference in references)
                    snapshots.Add(new DataValueSnapshot(
                        _values.GetValueOrDefault(reference),
                        _statusCodes.GetValueOrDefault(reference),
                        now,
                        now));
            }
            Interlocked.Increment(ref _pollCount);
            return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(snapshots);
        }
    }

    private static async Task WaitForAsync(Func<bool> condition, string because)
    {
        var deadline = DateTime.UtcNow + WaitTimeout;
        while (DateTime.UtcNow < deadline)
        {
            if (condition()) return;
            await Task.Delay(10, TestContext.Current.CancellationToken);
        }
        throw new TimeoutException($"Timed out after {WaitTimeout.TotalSeconds}s waiting for: {because}");
    }

    /// <summary>
    ///     The initial-data cycle batches every subscribed tag; a later cycle in which exactly one
    ///     value moved batches exactly that one item — and the per-tag callback fired for the same
    ///     items, so both callbacks see identical content.
    /// </summary>
    [Fact]
    public async Task Batch_fires_once_per_cycle_with_only_changed_items()
    {
        var reader = new FakeReader(("a", 1), ("b", 2), ("c", 3));
        var perTag = new ConcurrentQueue<(string Reference, DataValueSnapshot Snapshot)>();
        var batches = new ConcurrentQueue<IReadOnlyList<DataChangeItem>>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            (_, reference, snapshot) => perTag.Enqueue((reference, snapshot)),
            minInterval: Interval,
            onBatch: (_, items) => batches.Enqueue(items));

        engine.Subscribe(["a", "b", "c"], Interval);

        await WaitForAsync(() => batches.Count >= 1, "the initial-data batch");
        batches.TryDequeue(out var initial).ShouldBeTrue();
        initial!.Count.ShouldBe(3);
        initial.Select(i => i.FullReference).ShouldBe(["a", "b", "c"]);
        perTag.Count.ShouldBe(3);

        perTag.Clear();
        batches.Clear();
        reader.Set(("b", 99));

        await WaitForAsync(() => batches.Count >= 1, "the batch for the single changed tag");
        batches.TryDequeue(out var changed).ShouldBeTrue();
        changed!.Count.ShouldBe(1);
        changed[0].FullReference.ShouldBe("b");
        changed[0].Snapshot.Value.ShouldBe(99);

        // The per-tag callback saw the same single item, with the same snapshot instance.
        perTag.Count.ShouldBe(1);
        perTag.TryDequeue(out var perTagItem).ShouldBeTrue();
        perTagItem.Reference.ShouldBe("b");
        perTagItem.Snapshot.ShouldBeSameAs(changed[0].Snapshot);
    }

    /// <summary>A cycle in which no value moved raises neither the batch nor any per-tag callback.</summary>
    [Fact]
    public async Task No_batch_when_nothing_changed()
    {
        var reader = new FakeReader(("a", 1), ("b", 2));
        var perTag = new ConcurrentQueue<string>();
        var batches = new ConcurrentQueue<IReadOnlyList<DataChangeItem>>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            (_, reference, _) => perTag.Enqueue(reference),
            minInterval: Interval,
            onBatch: (_, items) => batches.Enqueue(items));

        engine.Subscribe(["a", "b"], Interval);

        await WaitForAsync(() => batches.Count >= 1, "the initial-data batch");
        var pollsAfterInitial = reader.PollCount;
        perTag.Clear();
        batches.Clear();

        // Several further cycles over identical snapshots, plus a grace window for the callbacks
        // of the last counted poll to have run.
        await WaitForAsync(() => reader.PollCount >= pollsAfterInitial + 5, "five further poll cycles");
        await Task.Delay(50, TestContext.Current.CancellationToken);

        batches.ShouldBeEmpty();
        perTag.ShouldBeEmpty();
    }

    /// <summary>
    ///     Constructed without <c>onBatch</c>, the engine behaves exactly as before: per-tag
    ///     callbacks only, with no batch collection and no change to the raise pattern.
    /// </summary>
    [Fact]
    public async Task Batch_is_not_invoked_when_callback_absent()
    {
        var reader = new FakeReader(("a", 1), ("b", 2), ("c", 3));
        var perTag = new ConcurrentQueue<string>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            (_, reference, _) => perTag.Enqueue(reference),
            minInterval: Interval);

        engine.Subscribe(["a", "b", "c"], Interval);

        await WaitForAsync(() => perTag.Count >= 3, "the initial-data per-tag callbacks");
        perTag.Count.ShouldBe(3);

        reader.Set(("c", 42));
        await WaitForAsync(() => perTag.Count >= 4, "the per-tag callback for the changed tag");

        // Let further cycles run: nothing else changed, so the count must stay at 4.
        var polls = reader.PollCount;
        await WaitForAsync(() => reader.PollCount >= polls + 3, "three further poll cycles");
        await Task.Delay(50, TestContext.Current.CancellationToken);
        perTag.Count.ShouldBe(4);
        perTag.ToArray()[3].ShouldBe("c");
    }

    /// <summary>Batch items follow the reader's reference order, not any incidental sort order.</summary>
    [Fact]
    public async Task Batch_items_are_in_reader_order()
    {
        // Deliberately non-alphabetical subscription order: an alphabetical (or insertion-ordered
        // dictionary) collection would produce z/a/m in a different sequence.
        var reader = new FakeReader(("z", 1), ("a", 2), ("m", 3));
        var batches = new ConcurrentQueue<IReadOnlyList<DataChangeItem>>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            (_, _, _) => { },
            minInterval: Interval,
            onBatch: (_, items) => batches.Enqueue(items));

        engine.Subscribe(["z", "a", "m"], Interval);

        await WaitForAsync(() => batches.Count >= 1, "the initial-data batch");
        batches.TryDequeue(out var initial).ShouldBeTrue();
        initial!.Select(i => i.FullReference).ShouldBe(["z", "a", "m"]);

        // A cycle in which two of the three moved keeps the reader's relative order.
        batches.Clear();
        reader.Set(("m", 30), ("z", 10));

        await WaitForAsync(() => batches.Count >= 1, "the batch for the two changed tags");
        batches.TryDequeue(out var changed).ShouldBeTrue();
        changed!.Select(i => i.FullReference).ShouldBe(["z", "m"]);
    }

    /// <summary>
    ///     Both callbacks write into ONE ordered log, so the batch's position relative to the
    ///     per-tag raises of the same cycle is directly observable: in a cycle where three tags
    ///     moved, the batch marker must land strictly after all three per-tag markers.
    /// </summary>
    /// <remarks>
    ///     <para>This asserts ORDER, not final state — a batch raised mid-loop (after the first
    ///     change, say) still produces exactly the same four log entries, and is caught here only
    ///     because their sequence differs. A test that merely counted the entries, or compared the
    ///     two collections' contents, would pass against that implementation.</para>
    ///
    ///     <para>The thread assertion covers the other way the guarantee can be lost: a batch handed
    ///     to a detached <c>Task.Run</c> would usually still land last in the log (nothing else is
    ///     appended after it), so position alone does not catch it. The engine raises both callbacks
    ///     in one synchronous stretch of <c>PollOnceAsync</c> with no <c>await</c> between them, so a
    ///     correct implementation is on the same thread by construction and this can never flake red;
    ///     a detached raise lands on a different pool thread.</para>
    /// </remarks>
    [Fact]
    public async Task Batch_is_raised_after_every_per_tag_callback_of_the_same_cycle()
    {
        var reader = new FakeReader(("a", 1), ("b", 2), ("c", 3));
        var log = new ConcurrentQueue<string>();
        var perTagThreads = new ConcurrentQueue<int>();
        var batchThreads = new ConcurrentQueue<int>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            // The thread id is recorded BEFORE the log marker in both callbacks: the test awaits the
            // "batch" marker, so recording it second would let the assertions run against a
            // not-yet-enqueued thread id.
            (_, reference, _) =>
            {
                perTagThreads.Enqueue(Environment.CurrentManagedThreadId);
                log.Enqueue($"per-tag:{reference}");
            },
            minInterval: Interval,
            onBatch: (_, _) =>
            {
                batchThreads.Enqueue(Environment.CurrentManagedThreadId);
                log.Enqueue("batch");
            });

        engine.Subscribe(["a", "b", "c"], Interval);

        // Drain the initial-data cycle so the log holds exactly one cycle's markers.
        await WaitForAsync(() => log.Contains("batch"), "the initial-data batch");
        log.Clear();
        perTagThreads.Clear();
        batchThreads.Clear();

        // All three move under the reader's lock, so ONE poll cycle observes all three changes.
        reader.Set(("a", 10), ("b", 20), ("c", 30));
        await WaitForAsync(() => log.Contains("batch"), "the batch for the three changed tags");

        var ordered = log.ToArray();
        var batchIndex = Array.IndexOf(ordered, "batch");

        // Exactly one cycle's worth: three per-tag raises then the single batch, in that order.
        ordered.Count(e => e == "batch").ShouldBe(1);
        ordered.Take(batchIndex).ShouldBe(["per-tag:a", "per-tag:b", "per-tag:c"]);
        batchIndex.ShouldBe(3);

        // Raised inline on the polling thread, not posted to a detached task.
        batchThreads.Count.ShouldBe(1);
        batchThreads.Single().ShouldBe(perTagThreads.Last());
    }

    /// <summary>
    ///     A non-zero status code is a change in its own right (the value need not move), and the
    ///     snapshot rides into <see cref="DataChangeItem.Snapshot"/> untouched — the batch path
    ///     never re-derives or normalises quality (#519 at its origin).
    /// </summary>
    [Fact]
    public async Task Batch_carries_a_bad_status_snapshot_unchanged()
    {
        const uint BadStatus = 0x80010000u; // Bad_UnexpectedError

        var reader = new FakeReader(("a", 1), ("b", 2));
        var perTag = new ConcurrentQueue<(string Reference, DataValueSnapshot Snapshot)>();
        var batches = new ConcurrentQueue<IReadOnlyList<DataChangeItem>>();

        await using var engine = new PollGroupEngine(
            reader.ReadAsync,
            (_, reference, snapshot) => perTag.Enqueue((reference, snapshot)),
            minInterval: Interval,
            onBatch: (_, items) => batches.Enqueue(items));

        engine.Subscribe(["a", "b"], Interval);

        await WaitForAsync(() => batches.Count >= 1, "the initial-data batch");
        perTag.Clear();
        batches.Clear();

        // The device faults on "a": same value, bad quality.
        reader.SetStatus("a", BadStatus);

        await WaitForAsync(() => batches.Count >= 1, "the batch for the newly-bad tag");
        batches.TryDequeue(out var bad).ShouldBeTrue();
        bad!.Count.ShouldBe(1);
        bad[0].FullReference.ShouldBe("a");
        bad[0].Snapshot.StatusCode.ShouldBe(BadStatus);
        bad[0].Snapshot.Value.ShouldBe(1); // value unchanged — the status alone drove the raise

        // The per-tag callback saw the identical snapshot instance.
        perTag.TryDequeue(out var perTagItem).ShouldBeTrue();
        perTagItem.Reference.ShouldBe("a");
        perTagItem.Snapshot.ShouldBeSameAs(bad[0].Snapshot);
    }
}