mxaccessgw/src/ZB.MOM.WW.MxGateway.Tests/Security/Audit/ChannelAuditWriterTests.cs

using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Time.Testing;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.Auth.ApiKeys.Sqlite;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Security.Audit;
using ZB.MOM.WW.MxGateway.Tests.Security.Authentication;

namespace ZB.MOM.WW.MxGateway.Tests.Security.Audit;

/// <summary>
/// Tests the bounded, asynchronous audit path: <see cref="ChannelAuditWriter"/> (enqueue-only
/// once a drain is attached, drops rather than blocks when the channel is full, writes through
/// whenever nothing is draining) and <see cref="AuditDrainService"/> (batched drain, poison-batch
/// isolation, one-time table bootstrap, retention sweep). The channel makes the already-documented
/// best-effort audit contract explicit and bounded: a partially denied bulk RPC no longer pays a
/// SQLite round-trip per denied tag.
/// </summary>
public sealed class ChannelAuditWriterTests : IDisposable
{
    private readonly List<TempDatabaseDirectory> _tempDirectories = [];

    /// <summary>
    /// With a drain attached, <see cref="ChannelAuditWriter.WriteAsync"/> only enqueues — the
    /// sink is not touched until the drain runs, and the event arrives once it does.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_WithDrainAttached_EnqueuesOnlyAndReachesSinkAfterDrain()
    {
        CountingAuditSink sink = new();
        (ChannelAuditWriter writer, AuditDrainService drain) = CreateWriterAndDrain(sink);
        writer.AttachDrain();

        await writer.WriteAsync(MakeEvent("constraint-denied"), CancellationToken.None);

        // Enqueue-only: nothing has reached the sink yet.
        Assert.Equal(0, sink.InsertBatchCalls);
        Assert.Empty(sink.Events);

        int drained = await drain.DrainPendingAsync(CancellationToken.None);

        Assert.Equal(1, drained);
        Assert.Equal(1, sink.InsertBatchCalls);
        Assert.Equal("constraint-denied", Assert.Single(sink.Events).Action);
    }

    /// <summary>
    /// A drain pass batches up to <see cref="AuditDrainService.MaxBatchSize"/> events into one
    /// sink call, so a partially denied bulk RPC costs a handful of transactions, not one per tag.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DrainPendingAsync_BatchesEventsIntoBoundedInsertCalls()
    {
        CountingAuditSink sink = new();
        (ChannelAuditWriter writer, AuditDrainService drain) = CreateWriterAndDrain(sink);
        writer.AttachDrain();

        const int eventCount = 150;
        for (int index = 0; index < eventCount; index++)
        {
            await writer.WriteAsync(MakeEvent($"denied-{index}"), CancellationToken.None);
        }

        int drained = await drain.DrainPendingAsync(CancellationToken.None);

        Assert.Equal(eventCount, drained);
        Assert.Equal(eventCount, sink.Events.Count);
        // 150 events at a batch size of 64 → 3 transactions (64 + 64 + 22), not 150.
        Assert.Equal(3, sink.InsertBatchCalls);
        Assert.All(sink.BatchSizes, size => Assert.True(size <= AuditDrainService.MaxBatchSize));
    }

    /// <summary>
    /// A batch that will not commit is retried one event at a time, so a single unwritable row
    /// costs only itself instead of taking the up-to-63 good events sharing its transaction.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DrainPendingAsync_WhenBatchFails_RetriesIndividuallyAndKeepsSurvivors()
    {
        CountingAuditSink sink = new() { PoisonAction = "poison" };
        (ChannelAuditWriter writer, AuditDrainService drain) = CreateWriterAndDrain(sink);
        writer.AttachDrain();

        await writer.WriteAsync(MakeEvent("good-1"), CancellationToken.None);
        await writer.WriteAsync(MakeEvent("poison"), CancellationToken.None);
        await writer.WriteAsync(MakeEvent("good-2"), CancellationToken.None);

        int persisted = await drain.DrainPendingAsync(CancellationToken.None);

        // The batch commit fails on the poison event; the per-event retry still lands both others.
        Assert.Equal(2, persisted);
        Assert.Equal(["good-1", "good-2"], sink.Events.Select(auditEvent => auditEvent.Action));
    }

    /// <summary>
    /// When the bounded channel is full the write is dropped rather than blocking the caller,
    /// and the drop is counted. A stalled or slow audit database must never stall an RPC.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_WhenChannelFull_DropsWriteAndCountsItWithoutBlocking()
    {
        CountingAuditSink sink = new();
        (ChannelAuditWriter writer, _) = CreateWriterAndDrain(sink);
        writer.AttachDrain();

        const int overflow = 32;
        for (int index = 0; index < ChannelAuditWriter.ChannelCapacity + overflow; index++)
        {
            // Every call must complete synchronously: the channel never blocks a producer.
            Task write = writer.WriteAsync(MakeEvent($"denied-{index}"), CancellationToken.None);
            Assert.True(write.IsCompletedSuccessfully);
            await write;
        }

        Assert.Equal(overflow, writer.DroppedCount);
        // The channel still holds exactly its capacity; nothing reached the sink (no drain ran).
        Assert.Equal(0, sink.InsertBatchCalls);
    }

    /// <summary>
    /// With no drain attached (the <c>apikey</c> CLI, DI-only tests, post-shutdown) the writer
    /// writes through to the sink instead of enqueueing into a channel nobody will ever read.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_WithNoDrainAttached_WritesThroughToSink()
    {
        CountingAuditSink sink = new();
        (ChannelAuditWriter writer, _) = CreateWriterAndDrain(sink);

        await writer.WriteAsync(MakeEvent("dashboard-create-key"), CancellationToken.None);

        Assert.Equal("dashboard-create-key", Assert.Single(sink.Events).Action);
    }

    /// <summary>
    /// A completed channel has no future reader, so an attached writer must write through rather
    /// than discard. This is the re-attach-after-shutdown footgun: enqueueing would lose 100% of
    /// audit silently.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_AfterChannelCompleted_WritesThroughInsteadOfDiscarding()
    {
        CountingAuditSink sink = new();
        (ChannelAuditWriter writer, _) = CreateWriterAndDrain(sink);
        writer.AttachDrain();
        writer.CompleteWriting();

        await writer.WriteAsync(MakeEvent("constraint-denied"), CancellationToken.None);

        Assert.Equal("constraint-denied", Assert.Single(sink.Events).Action);
        Assert.Equal(0, writer.DroppedCount);
    }

    /// <summary>
    /// The retention sweep deletes rows older than <c>AuditRetentionDays</c>, measured from the
    /// injected clock, and runs once at startup after the one-time table bootstrap.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StartAsync_BootstrapsTableThenSweepsRetentionAtConfiguredCutoff()
    {
        CountingAuditSink sink = new();
        FakeTimeProvider clock = new(new DateTimeOffset(2026, 8, 15, 12, 0, 0, TimeSpan.Zero));
        (_, AuditDrainService drain) = CreateWriterAndDrain(
            sink,
            new SecurityOptions { AuditRetentionDays = 30 },
            clock);

        await drain.StartAsync(CancellationToken.None);
        await drain.StopAsync(CancellationToken.None);

        Assert.Equal(1, sink.EnsureInitializedCalls);
        Assert.Equal(new DateTimeOffset(2026, 7, 16, 12, 0, 0, TimeSpan.Zero), Assert.Single(sink.DeleteCutoffs));
    }

    /// <summary>
    /// The sweep compares instants, not stored text, and never deletes a row it cannot date.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task SqliteStore_DeleteOlderThan_ComparesInstantsAndKeepsUnparseableRows()
    {
        (SqliteCanonicalAuditStore store, AuthSqliteConnectionFactory factory) = CreateStore();

        DateTimeOffset cutoff = new(2026, 5, 17, 12, 0, 0, TimeSpan.Zero);

        await store.InsertBatchAsync(
            [
                MakeEvent("swept", new DateTimeOffset(2026, 5, 17, 11, 0, 0, TimeSpan.Zero)),
                MakeEvent("kept-fresh", new DateTimeOffset(2026, 5, 18, 0, 0, 0, TimeSpan.Zero)),
            ],
            CancellationToken.None);

        // Both discriminating rows are written as raw text, because AuditEvent.OccurredAtUtc
        // normalizes to UTC and so cannot express them. They stand for audit that reached the
        // table any other way — a repair script, an older schema, a future producer.
        //
        // 09:00 at -05:00 is 14:00 UTC, two hours AFTER the cutoff, yet its text sorts BEFORE
        // "2026-05-17T12:00:00.0000000+00:00". A lexicographic sweep deletes it; comparing
        // instants keeps it.
        await InsertRawRowAsync(factory, "kept-offset", "2026-05-17T09:00:00.1234567-05:00");

        // Undateable audit is kept, never guessed at. This text sorts BELOW the cutoff, so a
        // lexicographic sweep deletes it, while datetime() yields NULL and leaves it alone.
        await InsertRawRowAsync(factory, "kept-unparseable", "0000-not-a-timestamp");

        int deleted = await store.DeleteOlderThanAsync(cutoff, CancellationToken.None);

        Assert.Equal(1, deleted);
        Assert.Equal(
            ["kept-fresh", "kept-offset", "kept-unparseable"],
            (await ListActionsAsync(factory)).OrderBy(action => action, StringComparer.Ordinal));
    }

    /// <summary>
    /// The sweep above deliberately preserves rows it cannot date, so the dashboard's recent-audit
    /// view is guaranteed to meet one eventually. Reading must therefore be defensive: the
    /// undateable row is reported at <see cref="DateTimeOffset.MinValue"/> with every other column
    /// intact, rather than one bad row throwing the whole page away.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task SqliteStore_ListRecent_WithUnparseableRow_ReportsMinValueInsteadOfThrowing()
    {
        (SqliteCanonicalAuditStore store, AuthSqliteConnectionFactory factory) = CreateStore();

        DateTimeOffset occurred = new(2026, 5, 18, 0, 0, 0, TimeSpan.Zero);
        await store.InsertAsync(MakeEvent("dateable", occurred), CancellationToken.None);
        await InsertRawRowAsync(factory, "undateable", "0000-not-a-timestamp");

        IReadOnlyList<AuditEvent> recent = await store.ListRecentAsync(10, CancellationToken.None);

        Assert.Equal(2, recent.Count);
        Assert.Equal(occurred, recent.Single(auditEvent => auditEvent.Action == "dateable").OccurredAtUtc);

        AuditEvent undateable = recent.Single(auditEvent => auditEvent.Action == "undateable");
        Assert.Equal(DateTimeOffset.MinValue, undateable.OccurredAtUtc);
        Assert.Equal("operator01", undateable.Actor);
        Assert.Equal(AuditOutcome.Denied, undateable.Outcome);
    }

    private static (ChannelAuditWriter Writer, AuditDrainService Drain) CreateWriterAndDrain(
        IAuditEventSink sink,
        SecurityOptions? security = null,
        TimeProvider? timeProvider = null)
    {
        ChannelAuditWriter writer = new(
            new CanonicalAuditWriter(sink, NullLogger<CanonicalAuditWriter>.Instance),
            NullLogger<ChannelAuditWriter>.Instance);
        AuditDrainService drain = new(
            writer,
            sink,
            security ?? new SecurityOptions(),
            timeProvider ?? TimeProvider.System,
            NullLogger<AuditDrainService>.Instance);
        return (writer, drain);
    }

    private (SqliteCanonicalAuditStore Store, AuthSqliteConnectionFactory Factory) CreateStore()
    {
        TempDatabaseDirectory directory = TempDatabaseDirectory.Create("mxgateway-channel-audit");
        _tempDirectories.Add(directory);
        AuthSqliteConnectionFactory factory = new(directory.DatabasePath());
        return (new SqliteCanonicalAuditStore(factory), factory);
    }

    // Writes an audit row whose occurred_at_utc bypasses the store's DateTimeOffset formatting,
    // so the sweep can be shown to leave undateable audit alone.
    private static async Task InsertRawRowAsync(
        AuthSqliteConnectionFactory factory,
        string action,
        string occurredAtUtc)
    {
        await using SqliteConnection connection = await factory.OpenConnectionAsync(CancellationToken.None);
        await using SqliteCommand command = connection.CreateCommand();
        command.CommandText =
            """
            INSERT INTO audit_event (event_id, occurred_at_utc, actor, action, outcome)
            VALUES ($event_id, $occurred_at_utc, 'operator01', $action, 'Denied');
            """;
        command.Parameters.AddWithValue("$event_id", Guid.NewGuid().ToString());
        command.Parameters.AddWithValue("$occurred_at_utc", occurredAtUtc);
        command.Parameters.AddWithValue("$action", action);

        await command.ExecuteNonQueryAsync(CancellationToken.None);
    }

    // Reads actions straight from SQL so the sweep assertion depends on the DELETE alone, with no
    // opinion from the store's read path about how an undateable row is surfaced.
    private static async Task<List<string>> ListActionsAsync(AuthSqliteConnectionFactory factory)
    {
        await using SqliteConnection connection = await factory.OpenConnectionAsync(CancellationToken.None);
        await using SqliteCommand command = connection.CreateCommand();
        command.CommandText = "SELECT action FROM audit_event;";

        List<string> actions = [];

        await using SqliteDataReader reader = await command.ExecuteReaderAsync(CancellationToken.None);
        while (await reader.ReadAsync(CancellationToken.None))
        {
            actions.Add(reader.GetString(0));
        }

        return actions;
    }

    private static AuditEvent MakeEvent(string action, DateTimeOffset? occurredAtUtc = null) => new()
    {
        EventId = Guid.NewGuid(),
        OccurredAtUtc = occurredAtUtc ?? DateTimeOffset.UtcNow,
        Actor = "operator01",
        Action = action,
        Outcome = AuditOutcome.Denied,
        Category = "ApiKey",
    };

    /// <summary>Clears SQLite pools and deletes every temporary directory created by this test.</summary>
    public void Dispose()
    {
        foreach (TempDatabaseDirectory directory in _tempDirectories)
        {
            directory.Dispose();
        }

        _tempDirectories.Clear();
    }

    /// <summary>
    /// In-memory <see cref="IAuditEventSink"/> that records every call, so a test can prove the
    /// write path did NOT touch the sink and that the drain batched what it did write. Setting
    /// <see cref="PoisonAction"/> makes any write containing that action fail, modelling a row
    /// the store refuses.
    /// </summary>
    private sealed class CountingAuditSink : IAuditEventSink
    {
        private readonly Lock _gate = new();

        /// <summary>Gets the action whose presence makes a write fail, or null to accept everything.</summary>
        public string? PoisonAction { get; init; }

        /// <summary>Gets the events handed to the sink, in the order they were written.</summary>
        public List<AuditEvent> Events { get; } = [];

        /// <summary>Gets the size of each batch the sink accepted.</summary>
        public List<int> BatchSizes { get; } = [];

        /// <summary>Gets the cutoffs the sink was asked to delete below.</summary>
        public List<DateTimeOffset> DeleteCutoffs { get; } = [];

        /// <summary>Gets the number of accepted <see cref="InsertBatchAsync"/> calls.</summary>
        public int InsertBatchCalls => BatchSizes.Count;

        /// <summary>Gets the number of <see cref="EnsureInitializedAsync"/> calls.</summary>
        public int EnsureInitializedCalls { get; private set; }

        /// <inheritdoc />
        public Task EnsureInitializedAsync(CancellationToken cancellationToken)
        {
            lock (_gate)
            {
                EnsureInitializedCalls++;
            }

            return Task.CompletedTask;
        }

        /// <inheritdoc />
        public Task InsertAsync(AuditEvent auditEvent, CancellationToken cancellationToken) =>
            InsertBatchAsync([auditEvent], cancellationToken);

        /// <inheritdoc />
        public Task InsertBatchAsync(IReadOnlyList<AuditEvent> auditEvents, CancellationToken cancellationToken)
        {
            if (PoisonAction is not null
                && auditEvents.Any(auditEvent => auditEvent.Action == PoisonAction))
            {
                return Task.FromException(
                    new InvalidOperationException($"Refused a write of {auditEvents.Count} audit events."));
            }

            lock (_gate)
            {
                BatchSizes.Add(auditEvents.Count);
                Events.AddRange(auditEvents);
            }

            return Task.CompletedTask;
        }

        /// <inheritdoc />
        public Task<int> DeleteOlderThanAsync(DateTimeOffset cutoffUtc, CancellationToken cancellationToken)
        {
            lock (_gate)
            {
                DeleteCutoffs.Add(cutoffUtc);
            }

            return Task.FromResult(0);
        }
    }
}