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

using Microsoft.Extensions.Logging;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;

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

/// <summary>
/// The StreamEvents resume cursor kept by <see cref="RealMxGatewayClient"/> (ScadaBridge#128).
/// The gateway emits control sentinels with <c>worker_sequence</c> 0 — the ReplayGap sentinel
/// on a resume that hit evicted events and (opt-in only) the StreamAttached ack. Neither is a
/// worker event, so neither may move the cursor; a ReplayGap instead anchors it at
/// <c>oldest_available_sequence - 1</c>, the gateway's documented resume formula.
/// </summary>
public class RealMxGatewayClientResumeCursorTests
{
    private static MxEvent DataChange(ulong seq, int itemHandle = 999) => new()
    {
        Family = MxEventFamily.OnDataChange,
        WorkerSequence = seq,
        ItemHandle = itemHandle,
    };

    private static MxEvent WriteComplete(ulong seq) => new()
    {
        Family = MxEventFamily.OnWriteComplete,
        WorkerSequence = seq,
    };

    /// <summary>The gateway's ReplayGap sentinel: family UNSPECIFIED, worker_sequence 0.</summary>
    private static MxEvent ReplayGapSentinel(ulong requestedAfter, ulong oldestAvailable) => new()
    {
        Family = MxEventFamily.Unspecified,
        WorkerSequence = 0,
        ReplayGap = new ReplayGap
        {
            RequestedAfterSequence = requestedAfter,
            OldestAvailableSequence = oldestAvailable,
        },
    };

    /// <summary>
    /// The StreamAttached ack as a 0.2.1-contract client decodes it: the
    /// <c>stream_attached</c> field is unknown to that contract, so all that is visible is a
    /// family-UNSPECIFIED event with worker_sequence 0.
    /// </summary>
    private static MxEvent StreamAttachedSentinel() => new()
    {
        Family = MxEventFamily.Unspecified,
        WorkerSequence = 0,
    };

    private static async IAsyncEnumerable<MxEvent> Stream(params MxEvent[] events)
    {
        foreach (var ev in events)
        {
            await Task.Yield();
            yield return ev;
        }
    }

    private static Task Run(RealMxGatewayClient client, params MxEvent[] events) =>
        client.RunEventLoopCoreAsync(Stream(events), _ => { }, CancellationToken.None);

    /// <summary>Real worker events of every family advance the cursor to the last sequence seen.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task RealEvents_AdvanceCursor()
    {
        var client = new RealMxGatewayClient(null);

        await Run(client, DataChange(1), WriteComplete(2), DataChange(3));

        Assert.Equal(3UL, client.ResumeCursor);
    }

    /// <summary>
    /// The issue's case: a resume that hits a gap with nothing retained (age eviction or a
    /// disabled ring) yields ONLY the ReplayGap sentinel. Resetting the cursor to 0 would turn
    /// the next reconnect into a fresh stream — no replay and no gap signal.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReplayGapSentinel_AloneOnStream_AnchorsCursorAtOldestMinusOne()
    {
        var client = new RealMxGatewayClient(null);
        await Run(client, DataChange(9), DataChange(10));

        // Ring empty: the gateway reports the next deliverable sequence (highest observed + 1).
        await Run(client, ReplayGapSentinel(requestedAfter: 10, oldestAvailable: 100));

        Assert.Equal(99UL, client.ResumeCursor);
    }

    /// <summary>A gap followed by the retained replay batch ends on the last replayed sequence.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReplayGapSentinel_FollowedByReplay_CursorEndsOnLastReplayedEvent()
    {
        var client = new RealMxGatewayClient(null);
        await Run(client, DataChange(10));

        await Run(client, ReplayGapSentinel(10, 50), DataChange(50), DataChange(51));

        Assert.Equal(51UL, client.ResumeCursor);
    }

    /// <summary>The (opt-in) StreamAttached ack carries worker_sequence 0 and must not move the cursor.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamAttachedSentinel_DoesNotResetCursor()
    {
        var client = new RealMxGatewayClient(null);
        await Run(client, DataChange(7));

        await Run(client, StreamAttachedSentinel());

        Assert.Equal(7UL, client.ResumeCursor);
    }

    /// <summary>The cursor only ever advances: a lower sequence never moves it back.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Cursor_NeverMovesBackwards()
    {
        var client = new RealMxGatewayClient(null);

        await Run(client, DataChange(20), DataChange(5), ReplayGapSentinel(1, 3));

        Assert.Equal(20UL, client.ResumeCursor);
    }

    /// <summary>
    /// Events in the evicted interval are unrecoverable; the loss is surfaced as a warning
    /// naming the interval, never hidden.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReplayGapSentinel_LogsWarningNamingTheLostInterval()
    {
        var provider = new CapturingLoggerProvider();
        using var factory = LoggerFactory.Create(b => b.AddProvider(provider));
        var client = new RealMxGatewayClient(factory);

        await Run(client, ReplayGapSentinel(10, 50));

        var warning = Assert.Single(provider.Entries, e => e.Level == LogLevel.Warning);
        Assert.Contains("10", warning.Message);
        Assert.Contains("50", warning.Message);
    }

    /// <summary>Sentinels never reach the value callback.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Sentinels_AreNotDeliveredAsValueUpdates()
    {
        var client = new RealMxGatewayClient(null);
        var updates = 0;

        await client.RunEventLoopCoreAsync(
            Stream(StreamAttachedSentinel(), ReplayGapSentinel(1, 5)), _ => updates++, CancellationToken.None);

        Assert.Equal(0, updates);
    }

    private sealed class CapturingLoggerProvider : ILoggerProvider
    {
        public List<(LogLevel Level, string Message)> Entries { get; } = new();

        public ILogger CreateLogger(string categoryName) => new Capturing(this);

        public void Dispose()
        {
        }

        private sealed class Capturing(CapturingLoggerProvider owner) : ILogger
        {
            public IDisposable? BeginScope<TState>(TState state)
                where TState : notnull => null;

            public bool IsEnabled(LogLevel logLevel) => true;

            public void Log<TState>(
                LogLevel logLevel, EventId eventId, TState state, Exception? exception,
                Func<TState, Exception?, string> formatter)
            {
                lock (owner.Entries)
                    owner.Entries.Add((logLevel, formatter(state, exception)));
            }
        }
    }
}