ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs

using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;

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

public class EventLogPurgeServiceTests : IDisposable
{
    private readonly SiteEventLogger _eventLogger;
    private readonly TestLocalDb _localDb;
    private readonly string _dbPath;
    private readonly SiteEventLogOptions _options;

    /// <summary>
    /// SiteEventLogging-023: stop flag for the concurrent stress test. Declared as
    /// a <c>volatile</c> field so every writer thread observes the main thread's
    /// `_stop = true` write without depending on JIT/runtime quirks. A plain
    /// <c>bool</c> local would be legal-cached in a register inside the tight
    /// <c>while (!_stop)</c> loop under release-mode optimisation.
    /// </summary>
    private volatile bool _stop;

    public EventLogPurgeServiceTests()
    {
        _dbPath = Path.Combine(Path.GetTempPath(), $"test_purge_{Guid.NewGuid()}.db");
        _options = new SiteEventLogOptions
        {
            DatabasePath = _dbPath,
            RetentionDays = 30,
            MaxStorageMb = 1024
        };
        _localDb = TestLocalDb.Create(_dbPath);
        _eventLogger = new SiteEventLogger(
            Options.Create(_options),
            NullLogger<SiteEventLogger>.Instance,
            _localDb.Db);
    }

    public void Dispose()
    {
        _eventLogger.Dispose();
        _localDb.Dispose();
        TestLocalDb.DeleteFiles(_dbPath);
        GC.SuppressFinalize(this);
    }

    private EventLogPurgeService CreatePurgeService(
        SiteEventLogOptions? optionsOverride = null,
        SiteEventLogActiveNodeCheck? isActiveNode = null,
        SiteEventLogReplicationCheck? isReplicationConfigured = null,
        ILogger<EventLogPurgeService>? logger = null)
    {
        var opts = optionsOverride ?? _options;
        return new EventLogPurgeService(
            _eventLogger,
            Options.Create(opts),
            logger ?? NullLogger<EventLogPurgeService>.Instance,
            isActiveNode,
            isReplicationConfigured);
    }

    private void InsertEventWithTimestamp(DateTimeOffset timestamp)
    {
        _eventLogger.WithConnection(connection =>
        {
            using var cmd = connection.CreateCommand();
            cmd.CommandText = """
                INSERT INTO site_events (id, timestamp, event_type, severity, source, message)
                VALUES ($id, $ts, 'script', 'Info', 'Test', 'Test message')
                """;
            cmd.Parameters.AddWithValue("$id", Guid.NewGuid().ToString("N"));
            cmd.Parameters.AddWithValue("$ts", timestamp.ToString("o"));
            cmd.ExecuteNonQuery();
        });
    }

    private long GetEventCount()
    {
        return _eventLogger.WithConnection(connection =>
        {
            using var cmd = connection.CreateCommand();
            cmd.CommandText = "SELECT COUNT(*) FROM site_events";
            return (long)cmd.ExecuteScalar()!;
        });
    }

    [Fact]
    public void PurgeByRetention_DeletesOldEvents()
    {
        // Insert an old event (31 days ago) and a recent one
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
        InsertEventWithTimestamp(DateTimeOffset.UtcNow);

        var purge = CreatePurgeService();
        purge.RunPurge();

        Assert.Equal(1, GetEventCount());
    }

    [Fact]
    public void PurgeByRetention_KeepsRecentEvents()
    {
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-29));
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-1));
        InsertEventWithTimestamp(DateTimeOffset.UtcNow);

        var purge = CreatePurgeService();
        purge.RunPurge();

        Assert.Equal(3, GetEventCount());
    }

    [Fact]
    public void PurgeByRetention_OverBatchSize_DeletesAllExpiredRowsAcrossSlices()
    {
        // WP1.7: the retention purge must slice a large expired backlog into
        // bounded DELETE batches (1000 rows/iteration, mirroring the storage-cap
        // purge a few lines below in the same file) rather than one unbounded
        // DELETE. Seed more than one slice's worth of expired rows plus a
        // handful of recent rows, and assert the whole expired set is gone —
        // not just the first 1000 rows — while the recent rows survive.
        const int expiredCount = 2500; // > one 1000-row slice, not an exact multiple
        var expiredBase = DateTimeOffset.UtcNow.AddDays(-31);
        _eventLogger.WithConnection(connection =>
        {
            for (var i = 0; i < expiredCount; i++)
            {
                using var cmd = connection.CreateCommand();
                cmd.CommandText = """
                    INSERT INTO site_events (id, timestamp, event_type, severity, source, message)
                    VALUES ($id, $ts, 'script', 'Info', 'Test', 'Test message')
                    """;
                cmd.Parameters.AddWithValue("$id", Guid.NewGuid().ToString("N"));
                cmd.Parameters.AddWithValue("$ts", expiredBase.AddSeconds(i).ToString("o"));
                cmd.ExecuteNonQuery();
            }
        });
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-1));
        InsertEventWithTimestamp(DateTimeOffset.UtcNow);

        var purge = CreatePurgeService();
        purge.RunPurge();

        Assert.Equal(2, GetEventCount());
    }

    [Fact]
    public void PurgeByStorageCap_DeletesOldestWhenOverCap()
    {
        // Insert enough events to have some data
        for (int i = 0; i < 100; i++)
        {
            InsertEventWithTimestamp(DateTimeOffset.UtcNow);
        }

        // Set an artificially small cap to trigger purge
        var smallCapOptions = new SiteEventLogOptions
        {
            DatabasePath = _dbPath,
            RetentionDays = 30,
            MaxStorageMb = 0  // 0 MB cap forces purge
        };

        var purge = CreatePurgeService(smallCapOptions);
        purge.RunPurge();

        // All events should be purged since cap is 0
        Assert.Equal(0, GetEventCount());
    }

    [Fact]
    public void GetDatabaseSizeBytes_ReturnsPositiveValue()
    {
        InsertEventWithTimestamp(DateTimeOffset.UtcNow);

        var purge = CreatePurgeService();
        var size = purge.GetDatabaseSizeBytes();

        Assert.True(size > 0);
    }

    /// <summary>
    /// Base instant for bulk-seeded events: one day ago, evaluated once so a run is
    /// self-consistent.
    /// <para>
    /// Must stay INSIDE the retention window (<c>RetentionDays</c>, default 30).
    /// <see cref="EventLogPurgeService.RunPurge"/> applies the retention purge before
    /// the storage-cap purge, so a fixed calendar date in the past would be deleted
    /// wholesale as stale and the storage-cap assertions would silently test an empty
    /// table rather than cap-trimming behaviour.
    /// </para>
    /// </summary>
    private static readonly DateTimeOffset BulkSeedStart =
        DateTimeOffset.UtcNow.AddDays(-1);

    /// <summary>
    /// Timestamp of the i-th bulk-seeded event. Each row is one second newer than the
    /// last, so "oldest" is unambiguous — the storage-cap purge orders by timestamp,
    /// and previously every bulk row shared the same <c>UtcNow</c>, which left the
    /// oldest-first guarantee untestable.
    /// </summary>
    private static DateTimeOffset BulkSeedTimestamp(int index) =>
        BulkSeedStart.AddSeconds(index);

    private void InsertBulkEvents(int count)
    {
        // Each event carries a sizeable details payload so the database grows
        // measurably and the storage cap can be exercised against a realistic file.
        var details = new string('x', 2000);
        _eventLogger.WithConnection(connection =>
        {
            for (int i = 0; i < count; i++)
            {
                using var cmd = connection.CreateCommand();
                cmd.CommandText = """
                    INSERT INTO site_events (id, timestamp, event_type, severity, source, message, details)
                    VALUES ($id, $ts, 'script', 'Info', 'Test', 'Bulk event', $details)
                    """;
                cmd.Parameters.AddWithValue("$id", Guid.NewGuid().ToString("N"));
                cmd.Parameters.AddWithValue("$ts", BulkSeedTimestamp(i).ToString("o"));
                cmd.Parameters.AddWithValue("$details", details);
                cmd.ExecuteNonQuery();
            }
        });
    }

    /// <summary>
    /// Oldest surviving event timestamp. Replaces the former <c>MIN(id)</c> probe:
    /// ids are GUIDs since LocalDb Phase 1 and carry no ordering, so age is measured
    /// on the timestamp column the purge itself orders by.
    /// </summary>
    private string MinEventTimestamp()
    {
        return _eventLogger.WithConnection(connection =>
        {
            using var cmd = connection.CreateCommand();
            cmd.CommandText = "SELECT MIN(timestamp) FROM site_events";
            var result = cmd.ExecuteScalar();
            return result as string ?? "";
        });
    }

    [Fact]
    public void PurgeByStorageCap_StopsWhenUnderCap_DoesNotEmptyTable()
    {
        // Regression test for SiteEventLogging-001 / -002:
        // a realistic non-zero cap must trim the oldest events to the budget,
        // not delete the entire table.
        InsertBulkEvents(3000);

        var purge = CreatePurgeService();
        var totalSize = purge.GetDatabaseSizeBytes();

        // Cap at roughly half the current database size — purge must keep some rows.
        var capBytes = totalSize / 2;
        var capOptions = new SiteEventLogOptions
        {
            DatabasePath = _dbPath,
            RetentionDays = 30,
            MaxStorageMb = (int)Math.Max(1, capBytes / (1024 * 1024))
        };

        var cappedPurge = CreatePurgeService(capOptions);
        cappedPurge.RunPurge();

        var remaining = GetEventCount();
        Assert.True(remaining > 0, "Storage-cap purge must not delete the entire table.");
        Assert.True(remaining < 3000, "Storage-cap purge must remove some events when over cap.");

        // The database must actually be back under the cap after purge.
        var finalSize = cappedPurge.GetDatabaseSizeBytes();
        var finalCapBytes = (long)capOptions.MaxStorageMb * 1024 * 1024;
        Assert.True(finalSize <= finalCapBytes,
            $"Database size {finalSize} must be at or below cap {finalCapBytes} after purge.");
    }

    [Fact]
    public void PurgeByStorageCap_RemovesOldestEventsFirst()
    {
        // Regression test for SiteEventLogging-002: only the oldest events should be
        // removed when trimming to the cap.
        //
        // "Oldest" used to mean "lowest autoincrement id". Since LocalDb Phase 1 the id
        // is a GUID with no ordering, so both the purge and this test key on timestamp.
        // Ordering by id here would delete an arbitrary batch, which is precisely the
        // regression this test now guards.
        const int eventCount = 3000;
        InsertBulkEvents(eventCount);

        var purge = CreatePurgeService();
        var totalSize = purge.GetDatabaseSizeBytes();

        var capOptions = new SiteEventLogOptions
        {
            DatabasePath = _dbPath,
            RetentionDays = 30,
            MaxStorageMb = (int)Math.Max(1, (totalSize / 2) / (1024 * 1024))
        };

        var oldestBefore = MinEventTimestamp();
        var cappedPurge = CreatePurgeService(capOptions);
        cappedPurge.RunPurge();
        var oldestAfter = MinEventTimestamp();

        // The surviving rows must be the newest ones — the oldest timestamp advanced.
        Assert.True(
            string.CompareOrdinal(oldestAfter, oldestBefore) > 0,
            $"Oldest events must be purged first; oldest went from '{oldestBefore}' to '{oldestAfter}'.");

        // The newest event must still be present.
        var newestTimestamp = BulkSeedTimestamp(eventCount - 1).ToString("o");
        var newestPresent = _eventLogger.WithConnection(connection =>
        {
            using var cmd = connection.CreateCommand();
            cmd.CommandText = "SELECT COUNT(*) FROM site_events WHERE timestamp = $ts";
            cmd.Parameters.AddWithValue("$ts", newestTimestamp);
            return (long)cmd.ExecuteScalar()!;
        });
        Assert.Equal(1L, newestPresent);
    }

    [Fact]
    public async Task StartAsync_DoesNotBlock_OnTheInitialPurge()
    {
        // SiteEventLogging-014 (re-triaged): on .NET 8+ BackgroundService runs
        // ExecuteAsync on a thread-pool thread — the synchronous prelude (the
        // initial RunPurge()) does NOT execute on the host startup thread, so
        // StartAsync returns promptly and host startup / the /health/ready gate is
        // not blocked even by a large initial purge. This test pins that behaviour:
        // StartAsync returns fast, and the initial purge still happens shortly
        // afterwards on the background scheduler.
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
        Assert.Equal(1, GetEventCount());

        var purge = CreatePurgeService();
        using var cts = new CancellationTokenSource();

        var sw = System.Diagnostics.Stopwatch.StartNew();
        await purge.StartAsync(cts.Token);
        sw.Stop();

        Assert.True(sw.ElapsedMilliseconds < 1000,
            $"StartAsync blocked for {sw.ElapsedMilliseconds} ms — the initial purge " +
            "must not run on the host startup thread.");

        // The initial purge still runs on the background scheduler.
        var deadline = DateTime.UtcNow.AddSeconds(5);
        while (GetEventCount() != 0 && DateTime.UtcNow < deadline)
        {
            await Task.Delay(25);
        }
        Assert.Equal(0, GetEventCount());

        await cts.CancelAsync();
        await purge.StopAsync(CancellationToken.None);
    }

    [Fact]
    public async Task PurgeByStorageCap_ConcurrentWritesDoNotCorruptConnection()
    {
        // Regression test for SiteEventLogging-003: purge running on a background
        // thread while events are recorded on other threads must not throw
        // "DataReader already open" / "connection busy" from a shared connection.
        InsertBulkEvents(2000);

        var purge = CreatePurgeService();
        var totalSize = purge.GetDatabaseSizeBytes();
        var capOptions = new SiteEventLogOptions
        {
            DatabasePath = _dbPath,
            RetentionDays = 30,
            MaxStorageMb = (int)Math.Max(1, (totalSize / 2) / (1024 * 1024))
        };

        var exceptions = new System.Collections.Concurrent.ConcurrentBag<Exception>();
        // SiteEventLogging-023: must be volatile so the writer threads observe the
        // main thread's `stop = true` flip after the purge task completes. Without
        // it, a release-mode JIT is permitted to cache the `stop = false` read in
        // a register inside the tight `while (!stop)` loop and never see the flip,
        // causing the writer tasks to hang past xUnit's per-test timeout instead
        // of asserting `Empty(exceptions)`.
        _stop = false;

        var purgeTask = Task.Run(() =>
        {
            try
            {
                var p = CreatePurgeService(capOptions);
                for (int i = 0; i < 20; i++) p.RunPurge();
            }
            catch (Exception ex) { exceptions.Add(ex); }
        });

        var writeTasks = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
        {
            try
            {
                while (!_stop)
                {
                    await _eventLogger.LogEventAsync("script", "Info", null, "Concurrent", "Concurrent write");
                }
            }
            catch (Exception ex) { exceptions.Add(ex); }
        })).ToArray();

        await purgeTask;
        _stop = true;
        await Task.WhenAll(writeTasks);

        Assert.Empty(exceptions);
    }

    // ── SiteEventLogging-019: purge runs only on the active node ──

    [Fact]
    public void RunPurge_OnStandbyNode_SkipsAllWork()
    {
        // SiteEventLogging-019: per design, the daily purge runs on the active
        // node only. The standby's local SQLite receives no writes, so purging
        // there is unnecessary; we gate the purge tick on the injected
        // active-node check and early-exit when it returns false. The row
        // inserted here is well past retention, so a real purge would delete
        // it — the standby gate must leave it intact.
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
        Assert.Equal(1, GetEventCount());

        var purge = CreatePurgeService(isActiveNode: () => false);
        purge.RunPurge();

        Assert.Equal(1, GetEventCount());
    }

    [Fact]
    public void RunPurge_OnActiveNode_RunsTheRetentionPurge()
    {
        // SiteEventLogging-019: when the active-node check returns true the
        // service runs the purge as before. Pinned alongside the standby case
        // so a future regression that inverts the gate is caught.
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
        InsertEventWithTimestamp(DateTimeOffset.UtcNow);

        var purge = CreatePurgeService(isActiveNode: () => true);
        purge.RunPurge();

        Assert.Equal(1, GetEventCount());
    }

    [Fact]
    public void RunPurge_WithNullCheck_FallsBackToRunning()
    {
        // SiteEventLogging-019: when no active-node check is supplied (the
        // default for non-clustered hosts and pre-existing tests), the service
        // preserves the pre-fix "run on every tick" behaviour rather than
        // silently skipping every tick. Backward compatibility guard.
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));

        var purge = CreatePurgeService(isActiveNode: null);
        purge.RunPurge();

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

    // ── R7: replication-backlog operator note on the purge log line ──

    /// <summary>
    /// Fragment unique to the R7 note, matched against the rendered log message.
    /// Deliberately not the whole sentence — the wording is operator prose and may be
    /// reworded; what must not regress is that the note fires (or does not) per the
    /// replication predicate.
    /// </summary>
    private const string BacklogNoteFragment = "transient LocalDb oplog backlog";

    [Fact]
    public void PurgeByRetention_OnReplicatedNode_LogsTheOplogBacklogNote()
    {
        // R7: site_events is a replicated table and the retention DELETE is CDC-captured
        // like any other write — no purge-exemption path by design — so the purge inflates
        // LocalDbOplogBacklog / localdb_oplog_depth until the peer acks. The Information
        // line is the breadcrumb that lets an operator tie the spike to the purge instead
        // of reading it as a replication fault.
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));

        var logger = new CapturingLogger();
        var purge = CreatePurgeService(isReplicationConfigured: () => true, logger: logger);
        purge.RunPurge();

        Assert.Equal(0, GetEventCount());
        var note = Assert.Single(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
        Assert.Equal(LogLevel.Information, note.Level);
    }

    [Fact]
    public void PurgeByRetention_OnUnreplicatedNode_DoesNotLogTheNote()
    {
        // The other half of the gate: site-b/site-c on the rig run with no replication
        // configured, have no CDC triggers at all, and therefore no backlog to explain.
        // Emitting the note there would be noise pointing at a metric that reads null.
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));

        var logger = new CapturingLogger();
        var purge = CreatePurgeService(isReplicationConfigured: () => false, logger: logger);
        purge.RunPurge();

        Assert.Equal(0, GetEventCount());
        Assert.DoesNotContain(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
    }

    [Fact]
    public void PurgeByRetention_WithNoRowsDeleted_DoesNotLogTheNote()
    {
        // The note is a companion to a purge line, not a tick heartbeat: a daily tick that
        // deleted nothing produces no oplog rows and so must stay silent, otherwise the
        // breadcrumb loses all correlating value.
        InsertEventWithTimestamp(DateTimeOffset.UtcNow);

        var logger = new CapturingLogger();
        var purge = CreatePurgeService(isReplicationConfigured: () => true, logger: logger);
        purge.RunPurge();

        Assert.Equal(1, GetEventCount());
        Assert.DoesNotContain(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
    }

    [Fact]
    public void PurgeByStorageCap_OnReplicatedNode_LogsTheOplogBacklogNote()
    {
        // A cap-driven trim deletes rows through the same replicated table and is usually
        // the larger of the two bursts, so it carries the same note.
        for (int i = 0; i < 100; i++)
        {
            InsertEventWithTimestamp(DateTimeOffset.UtcNow);
        }

        var capOptions = new SiteEventLogOptions
        {
            DatabasePath = _dbPath,
            RetentionDays = 30,
            MaxStorageMb = 0  // 0 MB cap forces the cap purge; retention deletes nothing here
        };

        var logger = new CapturingLogger();
        var purge = CreatePurgeService(capOptions, isReplicationConfigured: () => true, logger: logger);
        purge.RunPurge();

        Assert.Equal(0, GetEventCount());
        var note = Assert.Single(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
        Assert.Equal(LogLevel.Information, note.Level);
    }

    [Fact]
    public void RunPurge_WhenBothPhasesDeleteRows_LogsTheBacklogNoteOnceForTheWholeTick()
    {
        // Register row 84. The note used to be emitted by each phase, so a tick in which the
        // retention purge AND the cap trim both deleted rows produced two identical Information
        // lines. It is now emitted once, from the tick's shared delete budget — which also makes
        // the number it reports the figure an operator actually wants next to an oplog spike:
        // the whole tick's contribution, not one phase's.
        const int expiredRows = 50;
        const int recentRows = 50;
        for (int i = 0; i < expiredRows; i++)
        {
            InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
        }
        for (int i = 0; i < recentRows; i++)
        {
            InsertEventWithTimestamp(DateTimeOffset.UtcNow);
        }

        var capOptions = new SiteEventLogOptions
        {
            DatabasePath = _dbPath,
            RetentionDays = 30,
            MaxStorageMb = 0  // retention takes the expired rows; the cap trim takes the rest
        };

        var logger = new CapturingLogger();
        var purge = CreatePurgeService(capOptions, isReplicationConfigured: () => true, logger: logger);
        purge.RunPurge();

        // Both phases really did delete — otherwise this would pass for the wrong reason.
        Assert.Contains(logger.Entries, e =>
            e.Message.Contains($"Purged {expiredRows} events older than", StringComparison.Ordinal));
        Assert.Contains(logger.Entries, e => e.Message.Contains("over the 0 MB cap", StringComparison.Ordinal));
        Assert.Equal(0, GetEventCount());

        var note = Assert.Single(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
        Assert.Equal(LogLevel.Information, note.Level);
        Assert.Contains($"Purged {expiredRows + recentRows} site_events rows", note.Message, StringComparison.Ordinal);
    }

    [Fact]
    public void RunPurge_WhenReplicationCheckThrows_StillPurgesAndSwallows()
    {
        // Defensive: the replication predicate only selects log wording. A throw from it
        // must never escape the purge — the rows are already deleted by the time it is
        // consulted, and an exception here would surface as "Error during event log purge"
        // for a purge that in fact succeeded.
        InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));

        var logger = new CapturingLogger();
        var purge = CreatePurgeService(
            isReplicationConfigured: () => throw new InvalidOperationException("boom"),
            logger: logger);
        purge.RunPurge();

        Assert.Equal(0, GetEventCount());
        Assert.DoesNotContain(logger.Entries, e => e.Level == LogLevel.Error);
        Assert.DoesNotContain(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
    }

    // ── Register row 34: per-tick delete ceiling ──

    /// <summary>
    /// Fragment unique to the ceiling-reached line. As with the backlog note, the operator prose
    /// may be reworded; what must not regress is that the line fires when a tick is cut short.
    /// </summary>
    private const string CeilingLineFragment = "per-tick ceiling";

    /// <summary>The counter incremented once per ceiling-limited tick.</summary>
    private const string CeilingHitCounter = "scadabridge.site.event_log.purge_ceiling_hits";

    /// <summary>Seeds <paramref name="count"/> events one second apart, all past retention.</summary>
    private void InsertExpiredEvents(int count)
    {
        var expiredBase = DateTimeOffset.UtcNow.AddDays(-31);
        _eventLogger.WithConnection(connection =>
        {
            using var tx = connection.BeginTransaction();
            using var cmd = connection.CreateCommand();
            cmd.Transaction = tx;
            cmd.CommandText = """
                INSERT INTO site_events (id, timestamp, event_type, severity, source, message)
                VALUES ($id, $ts, 'script', 'Info', 'Test', 'Test message')
                """;
            var id = cmd.Parameters.Add("$id", Microsoft.Data.Sqlite.SqliteType.Text);
            var ts = cmd.Parameters.Add("$ts", Microsoft.Data.Sqlite.SqliteType.Text);
            for (var i = 0; i < count; i++)
            {
                id.Value = Guid.NewGuid().ToString("N");
                ts.Value = expiredBase.AddSeconds(i).ToString("o");
                cmd.ExecuteNonQuery();
            }
            tx.Commit();
        });
    }

    /// <summary>Options with the per-tick ceiling set and a short catch-up interval.</summary>
    private SiteEventLogOptions CeilingOptions(int ceiling, int maxStorageMb = 1024) => new()
    {
        DatabasePath = _dbPath,
        RetentionDays = 30,
        MaxStorageMb = maxStorageMb,
        MaxPurgedRowsPerTick = ceiling,
        PurgeCatchUpInterval = TimeSpan.FromSeconds(1),
    };

    [Fact]
    public void PurgeByRetention_StopsAtThePerTickCeiling()
    {
        // Row 34 core behaviour. site_events is CDC-captured on a replicated node with no
        // purge-exemption path, so an unbounded tick queues one oplog row per deleted event —
        // measured at 2,731,947 rows for a full 1 GB log, 2.7x over LocalDb's MaxOplogRows
        // default, which costs a full ten-table snapshot resync. The ceiling bounds the burst.
        InsertExpiredEvents(2500);

        var purge = CreatePurgeService(CeilingOptions(1000));
        var ceilingHit = purge.RunPurge();

        Assert.True(ceilingHit, "A tick cut short by the ceiling must ask for an accelerated catch-up tick.");
        Assert.Equal(1500, GetEventCount());
    }

    [Fact]
    public void PurgeByRetention_WithCeiling_ConvergesAcrossSuccessiveTicks()
    {
        // The remainder must be picked up, or pacing would trade a burst for a retention policy
        // that never completes. 2500 expired rows at 1000/tick = three ticks, the third of which
        // finishes the work and reports no ceiling hit.
        InsertExpiredEvents(2500);
        InsertEventWithTimestamp(DateTimeOffset.UtcNow);

        var purge = CreatePurgeService(CeilingOptions(1000));

        Assert.True(purge.RunPurge());
        Assert.Equal(1501, GetEventCount());
        Assert.True(purge.RunPurge());
        Assert.Equal(501, GetEventCount());
        Assert.False(purge.RunPurge(), "The final tick clears the backlog and must not ask for another.");
        Assert.Equal(1, GetEventCount());
    }

    [Fact]
    public void PurgeByRetention_ExpiredSetLandingExactlyOnTheCeiling_IsNotACeilingHit()
    {
        // Exhausting the budget is not the same as having work left. A tick that deleted the last
        // expired row and happened to land on the ceiling must not arm a catch-up tick that will
        // delete nothing and log that retention is behind.
        InsertExpiredEvents(1000);

        var purge = CreatePurgeService(CeilingOptions(1000));

        Assert.False(purge.RunPurge());
        Assert.Equal(0, GetEventCount());
    }

    [Fact]
    public void MaxPurgedRowsPerTick_Zero_RestoresUnboundedPurges()
    {
        // The documented escape hatch: 0 means unlimited, i.e. byte-for-byte the pre-row-34
        // behaviour, for an operator who would rather take one long burst than a paced one.
        InsertExpiredEvents(2500);

        var purge = CreatePurgeService(CeilingOptions(0));

        Assert.False(purge.RunPurge());
        Assert.Equal(0, GetEventCount());
    }

    [Fact]
    public void RunPurge_HittingTheCeiling_LogsTheCeilingLineAndCountsIt()
    {
        // Visibility requirement: a ceiling-limited tick is silent progress otherwise — the log
        // line names the ceiling and the accelerated retry, and the counter carries every hit so
        // "converging" can be told from "never converging".
        InsertExpiredEvents(2500);

        using var probe = new CounterProbe(CeilingHitCounter);
        var logger = new CapturingLogger();
        var purge = CreatePurgeService(CeilingOptions(1000), logger: logger);
        purge.RunPurge();

        var line = Assert.Single(
            logger.Entries, e => e.Message.Contains(CeilingLineFragment, StringComparison.Ordinal));
        Assert.Equal(LogLevel.Warning, line.Level);
        Assert.Equal(1, probe.Total);
    }

    [Fact]
    public void RunPurge_RepeatedCeilingHits_RateLimitTheLogLineButNotTheCounter()
    {
        // A large backlog produces a RUN of ceiling-limited ticks, one every catch-up interval.
        // Logging each would bury the rest of the log; the counter is the unthrottled signal.
        InsertExpiredEvents(3000);

        using var probe = new CounterProbe(CeilingHitCounter);
        var logger = new CapturingLogger();
        var purge = CreatePurgeService(CeilingOptions(1000), logger: logger);
        purge.RunPurge();
        purge.RunPurge();

        Assert.Single(logger.Entries, e => e.Message.Contains(CeilingLineFragment, StringComparison.Ordinal));
        Assert.Equal(2, probe.Total);
    }

    [Fact]
    public void PurgeByStorageCap_SharesTheTickBudgetWithRetention()
    {
        // The ceiling bounds the TICK, not each phase: two independently-capped phases would let
        // one tick emit twice the intended oplog burst. Here the retention purge consumes the
        // whole budget, so the cap trim does not run at all — and must not log that it is
        // purging oldest events while deleting nothing.
        InsertExpiredEvents(1200);
        for (var i = 0; i < 200; i++)
            InsertEventWithTimestamp(DateTimeOffset.UtcNow);

        var logger = new CapturingLogger();
        // A 0 MB cap means the cap trim would empty whatever the retention purge left, if it ran.
        var purge = CreatePurgeService(CeilingOptions(1000, maxStorageMb: 0), logger: logger);

        Assert.True(purge.RunPurge());
        Assert.Equal(400, GetEventCount());
        Assert.DoesNotContain(logger.Entries, e => e.Message.Contains("exceeds cap", StringComparison.Ordinal));
    }

    [Fact]
    public void PurgeByStorageCap_WithCeiling_StopsAtTheBudgetAndAsksForCatchUp()
    {
        // The cap path pays the same CDC cost and is usually the larger burst of the two, so it
        // draws from the same budget and reports the same ceiling hit when it runs out.
        for (var i = 0; i < 2500; i++)
            InsertEventWithTimestamp(DateTimeOffset.UtcNow);

        var purge = CreatePurgeService(CeilingOptions(1000, maxStorageMb: 0));

        Assert.True(purge.RunPurge());
        Assert.Equal(1500, GetEventCount());
    }

    /// <summary>
    /// Accumulates measurements on one counter instrument for as long as it is alive.
    /// </summary>
    /// <remarks>
    /// A <c>Counter&lt;long&gt;</c> is not observable, so a listener started after the fact sees
    /// nothing — the probe must be constructed BEFORE the code under test runs. Tests in this
    /// class run sequentially (xUnit parallelises across classes, not within one) and no other
    /// class in the assembly touches this instrument, so the process-global meter is safe to
    /// read absolutely here.
    /// </remarks>
    private sealed class CounterProbe : IDisposable
    {
        private readonly System.Diagnostics.Metrics.MeterListener _listener = new();
        private long _total;

        public CounterProbe(string instrumentName)
        {
            _listener.InstrumentPublished = (instrument, listener) =>
            {
                if (instrument.Name == instrumentName)
                    listener.EnableMeasurementEvents(instrument);
            };
            _listener.SetMeasurementEventCallback<long>(
                (_, measurement, _, _) => Interlocked.Add(ref _total, measurement));
            _listener.Start();
        }

        /// <summary>Measurements recorded since construction.</summary>
        public long Total => Interlocked.Read(ref _total);

        public void Dispose() => _listener.Dispose();
    }

    /// <summary>
    /// Minimal <see cref="ILogger{TCategoryName}"/> that records level + rendered message.
    /// The purge service has no other observable output for a log-only change, and the
    /// suite has no shared capturing logger to reuse.
    /// </summary>
    private sealed class CapturingLogger : ILogger<EventLogPurgeService>
    {
        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)));
        }
    }
}