ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterRetentionTests.cs

using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.AuditLog.Site;
using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;

namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Site;

/// <summary>
/// Tests for <see cref="SqliteAuditWriter.PurgeExpiredAsync"/> (arch-review 04, S1/U2).
/// The site SQLite store is ephemeral (~7-day retention); the retention purge deletes
/// rows that have already left the site (Forwarded/Reconciled) and are older than the
/// cutoff, while honoring the hard <see cref="AuditForwardState"/> invariant: a
/// <see cref="AuditForwardState.Pending"/> row is NEVER purged on age alone — dropping
/// an un-forwarded row would silently lose audit data that never reached central.
/// Fixture mirrors <see cref="SqliteAuditWriterBacklogStatsTests"/> (file-backed so the
/// post-purge incremental_vacuum has a real file to shrink).
/// </summary>
public class SqliteAuditWriterRetentionTests : IDisposable
{
    private readonly string _dbPath;

    /// <summary>Initializes the test class with a unique temporary database path.</summary>
    public SqliteAuditWriterRetentionTests()
    {
        _dbPath = Path.Combine(Path.GetTempPath(),
            $"audit-retention-{Guid.NewGuid():N}.db");
    }

    /// <summary>Releases the resources held by the test fixture.</summary>
    public void Dispose()
    {
        if (File.Exists(_dbPath))
        {
            try { File.Delete(_dbPath); } catch { /* test cleanup best-effort */ }
        }
    }

    private SqliteAuditWriter CreateWriter(int purgeBatchSize = 5000, int channelCapacity = 4096, int batchSize = 256)
    {
        var options = new SqliteAuditWriterOptions
        {
            DatabasePath = _dbPath,
            PurgeBatchSize = purgeBatchSize,
            ChannelCapacity = channelCapacity,
            BatchSize = batchSize,
        };
        return new SqliteAuditWriter(
            Options.Create(options),
            NullLogger<SqliteAuditWriter>.Instance,
            new FakeNodeIdentityProvider());
    }

    private static async Task<Guid> WriteEventAsync(SqliteAuditWriter writer, DateTime occurredAtUtc)
    {
        var id = Guid.NewGuid();
        var evt = ScadaBridgeAuditEventFactory.Create(
            eventId: id,
            occurredAtUtc: occurredAtUtc,
            channel: AuditChannel.ApiOutbound,
            kind: AuditKind.ApiCall,
            status: AuditStatus.Delivered);
        await writer.WriteAsync(evt);
        return id;
    }

    /// <summary>Verifies that purging deletes forwarded and reconciled rows older than the cutoff.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task PurgeExpired_DeletesForwardedAndReconciled_OlderThanCutoff()
    {
        await using var writer = CreateWriter();
        var oldForwarded = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10));
        var oldReconciled = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10));
        var oldPending = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10));
        var fresh = await WriteEventAsync(writer, DateTime.UtcNow);
        await writer.MarkForwardedAsync(new[] { oldForwarded, oldReconciled, fresh });
        await writer.MarkReconciledAsync(new[] { oldReconciled });

        var purged = await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7));

        Assert.Equal(2, purged); // oldForwarded + oldReconciled
        // Hard ForwardState invariant: the old *Pending* row survives on age alone.
        var pending = await writer.ReadPendingAsync(10);
        Assert.Contains(pending, e => e.EventId == oldPending);
        // Fresh forwarded row inside the window survives; only oldPending remains Pending.
        var stats = await writer.GetBacklogStatsAsync();
        Assert.Equal(1, stats.PendingCount);
    }

    /// <summary>Verifies that purging expired rows is idempotent.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task PurgeExpired_IsIdempotent()
    {
        await using var writer = CreateWriter();
        var id = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10));
        await writer.MarkForwardedAsync(new[] { id });
        Assert.Equal(1, await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7)));
        Assert.Equal(0, await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7)));
    }

    /// <summary>
    /// Review 2026-10-21 06/MEM-1: the purge deletes in bounded pages (one transaction
    /// each, write lock released between them) and still removes every eligible row.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task PurgeExpired_IsPaged_AndStillRemovesEveryEligibleRow()
    {
        await using var writer = CreateWriter(purgeBatchSize: 10);
        var ids = new List<Guid>();
        for (var i = 0; i < 35; i++)
        {
            ids.Add(await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10)));
        }
        var pendingId = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10));
        await writer.MarkForwardedAsync(ids);

        var purged = await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7));

        Assert.Equal(35, purged);
        Assert.Equal(4, writer.PurgePageCountForTests); // 10 + 10 + 10 + 5
        var pending = await writer.ReadPendingAsync(10);
        Assert.Equal(pendingId, Assert.Single(pending).EventId);
    }

    /// <summary>
    /// Review 2026-10-21 06/MEM-1: with the background writer parked (as behind a long purge
    /// page) and the bounded queue full, <c>WriteAsync</c> must fail fast with
    /// <see cref="SqliteAuditWriterSaturatedException"/> instead of parking a waiter that
    /// holds its event for as long as the writer stays stuck.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_QueueFull_FailsFast_InsteadOfParking()
    {
        await using var writer = CreateWriter(channelCapacity: 1, batchSize: 1);
        Task first, third;
        System.Threading.Monitor.Enter(writer.WriteLockForTests);
        try
        {
            first = writer.WriteAsync(NewEvent());
            // Wait until the writer loop has dequeued `first` and is blocked on the lock.
            Assert.True(SpinWait.SpinUntil(() =>
            {
                var probe = writer.WriteAsync(NewEvent());
                if (probe.IsFaulted) { probe.Exception!.Handle(_ => true); return false; }
                return true;
            }, TimeSpan.FromSeconds(5)));
            // The queue (capacity 1) now holds the probe; the next write cannot fit.
            third = writer.WriteAsync(NewEvent());
            Assert.True(third.IsFaulted, "a full queue must fault immediately, not park");
            Assert.IsType<SqliteAuditWriterSaturatedException>(third.Exception!.InnerException);
            Assert.True(writer.SaturatedRejectionCount >= 1);
        }
        finally
        {
            System.Threading.Monitor.Exit(writer.WriteLockForTests);
        }

        // Lock released: the parked writer resumes and the accepted event lands.
        await first.WaitAsync(TimeSpan.FromSeconds(5));
    }

    private static AuditEvent NewEvent() => ScadaBridgeAuditEventFactory.Create(
        eventId: Guid.NewGuid(),
        occurredAtUtc: DateTime.UtcNow,
        channel: AuditChannel.ApiOutbound,
        kind: AuditKind.ApiCall,
        status: AuditStatus.Delivered);
}