ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.Transport.IntegrationTests/Import/FleetWideUnlockThrottleTests.cs

using Microsoft.Data.Sqlite;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using NSubstitute;
using NSubstitute.ExceptionExtensions;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
using ZB.MOM.WW.ScadaBridge.Transport.Import;

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

/// <summary>
/// Register row 52: the bundle-unlock throttle's counters used to be per-node AND
/// per-process, so the effective fleet cap was N x <c>MaxUnlockAttemptsPerIpPerHour</c>
/// per client and a node restart silently cleared the window. These tests stand up TWO
/// independent limiter instances — the stand-in for central-a and central-b — over ONE
/// shared relational store and prove they enforce a single combined window, plus the
/// fail-open behaviour a store outage must produce.
/// </summary>
public sealed class FleetWideUnlockThrottleTests : IDisposable
{
    private readonly SqliteConnection _connection;
    private readonly ServiceProvider _provider;

    /// <summary>Initializes a new instance of the <c>FleetWideUnlockThrottleTests</c> class.</summary>
    public FleetWideUnlockThrottleTests()
    {
        // One open in-memory SQLite connection for the fixture's lifetime — the database
        // lives exactly as long as the connection, and both "nodes" resolve scoped
        // contexts over it, which is what makes the store genuinely shared.
        _connection = new SqliteConnection("DataSource=:memory:");
        _connection.Open();

        var services = new ServiceCollection();
        services.AddDbContext<ScadaBridgeDbContext>(opts => opts.UseSqlite(_connection));
        services.AddScoped<IBundleUnlockAttemptRepository, BundleUnlockAttemptRepository>();
        _provider = services.BuildServiceProvider();

        using var scope = _provider.CreateScope();
        scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>().Database.EnsureCreated();
    }

    /// <summary>Disposes the service provider and the shared connection.</summary>
    public void Dispose()
    {
        _provider.Dispose();
        _connection.Dispose();
    }

    /// <summary>
    /// Builds a limiter over the shared store, standing in for one central node's
    /// singleton. Each call returns a fully independent instance (its own process-local
    /// fallback buckets, its own prune claim) — exactly the isolation the two real nodes
    /// have.
    /// </summary>
    /// <param name="clock">Clock the node observes.</param>
    /// <returns>A limiter whose window is the shared store.</returns>
    private BundleUnlockRateLimiter NewNode(TimeProvider clock) => new(
        clock,
        BundleUnlockRateLimiter.DefaultWindow,
        new SqlBundleUnlockAttemptWindow(
            _provider.GetRequiredService<IServiceScopeFactory>(), clock),
        logger: null);

    /// <summary>Verifies that two nodes sharing one store enforce a single combined throttle window.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task TwoNodesSharingOneStore_EnforceASingleCombinedWindow()
    {
        // The headline contract: attempts spread across both central nodes count
        // against ONE budget. Before this, five attempts per node were both allowed.
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var nodeA = NewNode(clock);
        var nodeB = NewNode(clock);

        for (var i = 0; i < 5; i++)
        {
            Assert.True(await nodeA.TryRegisterAttemptAsync("10.0.0.1", 10));
            Assert.True(await nodeB.TryRegisterAttemptAsync("10.0.0.1", 10));
        }

        // The combined budget of 10 is now spent — on EITHER node.
        Assert.False(await nodeA.TryRegisterAttemptAsync("10.0.0.1", 10));
        Assert.False(await nodeB.TryRegisterAttemptAsync("10.0.0.1", 10));
    }

    /// <summary>Verifies that denied attempts do not consume any of the throttle budget.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DeniedAttempts_BurnNoBudget()
    {
        // Register row 44's contract, preserved through the move to the shared store: a
        // rejected attempt is not recorded, so hammering cannot roll the window forward
        // and extend the client's own lockout.
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var nodeA = NewNode(clock);

        for (var i = 0; i < 3; i++)
        {
            Assert.True(await nodeA.TryRegisterAttemptAsync("10.0.0.1", 3));
        }

        for (var i = 0; i < 5; i++)
        {
            Assert.False(await nodeA.TryRegisterAttemptAsync("10.0.0.1", 3));
        }

        // Exactly the 3 permitted attempts were persisted — the 5 denials added nothing.
        using var scope = _provider.CreateScope();
        var repo = scope.ServiceProvider.GetRequiredService<IBundleUnlockAttemptRepository>();
        var recorded = await repo.CountAttemptsSinceAsync(
            "10.0.0.1", clock.GetUtcNow().UtcDateTime.AddHours(-1));
        Assert.Equal(3, recorded);
    }

    /// <summary>Verifies that the throttle window survives a restart.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task TheWindowIsRestartDurable()
    {
        // A node restart used to clear the window outright. A brand-new limiter instance
        // (the restarted process) must still see the budget the old one spent.
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var before = NewNode(clock);

        for (var i = 0; i < 3; i++)
        {
            Assert.True(await before.TryRegisterAttemptAsync("10.0.0.1", 3));
        }

        var afterRestart = NewNode(clock);
        Assert.False(await afterRestart.TryRegisterAttemptAsync("10.0.0.1", 3));
    }

    /// <summary>Verifies that the throttle window slides so an aged-out budget is released.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task TheWindowSlides_SoAnAgedOutBudgetIsReleased()
    {
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var node = NewNode(clock);

        for (var i = 0; i < 3; i++)
        {
            Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 3));
        }

        Assert.False(await node.TryRegisterAttemptAsync("10.0.0.1", 3));

        clock.Advance(TimeSpan.FromMinutes(61));
        Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 3));
    }

    /// <summary>Verifies that throttle buckets are isolated per client.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task BucketsAreIsolatedPerClient()
    {
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var node = NewNode(clock);

        Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 1));
        Assert.False(await node.TryRegisterAttemptAsync("10.0.0.1", 1));

        // A different client — and the shared unattributed bucket — are unaffected.
        Assert.True(await node.TryRegisterAttemptAsync("10.0.0.2", 1));
        Assert.True(await node.TryRegisterAttemptAsync("<unattributed>", 1));
    }

    /// <summary>Verifies that a store failure fails open and does not block the unlock.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StoreFailure_FailsOpen_AndDoesNotBlockTheUnlock()
    {
        // Availability of legitimate imports outranks a best-effort rate cap: a database
        // error must permit the attempt, not take the import surface down with the DB.
        var window = Substitute.For<IBundleUnlockAttemptWindow>();
        window
            .TryRegisterAttemptAsync(
                Arg.Any<string>(), Arg.Any<int>(), Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>())
            .ThrowsAsync(new InvalidOperationException("database unavailable"));

        var limiter = new BundleUnlockRateLimiter(
            TimeProvider.System, BundleUnlockRateLimiter.DefaultWindow, window, logger: null);

        for (var i = 0; i < 20; i++)
        {
            Assert.True(await limiter.TryRegisterAttemptAsync("10.0.0.1", 1));
        }
    }

    /// <summary>Verifies that caller cancellation is not swallowed by the fail-open path.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task CallerCancellation_IsNotSwallowedByTheFailOpenPath()
    {
        // Fail-open must not hide the caller's own cancellation — that would turn an
        // aborted request into a silently permitted attempt.
        var window = Substitute.For<IBundleUnlockAttemptWindow>();
        window
            .TryRegisterAttemptAsync(
                Arg.Any<string>(), Arg.Any<int>(), Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>())
            .ThrowsAsync(new OperationCanceledException());

        var limiter = new BundleUnlockRateLimiter(
            TimeProvider.System, BundleUnlockRateLimiter.DefaultWindow, window, logger: null);

        using var cts = new CancellationTokenSource();
        await cts.CancelAsync();

        await Assert.ThrowsAnyAsync<OperationCanceledException>(
            () => limiter.TryRegisterAttemptAsync("10.0.0.1", 10, cts.Token));
    }

    /// <summary>Verifies that without a shared window the limiter still enforces a process-local cap.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WithoutASharedWindow_TheLimiterStillEnforcesAProcessLocalCap()
    {
        // Hosts that register Transport without a shared window keep the pre-shared-window
        // behaviour rather than losing the cap entirely.
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var limiter = new BundleUnlockRateLimiter(clock, BundleUnlockRateLimiter.DefaultWindow);

        Assert.False(limiter.IsFleetWide);
        Assert.True(await limiter.TryRegisterAttemptAsync("10.0.0.1", 1));
        Assert.False(await limiter.TryRegisterAttemptAsync("10.0.0.1", 1));
    }

    /// <summary>Verifies that aged rows are pruned opportunistically once the prune interval has elapsed.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AgedRowsArePrunedOpportunistically_OnceThePruneIntervalHasElapsed()
    {
        // No background service: the prune rides the unlock path, claimed at most once per
        // PruneMinInterval. Rows are retained for PruneRetentionWindows x window so a prune
        // racing a count can never delete a row the count still needs.
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var node = NewNode(clock);

        Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 10));

        // Move past both the retention horizon (2 hours) and the prune cadence.
        clock.Advance(TimeSpan.FromHours(3));
        Assert.True(await node.TryRegisterAttemptAsync("10.0.0.2", 10));

        using var scope = _provider.CreateScope();
        var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
        var remaining = await ctx.BundleUnlockAttempts.AsNoTracking().ToListAsync();

        // The aged row is gone; the one just written by the pruning attempt survives.
        Assert.Equal("10.0.0.2", Assert.Single(remaining).ClientKey);
    }

    /// <summary>Verifies that recent rows survive a prune pass.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task RecentRowsSurviveAPrunePass()
    {
        // Negative control for the prune horizon: an attempt still inside the retention
        // window must not be deleted, or the throttle would forget live budget.
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var node = NewNode(clock);

        Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 10));

        // Past the prune cadence (15 min) but inside the 2-hour retention horizon.
        clock.Advance(TimeSpan.FromMinutes(30));
        Assert.True(await node.TryRegisterAttemptAsync("10.0.0.2", 10));

        using var scope = _provider.CreateScope();
        var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
        Assert.Equal(2, await ctx.BundleUnlockAttempts.AsNoTracking().CountAsync());
    }

    /// <summary>Verifies that the scheduled prune clears aged rows even when there is no unlock traffic.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ScheduledPrune_ClearsAgedRows_WithNoUnlockTrafficAtAll()
    {
        // Register row 65: the prune used to ride the unlock path ONLY, so a cluster that
        // stopped importing entirely left its last rows in place indefinitely. The
        // periodic sweep now drives PruneAsync directly — no second unlock attempt is made
        // anywhere in this test, which is the whole point.
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var window = new SqlBundleUnlockAttemptWindow(
            _provider.GetRequiredService<IServiceScopeFactory>(), clock);

        Assert.True(await window.TryRegisterAttemptAsync(
            "10.0.0.1", 10, BundleUnlockRateLimiter.DefaultWindow));

        // Past both the 2-hour retention horizon and the 15-minute prune cadence.
        clock.Advance(TimeSpan.FromHours(3));
        await window.PruneAsync(BundleUnlockRateLimiter.DefaultWindow);

        using var scope = _provider.CreateScope();
        var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
        Assert.Empty(await ctx.BundleUnlockAttempts.AsNoTracking().ToListAsync());
    }

    /// <summary>Verifies that the scheduled prune respects the retention horizon and the cadence claim.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ScheduledPrune_RespectsTheRetentionHorizonAndTheCadenceClaim()
    {
        // Two negative controls in one: a row inside the retention horizon survives, and
        // the shared cadence claim means a burst of ticks cannot prune more often than
        // PruneMinInterval (so ticking on the 1-minute session sweep is cheap).
        var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
        var window = new SqlBundleUnlockAttemptWindow(
            _provider.GetRequiredService<IServiceScopeFactory>(), clock);

        Assert.True(await window.TryRegisterAttemptAsync(
            "10.0.0.1", 10, BundleUnlockRateLimiter.DefaultWindow));

        // Recent row, past the cadence: pruning runs but must delete nothing.
        clock.Advance(TimeSpan.FromMinutes(30));
        for (var i = 0; i < 5; i++)
        {
            await window.PruneAsync(BundleUnlockRateLimiter.DefaultWindow);
        }

        using var scope = _provider.CreateScope();
        var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
        Assert.Equal(1, await ctx.BundleUnlockAttempts.AsNoTracking().CountAsync());
    }

    /// <summary>Verifies that the scheduled prune does not throw when the store is unavailable.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ScheduledPrune_DoesNotThrow_WhenTheStoreIsUnavailable()
    {
        // The sweep that drives this has no caller to report to, so PruneAsync's contract
        // is "never throw" — a database blip must not kill the background service loop.
        var services = new ServiceCollection();
        services.AddScoped<IBundleUnlockAttemptRepository>(
            _ => throw new InvalidOperationException("database unavailable"));
        using var broken = services.BuildServiceProvider();

        var window = new SqlBundleUnlockAttemptWindow(
            broken.GetRequiredService<IServiceScopeFactory>(), TimeProvider.System);

        await window.PruneAsync(BundleUnlockRateLimiter.DefaultWindow);
    }

    /// <summary>Minimal mutable-clock <see cref="TimeProvider"/> for deterministic windows.</summary>
    private sealed class TestClock : TimeProvider
    {
        private DateTimeOffset _now;
        /// <summary>Initializes a new instance of the <c>TestClock</c> class.</summary>
        /// <param name="start">The initial time of the clock.</param>
        public TestClock(DateTimeOffset start) { _now = start; }
        /// <inheritdoc />
        public override DateTimeOffset GetUtcNow() => _now;
        /// <summary>Moves the test clock forward by the given amount.</summary>
        /// <param name="by">The amount to advance the clock by.</param>
        public void Advance(TimeSpan by) => _now = _now.Add(by);
    }
}