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;

    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();
    }

    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);

    [Fact]
    public async Task TwoNodesSharingOneStore_EnforceASingleCombinedWindow()
    {
        // The headline row-52 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));
    }

    [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);
    }

    [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));
    }

    [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));
    }

    [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));
    }

    [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));
        }
    }

    [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));
    }

    [Fact]
    public async Task WithoutASharedWindow_TheLimiterStillEnforcesAProcessLocalCap()
    {
        // Hosts that register Transport without a shared window keep the pre-row-52
        // 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));
    }

    [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);
    }

    [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>Minimal mutable-clock <see cref="TimeProvider"/> for deterministic windows.</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);
    }
}