ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/BatchSeamPrimitiveTests.cs

using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;

namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;

/// <summary>
/// WP2.1b — the primitives the batch seam is built from, tested in isolation because the
/// OPC Foundation Session/Subscription types they drive cannot be faked without a live
/// server: monitored-item shard placement, the apply-serialization gate, the bounded
/// pipeline behind the MxGateway supervisory advise, the alarm-stream union prefix, and
/// the tag-resolution backoff step.
/// </summary>
public class BatchSeamPrimitiveTests
{
    // ── Shard placement ──

    [Fact]
    public void Sharding_SplitsItemsAtTheBudget()
    {
        // 12,001 items at the 5,000 default → 3 shards (memo §2 sizing example).
        Assert.Equal(3, MonitoredItemShardPlanner.ShardCountFor(12_001, 5_000));
        // 37,500 tags → 8 shards.
        Assert.Equal(8, MonitoredItemShardPlanner.ShardCountFor(37_500, 5_000));

        var placement = MonitoredItemShardPlanner.Plan([], 5_000, 12_001);
        Assert.Equal(12_001, placement.Count);
        Assert.Equal(0, placement[0]);
        Assert.Equal(0, placement[4_999]);
        Assert.Equal(1, placement[5_000]);
        Assert.Equal(2, placement[10_000]);
        Assert.Equal(2, placement[12_000]);
    }

    [Fact]
    public void Sharding_FillsTheFirstShardWithFreeCapacityBeforeCreatingOne()
    {
        // Shard 0 full, shard 1 has one free slot: the next two items fill shard 1 then
        // open shard 2 — the "first shard with free capacity, else a new shard" policy,
        // which is also what makes an emptied-and-deleted shard's capacity reusable.
        var placement = MonitoredItemShardPlanner.Plan([5_000, 4_999], 5_000, 2);
        Assert.Equal([1, 2], placement);
    }

    [Fact]
    public void Sharding_NonPositiveBudgetDegradesToOneItemPerShard()
    {
        Assert.Equal([0, 1, 2], MonitoredItemShardPlanner.Plan([], 0, 3));
    }

    // ── Apply serialization ──

    [Fact]
    public async Task ApplyGate_NeverRunsTwoSectionsConcurrently()
    {
        var gate = new AsyncSerialGate();
        var inFlight = 0;
        var maxObserved = 0;

        var tasks = Enumerable.Range(0, 32).Select(_ => gate.RunAsync(async () =>
        {
            var now = Interlocked.Increment(ref inFlight);
            InterlockedMax(ref maxObserved, now);
            await Task.Delay(5);
            Interlocked.Decrement(ref inFlight);
        })).ToArray();

        await Task.WhenAll(tasks);

        Assert.Equal(1, maxObserved);
    }

    [Fact]
    public async Task ApplyGate_ReleasesWhenASectionThrows()
    {
        var gate = new AsyncSerialGate();

        await Assert.ThrowsAsync<InvalidOperationException>(() =>
            gate.RunAsync(() => throw new InvalidOperationException("apply failed")));

        // The gate must still admit the next caller — a faulted ApplyChanges must not
        // wedge every later subscribe/unsubscribe on this client.
        var ran = false;
        await gate.RunAsync(() =>
        {
            ran = true;
            return Task.CompletedTask;
        });
        Assert.True(ran);
    }

    // ── Bounded pipeline (MxGateway supervisory advise) ──

    [Fact]
    public async Task BulkPipeline_KeepsInFlightCountWithinTheWindow()
    {
        const int parallelism = 16;
        var inFlight = 0;
        var maxObserved = 0;
        var items = Enumerable.Range(0, 200).ToList();

        var results = await BulkPipeline.RunAsync(
            items,
            parallelism,
            async (item, _) =>
            {
                var now = Interlocked.Increment(ref inFlight);
                InterlockedMax(ref maxObserved, now);
                await Task.Delay(2);
                Interlocked.Decrement(ref inFlight);
                return item * 2;
            },
            (item, _) => -item);

        Assert.Equal(items.Count, results.Length);
        Assert.Equal(items.Select(i => i * 2), results);
        Assert.InRange(maxObserved, 1, parallelism);
    }

    [Fact]
    public async Task BulkPipeline_CapturesPerItemFaultsWithoutAbortingTheBatch()
    {
        var results = await BulkPipeline.RunAsync(
            [1, 2, 3],
            4,
            (item, _) => item == 2
                ? throw new InvalidOperationException("advise failed")
                : Task.FromResult(item),
            (item, _) => -item);

        Assert.Equal([1, -2, 3], results);
    }

    // ── Alarm-stream union prefix ──

    [Theory]
    [InlineData(new[] { "Area1.Tank1", "Area1.Tank2" }, "Area1.Tank")]
    [InlineData(new[] { "Area1.Tank1" }, "Area1.Tank1")]
    [InlineData(new[] { "Area1.Tank1", "Area2.Tank1" }, "Area")]
    [InlineData(new[] { "Plant.A", "Zone.B" }, "")]
    [InlineData(new[] { "Area1.Tank1", "" }, "")]
    [InlineData(new string[0], "")]
    public void AlarmPrefix_IsTheLongestCommonPrefix(string[] sources, string expected)
    {
        Assert.Equal(expected, AlarmFilterPrefix.LongestCommonPrefix(sources));
    }

    [Fact]
    public void AlarmPrefix_CoversOnlySourcesUnderTheLivePrefix()
    {
        Assert.True(AlarmFilterPrefix.Covers("Area1.", "Area1.Tank1"));
        Assert.False(AlarmFilterPrefix.Covers("Area1.", "Area2.Tank1"));
        // A gateway-wide stream covers everything.
        Assert.True(AlarmFilterPrefix.Covers("", "Anything.At.All"));
    }

    // ── Tag-resolution backoff ──

    [Fact]
    public void Backoff_DoublesPerFailedRoundAndCapsAtTheMaximum()
    {
        var floor = TimeSpan.FromSeconds(10);
        var max = TimeSpan.FromMinutes(5);

        var sequence = new List<TimeSpan>();
        var current = floor;
        for (var round = 0; round < 8; round++)
        {
            current = DataConnectionActor.NextTagResolutionInterval(current, floor, max);
            sequence.Add(current);
        }

        Assert.Equal(
        [
            TimeSpan.FromSeconds(20),
            TimeSpan.FromSeconds(40),
            TimeSpan.FromSeconds(80),
            TimeSpan.FromSeconds(160),
            TimeSpan.FromSeconds(300),
            TimeSpan.FromSeconds(300),
            TimeSpan.FromSeconds(300),
            TimeSpan.FromSeconds(300),
        ], sequence);
    }

    [Fact]
    public void Backoff_MisconfiguredCeilingBelowFloorDegradesToAFixedInterval()
    {
        var floor = TimeSpan.FromSeconds(10);
        var next = DataConnectionActor.NextTagResolutionInterval(floor, floor, TimeSpan.FromSeconds(1));
        Assert.Equal(floor, next);
    }

    private static void InterlockedMax(ref int target, int value)
    {
        int seen;
        do
        {
            seen = Volatile.Read(ref target);
            if (value <= seen)
                return;
        }
        while (Interlocked.CompareExchange(ref target, value, seen) != seen);
    }
}