ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs

using System.Diagnostics.Metrics;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport;
using ZB.MOM.WW.ScadaBridge.Transport.Import;

namespace ZB.MOM.WW.ScadaBridge.Transport.Tests.Import;

/// <summary>
/// Register row 53: <c>MaxConcurrentImportSessions</c> had no observability — nothing
/// counted open sessions and nothing logged a rejection at the cap, so the register-row-43
/// session leak could only be inferred from rotating test failures. These tests pin the
/// two signals that close that gap: the
/// <c>scadabridge.central.transport.import_sessions.open</c> gauge and the Warning-level
/// log emitted when <see cref="BundleSessionStore.Open"/> rejects at the cap.
/// </summary>
public sealed class BundleSessionObservabilityTests
{
    private const string OpenSessionsGauge = "scadabridge.central.transport.import_sessions.open";
    private const string RejectedSessionsCounter = "scadabridge.central.transport.import_sessions.rejected";

    // ------------------------------------------------------------------
    // Warning log on cap rejection
    // ------------------------------------------------------------------

    [Fact]
    public void Open_AtCap_LogsWarningWithCountAndCap()
    {
        // The rejection used to be visible only as the exception the caller saw. The
        // operator-facing signal must name BOTH the current count and the cap — that
        // pair is what distinguishes genuine concurrency from a leak.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
        var logger = new CapturingLogger();
        var sut = new BundleSessionStore(opts, clock, logger);
        var future = clock.GetUtcNow().AddMinutes(30);

        for (var i = 0; i < 3; i++)
        {
            sut.Open(SessionExpiringAt(future));
        }

        Assert.Throws<InvalidOperationException>(() => sut.Open(SessionExpiringAt(future)));

        var warning = Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning);
        Assert.Contains("Bundle import session rejected", warning.Message, StringComparison.Ordinal);
        Assert.Contains("3 of 3", warning.Message, StringComparison.Ordinal);
        Assert.Contains("MaxConcurrentImportSessions", warning.Message, StringComparison.Ordinal);
    }

    [Fact]
    public void Open_UnderCap_LogsNothing()
    {
        // Negative control: the Warning must be a cap signal, not per-open noise.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
        var logger = new CapturingLogger();
        var sut = new BundleSessionStore(opts, clock, logger);

        sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));

        Assert.Empty(logger.Entries);
    }

    [Fact]
    public void Open_AtCapOfExpiredSessions_ReclaimsAndLogsNothing()
    {
        // A store that is full only of stale entries reclaims and admits the caller —
        // that is not a cap hit and must not produce a Warning, or the signal would fire
        // routinely on any node whose sweep has not yet run.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 });
        var logger = new CapturingLogger();
        var sut = new BundleSessionStore(opts, clock, logger);

        sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
        sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
        clock.Advance(TimeSpan.FromMinutes(10));

        sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));

        Assert.Empty(logger.Entries);
    }

    // ------------------------------------------------------------------
    // Rejection counter (register row 66)
    // ------------------------------------------------------------------

    [Fact]
    public void Open_AtCap_IncrementsTheRejectionCounter()
    {
        // Row 53 shipped the gauge + Warning; a log line is not alertable. This counter is
        // the series an operator can page on, so it must move on a genuine cap hit.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
        var future = clock.GetUtcNow().AddMinutes(30);

        for (var i = 0; i < 3; i++)
        {
            sut.Open(SessionExpiringAt(future));
        }

        var delta = ReadRejectionCounter(() =>
            Assert.Throws<InvalidOperationException>(() => sut.Open(SessionExpiringAt(future))));

        Assert.Equal(1, delta);
    }

    [Fact]
    public void Open_RepeatedCapHits_AccumulateOnTheRejectionCounter()
    {
        // A counter that saturated at 1 would hide exactly the case it exists for: a
        // sustained rejection rate is what distinguishes real contention from a one-off.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 1 });
        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
        var future = clock.GetUtcNow().AddMinutes(30);
        sut.Open(SessionExpiringAt(future));

        var delta = ReadRejectionCounter(() =>
        {
            for (var i = 0; i < 3; i++)
            {
                Assert.Throws<InvalidOperationException>(() => sut.Open(SessionExpiringAt(future)));
            }
        });

        Assert.Equal(3, delta);
    }

    [Fact]
    public void Open_UnderCap_DoesNotIncrementTheRejectionCounter()
    {
        // Negative control: the counter is a rejection signal, not an open-session signal.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);

        var delta = ReadRejectionCounter(() =>
            sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30))));

        Assert.Equal(0, delta);
    }

    [Fact]
    public void Open_AtCapOfExpiredSessions_ReclaimsWithoutIncrementingTheRejectionCounter()
    {
        // The load-bearing negative control. The cap check runs TWICE — once before
        // EvictExpired and once after — and only the post-eviction check is a rejection.
        // Counting the first would make the series fire routinely on any node whose sweep
        // has not yet run, turning a page-worthy signal into background noise.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 });
        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);

        sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
        sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
        clock.Advance(TimeSpan.FromMinutes(10));

        var delta = ReadRejectionCounter(() =>
            sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30))));

        Assert.Equal(0, delta);
    }

    [Fact]
    public void Open_ReopeningAnExistingSessionIdAtCap_DoesNotIncrementTheRejectionCounter()
    {
        // Re-Opening an id already in the store skips the cap block entirely and overwrites
        // in place. It consumes no new slot and turns nobody away, so it must not count.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 });
        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
        var future = clock.GetUtcNow().AddMinutes(30);

        var first = SessionExpiringAt(future);
        sut.Open(first);
        sut.Open(SessionExpiringAt(future));

        // Store is now exactly at the cap of 2.
        var delta = ReadRejectionCounter(() => sut.Open(first));

        Assert.Equal(0, delta);
    }

    // ------------------------------------------------------------------
    // OpenSessionCount / gauge
    // ------------------------------------------------------------------

    [Fact]
    public void OpenSessionCount_ExcludesExpiredButUnsweptSessions()
    {
        // The eviction sweep runs once a minute, so a raw dictionary count would report
        // the store as pinned at the cap for up to a sweep interval after a burst had
        // actually drained — exactly the false signal this instrument exists to rule out.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var sut = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);

        sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
        var live = SessionExpiringAt(clock.GetUtcNow().AddMinutes(60));
        sut.Open(live);

        Assert.Equal(2, sut.OpenSessionCount);

        clock.Advance(TimeSpan.FromMinutes(10));

        // The short session has expired but nothing has swept it yet.
        Assert.Equal(1, sut.OpenSessionCount);
    }

    [Fact]
    public async Task EvictionService_RegistersTheGaugeProvider_AndClearsItOnStop()
    {
        // The eviction service is Transport's only lifecycle seam, so it owns the
        // process-global gauge slot. The gauge is read as a delta against a baseline
        // because the instrument is a process-wide static.
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var store = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
        var service = new BundleSessionEvictionService(store, NullLogger<BundleSessionEvictionService>.Instance);

        var baseline = ReadGauge();

        await service.StartAsync(CancellationToken.None);
        try
        {
            store.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
            store.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));

            Assert.Equal(2, ReadGauge());
        }
        finally
        {
            await service.StopAsync(CancellationToken.None);
        }

        // After the clear the gauge falls back to reporting 0, not through a dead store.
        Assert.Equal(0, ReadGauge());
        Assert.Equal(0, baseline);
    }

    [Fact]
    public async Task EvictionService_StopDoesNotClobberANewerInstancesProvider()
    {
        // Identity-checked compare-and-clear: a late stop of a superseded instance must
        // leave a newer registrant's provider in place (mirrors the StoreAndForward
        // queue-depth gauge's contract).
        var clock = new TestClock(DateTimeOffset.UtcNow);
        var oldStore = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
        var newStore = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
        var oldService = new BundleSessionEvictionService(oldStore, NullLogger<BundleSessionEvictionService>.Instance);
        var newService = new BundleSessionEvictionService(newStore, NullLogger<BundleSessionEvictionService>.Instance);

        await oldService.StartAsync(CancellationToken.None);
        await newService.StartAsync(CancellationToken.None);
        try
        {
            newStore.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));

            // Late stop of the superseded instance.
            await oldService.StopAsync(CancellationToken.None);

            Assert.Equal(1, ReadGauge());
        }
        finally
        {
            await newService.StopAsync(CancellationToken.None);
        }

        Assert.Equal(0, ReadGauge());
    }

    // ------------------------------------------------------------------
    // Helpers
    // ------------------------------------------------------------------

    /// <summary>
    /// Forces one synchronous observation of the open-import-sessions gauge and returns
    /// the measured value. Mirrors the MeterListener helper used by the Communication
    /// and StoreAndForward telemetry suites.
    /// </summary>
    /// <returns>The gauge's current value, or 0 if it never reported.</returns>
    private static long ReadGauge()
    {
        long observed = 0;
        using var listener = new MeterListener();
        listener.InstrumentPublished = (instrument, l) =>
        {
            if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName
                && instrument.Name == OpenSessionsGauge)
            {
                l.EnableMeasurementEvents(instrument);
            }
        };
        listener.SetMeasurementEventCallback<long>((_, measurement, _, _) => observed = measurement);
        listener.Start();
        listener.RecordObservableInstruments();
        return observed;
    }

    /// <summary>
    /// Runs <paramref name="act"/> with a listener attached to the import-session rejection
    /// counter and returns the number of rejections recorded DURING that action.
    /// </summary>
    /// <remarks>
    /// Deliberately a delta over a scoped action rather than an absolute read. The counter
    /// is a process-global static, so an absolute assertion would inherit the register
    /// row 32 flake that makes <c>QueueDepthGaugeTests</c> fail only under the full
    /// concurrent suite. A counter measured this way is immune: the listener exists only
    /// for the duration of <paramref name="act"/>, and every increment is driven
    /// synchronously by this test's own <c>Open</c> call on its own store instance, so a
    /// concurrently-running test class cannot contribute a measurement to this window.
    /// </remarks>
    /// <param name="act">The action to measure.</param>
    /// <returns>Rejections counted while <paramref name="act"/> ran.</returns>
    private static long ReadRejectionCounter(Action act)
    {
        long delta = 0;
        using var listener = new MeterListener();
        listener.InstrumentPublished = (instrument, l) =>
        {
            if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName
                && instrument.Name == RejectedSessionsCounter)
            {
                l.EnableMeasurementEvents(instrument);
            }
        };
        listener.SetMeasurementEventCallback<long>((_, measurement, _, _) => delta += measurement);
        listener.Start();
        act();
        return delta;
    }

    private static BundleSession SessionExpiringAt(DateTimeOffset expiresAt) => new()
    {
        SessionId = Guid.NewGuid(),
        Manifest = StubManifest(),
        DecryptedContent = Array.Empty<byte>(),
        ExpiresAt = expiresAt,
    };

    private static BundleManifest StubManifest() => new(
        BundleFormatVersion: 1,
        SchemaVersion: "1.0",
        CreatedAtUtc: DateTimeOffset.UnixEpoch,
        SourceEnvironment: "test",
        ExportedBy: "t",
        ScadaBridgeVersion: "1",
        ContentHash: "0",
        Encryption: null,
        Summary: new BundleSummary(0, 0, 0, 0, 0, 0, 0, 0),
        Contents: Array.Empty<ManifestContentEntry>());

    /// <summary>Minimal mutable-clock <see cref="TimeProvider"/> for deterministic expiry.</summary>
    private sealed class TestClock : TimeProvider
    {
        private DateTimeOffset _now;
        public TestClock(DateTimeOffset start) { _now = start; }
        public override DateTimeOffset GetUtcNow() => _now;
        public void Advance(TimeSpan by) => _now = _now.Add(by);
    }

    /// <summary>Captures formatted log entries so the cap-rejection Warning can be asserted.</summary>
    private sealed class CapturingLogger : ILogger<BundleSessionStore>
    {
        public List<(LogLevel Level, string Message)> Entries { get; } = [];

        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)
        {
            Entries.Add((logLevel, formatter(state, exception)));
        }
    }
}