ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/TerminalPurgeRateLimiterTests.cs

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using NSubstitute.ExceptionExtensions;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening;

namespace ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests;

/// <summary>
/// Gitea #90: the terminal-record purge rate limiter is an injectable, clock-driven singleton
/// instead of a process-wide static, so the sweep's rate limit and its swallow-on-failure path
/// can be tested independently of every other test in the assembly.
/// </summary>
public class TerminalPurgeRateLimiterTests
{
    private static readonly DateTimeOffset Start = new(2026, 11, 9, 12, 0, 0, TimeSpan.Zero);

    private sealed class ManualClock(DateTimeOffset start) : TimeProvider
    {
        public DateTimeOffset Now { get; set; } = start;

        public override DateTimeOffset GetUtcNow() => Now;
    }

    private sealed class CapturingLogger : ILogger<DeploymentService>
    {
        public List<(LogLevel Level, Exception? Exception, string Message)> Entries { get; } = new();

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

    private static DeploymentService CreateService(
        IDeploymentManagerRepository repo,
        TerminalPurgeRateLimiter? limiter,
        ILogger<DeploymentService>? logger = null) =>
        new(
            repo,
            Substitute.For<ISiteRepository>(),
            Substitute.For<ITemplateEngineRepository>(),
            Substitute.For<IFlatteningPipeline>(),
            new CommunicationService(
                Options.Create(new CommunicationOptions()), NullLogger<CommunicationService>.Instance),
            new OperationLockManager(),
            Substitute.For<IAuditService>(),
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
            Options.Create(new DeploymentManagerOptions()),
            Options.Create(new CommunicationOptions()),
            logger ?? NullLogger<DeploymentService>.Instance,
            limiter);

    private static IDeploymentManagerRepository PurgingRepo(int terminal = 1, int artifact = 3)
    {
        var repo = Substitute.For<IDeploymentManagerRepository>();
        repo.PurgeTerminalDeploymentRecordsAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(terminal);
        repo.PurgeSystemArtifactDeploymentRecordsAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(artifact);
        return repo;
    }

    /// <summary>The first claim wins; a second inside the interval loses; after it, a claim wins again.</summary>
    [Fact]
    public void TryClaim_OncePerInterval_OnTheInjectedClock()
    {
        var clock = new ManualClock(Start);
        var limiter = new TerminalPurgeRateLimiter(clock);

        Assert.True(limiter.TryClaim(out var first));
        Assert.Equal(Start, first);

        clock.Now = Start + TerminalPurgeRateLimiter.MinInterval - TimeSpan.FromSeconds(1);
        Assert.False(limiter.TryClaim(out _));

        clock.Now = Start + TerminalPurgeRateLimiter.MinInterval;
        Assert.True(limiter.TryClaim(out var later));
        Assert.Equal(clock.Now, later);
        Assert.False(limiter.TryClaim(out _));
    }

    /// <summary>Of many concurrent claimants exactly one wins.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task TryClaim_ConcurrentCallers_ExactlyOneWins()
    {
        var limiter = new TerminalPurgeRateLimiter(new ManualClock(Start));
        using var gate = new Barrier(16);
        var wins = await Task.WhenAll(Enumerable.Range(0, 16).Select(_ => Task.Run(() =>
        {
            gate.SignalAndWait();
            return limiter.TryClaim(out var _);
        })));

        Assert.Equal(1, wins.Count(w => w));
    }

    /// <summary>
    /// The limiter is per instance, not process-wide: two services with their own limiters both
    /// sweep, and two sharing one (as the DI singleton makes them) sweep once.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Sweep_RateLimitIsPerLimiter_NotProcessWide()
    {
        var clock = new ManualClock(Start);

        Assert.Equal(4, await CreateService(PurgingRepo(), new TerminalPurgeRateLimiter(clock))
            .TryPurgeTerminalDeploymentRecordsAsync());
        Assert.Equal(4, await CreateService(PurgingRepo(), new TerminalPurgeRateLimiter(clock))
            .TryPurgeTerminalDeploymentRecordsAsync());

        var shared = new TerminalPurgeRateLimiter(clock);
        var secondRepo = PurgingRepo();
        Assert.Equal(4, await CreateService(PurgingRepo(), shared).TryPurgeTerminalDeploymentRecordsAsync());
        Assert.Equal(0, await CreateService(secondRepo, shared).TryPurgeTerminalDeploymentRecordsAsync());
        await secondRepo.DidNotReceiveWithAnyArgs().PurgeTerminalDeploymentRecordsAsync(default, default, default);
    }

    /// <summary>
    /// The sweep's cutoff is taken from the injected clock: now minus the retention window.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Sweep_CutoffIsInjectedClockMinusRetention()
    {
        var repo = PurgingRepo();
        await CreateService(repo, new TerminalPurgeRateLimiter(new ManualClock(Start)))
            .TryPurgeTerminalDeploymentRecordsAsync();

        var cutoff = Start - new DeploymentManagerOptions().TerminalDeploymentRecordRetention;
        await repo.Received(1).PurgeTerminalDeploymentRecordsAsync(cutoff, Arg.Any<int>(), Arg.Any<CancellationToken>());
        await repo.Received(1).PurgeSystemArtifactDeploymentRecordsAsync(cutoff, Arg.Any<int>(), Arg.Any<CancellationToken>());
    }

    /// <summary>
    /// A failing sweep never fails the caller: the exception is logged as a warning and swallowed,
    /// the rows already purged are still reported, and the failed sweep still consumes the
    /// interval — the next call inside it does not retry, the first call after it does.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Sweep_Failure_IsLoggedAndSwallowed_AndWaitsOutTheInterval()
    {
        var clock = new ManualClock(Start);
        var limiter = new TerminalPurgeRateLimiter(clock);
        var logger = new CapturingLogger();
        var boom = new InvalidOperationException("deadlock victim");
        var repo = Substitute.For<IDeploymentManagerRepository>();
        repo.PurgeTerminalDeploymentRecordsAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(2);
        repo.PurgeSystemArtifactDeploymentRecordsAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
            .ThrowsAsync(boom);
        var service = CreateService(repo, limiter, logger);

        var purged = await service.TryPurgeTerminalDeploymentRecordsAsync();

        Assert.Equal(2, purged); // the terminal batch that landed before the failure
        var warning = Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning);
        Assert.Same(boom, warning.Exception);
        Assert.Contains("retention sweep failed", warning.Message, StringComparison.Ordinal);
        Assert.DoesNotContain(logger.Entries, e => e.Level == LogLevel.Information);

        // Inside the interval: rate-limited, no retry.
        repo.ClearReceivedCalls();
        clock.Now = Start + TimeSpan.FromHours(1);
        Assert.Equal(0, await service.TryPurgeTerminalDeploymentRecordsAsync());
        await repo.DidNotReceiveWithAnyArgs().PurgeTerminalDeploymentRecordsAsync(default, default, default);

        // After the interval: the sweep runs again (and fails again, still swallowed).
        clock.Now = Start + TerminalPurgeRateLimiter.MinInterval;
        Assert.Equal(2, await service.TryPurgeTerminalDeploymentRecordsAsync());
        await repo.ReceivedWithAnyArgs(1).PurgeTerminalDeploymentRecordsAsync(default, default, default);
        Assert.Equal(2, logger.Entries.Count(e => e.Level == LogLevel.Warning));
    }

    /// <summary>A failure on the very first batch reports zero and is swallowed.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Sweep_FirstBatchFailure_ReturnsZero_DoesNotThrow()
    {
        var repo = Substitute.For<IDeploymentManagerRepository>();
        repo.PurgeTerminalDeploymentRecordsAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
            .ThrowsAsync(new TimeoutException("SQL timeout"));
        var logger = new CapturingLogger();

        var purged = await CreateService(repo, new TerminalPurgeRateLimiter(new ManualClock(Start)), logger)
            .TryPurgeTerminalDeploymentRecordsAsync();

        Assert.Equal(0, purged);
        Assert.IsType<TimeoutException>(Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning).Exception);
        await repo.DidNotReceiveWithAnyArgs().PurgeSystemArtifactDeploymentRecordsAsync(default, default, default);
    }

    /// <summary>
    /// AddDeploymentManager registers the limiter as a singleton, so every scoped DeploymentService
    /// shares one rate limit.
    /// </summary>
    [Fact]
    public void AddDeploymentManager_RegistersLimiterAsSingleton()
    {
        var services = new ServiceCollection();
        services.AddDeploymentManager();

        var descriptor = Assert.Single(services, d => d.ServiceType == typeof(TerminalPurgeRateLimiter));
        Assert.Equal(ServiceLifetime.Singleton, descriptor.Lifetime);

        using var provider = services.BuildServiceProvider();
        using var a = provider.CreateScope();
        using var b = provider.CreateScope();
        Assert.Same(
            a.ServiceProvider.GetRequiredService<TerminalPurgeRateLimiter>(),
            b.ServiceProvider.GetRequiredService<TerminalPurgeRateLimiter>());
    }
}