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

using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.AuditLog.Site;
using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
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>
/// C4 (Task 2.5) hot-path + drain tests for <see cref="SqliteAuditWriter"/>'s
/// two-table site schema. Exercise the Channel-based enqueue, the background
/// writer's per-event canonical(<c>audit_event</c>) + sidecar
/// (<c>audit_forward_state</c>) INSERTs, duplicate-EventId swallowing, the
/// <c>IsCachedKind</c> drain split, the four reads, and the
/// <see cref="SqliteAuditWriter.MarkForwardedAsync"/> /
/// <see cref="SqliteAuditWriter.MarkReconciledAsync"/> sidecar flips.
/// </summary>
public class SqliteAuditWriterWriteTests
{
    private static (SqliteAuditWriter writer, string dataSource) CreateWriter(
        string testName,
        int? channelCapacity = null,
        INodeIdentityProvider? nodeIdentity = null,
        int? flushIntervalMs = null)
    {
        var dataSource = $"file:{testName}-{Guid.NewGuid():N}?mode=memory&cache=shared";
        var opts = new SqliteAuditWriterOptions { DatabasePath = dataSource };
        if (channelCapacity is int cap)
        {
            opts.ChannelCapacity = cap;
        }
        if (flushIntervalMs is int flush)
        {
            opts.FlushIntervalMs = flush;
        }

        // Default identity provider returns null — existing tests pre-date
        // SourceNode stamping and have no expectation about it. New stamping
        // tests pass a real provider via the parameter.
        var identity = nodeIdentity ?? new FakeNodeIdentityProvider();

        var writer = new SqliteAuditWriter(
            Options.Create(opts),
            NullLogger<SqliteAuditWriter>.Instance,
            identity,
            connectionStringOverride: $"Data Source={dataSource};Cache=Shared");
        return (writer, dataSource);
    }

    private static SqliteConnection OpenVerifierConnection(string dataSource)
    {
        var connection = new SqliteConnection($"Data Source={dataSource};Cache=Shared");
        connection.Open();
        return connection;
    }

    /// <summary>
    /// Reads the sidecar <c>ForwardState</c> for one EventId (the column moved off
    /// the single legacy table onto <c>audit_forward_state</c> in C4).
    /// </summary>
    private static string? ReadForwardState(string dataSource, Guid eventId)
    {
        using var connection = OpenVerifierConnection(dataSource);
        using var cmd = connection.CreateCommand();
        cmd.CommandText = "SELECT ForwardState FROM audit_forward_state WHERE EventId = $id;";
        cmd.Parameters.AddWithValue("$id", eventId.ToString());
        return cmd.ExecuteScalar() as string;
    }

    /// <summary>Sidecar ForwardState → row-count, grouped (replaces the legacy single-table GROUP BY).</summary>
    private static Dictionary<string, long> ForwardStateCounts(string dataSource)
    {
        using var connection = OpenVerifierConnection(dataSource);
        using var cmd = connection.CreateCommand();
        cmd.CommandText =
            "SELECT ForwardState, COUNT(*) FROM audit_forward_state GROUP BY ForwardState;";
        using var reader = cmd.ExecuteReader();
        var byState = new Dictionary<string, long>();
        while (reader.Read())
        {
            byState[reader.GetString(0)] = reader.GetInt64(1);
        }
        return byState;
    }

    // C4 (Task 2.5): build the canonical ZB.MOM.WW.Audit.AuditEvent via the shared
    // factory. The SQLite writer stores the 10 canonical fields directly in
    // audit_event and writes a Pending sidecar row into audit_forward_state, with
    // IsCachedKind precomputed from the event's Kind. Reads recompose the canonical
    // record directly from audit_event's columns.
    private static AuditEvent NewEvent(
        Guid? id = null,
        DateTime? occurredAtUtc = null,
        Guid? executionId = null,
        string? sourceNode = null)
        => ScadaBridgeAuditEventFactory.Create(
            channel: AuditChannel.ApiOutbound,
            kind: AuditKind.ApiCall,
            status: AuditStatus.Delivered,
            eventId: id ?? Guid.NewGuid(),
            occurredAtUtc: occurredAtUtc ?? DateTime.UtcNow,
            executionId: executionId,
            sourceNode: sourceNode);

    /// <summary>A cached-lifecycle event (IsCachedKind=1) — drains via the cached read surface.</summary>
    private static AuditEvent NewCachedEvent(
        Guid? id = null,
        DateTime? occurredAtUtc = null,
        AuditKind kind = AuditKind.ApiCallCached)
        // Status is independent of IsCachedKind (which is derived from Kind);
        // Submitted is the natural first-row status for a cached lifecycle.
        => ScadaBridgeAuditEventFactory.Create(
            channel: AuditChannel.ApiOutbound,
            kind: kind,
            status: AuditStatus.Submitted,
            eventId: id ?? Guid.NewGuid(),
            occurredAtUtc: occurredAtUtc ?? DateTime.UtcNow);

    [Fact]
    public async Task WriteAsync_FreshEvent_PersistsCanonical_And_SidecarPending()
    {
        var (writer, dataSource) = CreateWriter(nameof(WriteAsync_FreshEvent_PersistsCanonical_And_SidecarPending));
        await using var _ = writer;

        var evt = NewEvent();
        await writer.WriteAsync(evt);

        // Canonical row landed in audit_event.
        using var connection = OpenVerifierConnection(dataSource);
        using var eventCmd = connection.CreateCommand();
        eventCmd.CommandText = "SELECT Action FROM audit_event WHERE EventId = $id;";
        eventCmd.Parameters.AddWithValue("$id", evt.EventId.ToString());
        Assert.Equal(evt.Action, eventCmd.ExecuteScalar() as string);

        // Sidecar row landed Pending.
        Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, evt.EventId));
    }

    [Fact]
    public async Task WriteAsync_Roundtrips_Canonical_Fields_Through_Read()
    {
        var (writer, _) = CreateWriter(nameof(WriteAsync_Roundtrips_Canonical_Fields_Through_Read));
        await using var _w = writer;

        var evt = NewEvent() with { Target = "target-1", Actor = "user-1" };
        await writer.WriteAsync(evt);

        var rows = await writer.ReadPendingAsync(limit: 10);
        var row = Assert.Single(rows);

        Assert.Equal(evt.EventId, row.EventId);
        Assert.Equal(evt.OccurredAtUtc, row.OccurredAtUtc);
        Assert.Equal("user-1", row.Actor);
        Assert.Equal(evt.Action, row.Action);
        Assert.Equal(evt.Outcome, row.Outcome);
        Assert.Equal(evt.Category, row.Category);
        Assert.Equal("target-1", row.Target);
        Assert.Equal(evt.CorrelationId, row.CorrelationId);
        // DetailsJson is stored verbatim and round-trips byte-for-byte.
        Assert.Equal(evt.DetailsJson, row.DetailsJson);
    }

    [Fact]
    public async Task WriteAsync_Concurrent_1000Calls_All_Persist_NoExceptions()
    {
        var (writer, dataSource) = CreateWriter(nameof(WriteAsync_Concurrent_1000Calls_All_Persist_NoExceptions));
        await using var _ = writer;

        var events = Enumerable.Range(0, 1000).Select(_ => NewEvent()).ToList();

        await Parallel.ForEachAsync(events, new ParallelOptions { MaxDegreeOfParallelism = 16 },
            async (evt, ct) => await writer.WriteAsync(evt, ct));

        using var connection = OpenVerifierConnection(dataSource);
        using var eventCmd = connection.CreateCommand();
        eventCmd.CommandText = "SELECT COUNT(*) FROM audit_event;";
        Assert.Equal(1000, Convert.ToInt64(eventCmd.ExecuteScalar()));

        // Every canonical row has its matching sidecar row.
        using var sidecarCmd = connection.CreateCommand();
        sidecarCmd.CommandText = "SELECT COUNT(*) FROM audit_forward_state;";
        Assert.Equal(1000, Convert.ToInt64(sidecarCmd.ExecuteScalar()));
    }

    // ----- FlushIntervalMs soft-flush coalescing (arch-review remediation WP1.2) ----- //

    /// <summary>
    /// A trickle of events — well under <see cref="SqliteAuditWriterOptions.BatchSize"/>,
    /// spaced closer together than <see cref="SqliteAuditWriterOptions.FlushIntervalMs"/> —
    /// must coalesce into ONE transaction rather than fsyncing once per event. This is the
    /// bug WP1.2 fixes: <c>FlushIntervalMs</c> was validated but never read by the writer
    /// loop, so every trickle-rate write incurred its own commit (and, under SQLite's
    /// default synchronous=FULL, its own fsync).
    /// </summary>
    [Fact]
    public async Task WriteAsync_TrickleLoad_CoalescesIntoOneTransaction()
    {
        var (writer, dataSource) = CreateWriter(
            nameof(WriteAsync_TrickleLoad_CoalescesIntoOneTransaction),
            flushIntervalMs: 500);
        await using var _ = writer;

        const int count = 5;
        var writeTasks = new List<Task>(count);
        for (int i = 0; i < count; i++)
        {
            writeTasks.Add(writer.WriteAsync(NewEvent()));
            await Task.Delay(15); // spaced well under the 500ms flush window
        }
        await Task.WhenAll(writeTasks);

        using var connection = OpenVerifierConnection(dataSource);
        using var cmd = connection.CreateCommand();
        cmd.CommandText = "SELECT COUNT(*) FROM audit_event;";
        Assert.Equal(count, Convert.ToInt64(cmd.ExecuteScalar()));

        // The defining assertion: all 5 trickled-in writes landed in a single
        // FlushBatch transaction, not up to 5 separate commits.
        Assert.Equal(1, writer.FlushCountForTests);
    }

    /// <summary>
    /// The flip side: two events spaced FURTHER apart than
    /// <see cref="SqliteAuditWriterOptions.FlushIntervalMs"/> must NOT be held open
    /// waiting for a partner — the deadline bounds worst-case added latency, it does
    /// not turn into an unbounded debounce.
    /// </summary>
    [Fact]
    public async Task WriteAsync_EventsSpacedBeyondFlushInterval_FlushSeparately()
    {
        var (writer, dataSource) = CreateWriter(
            nameof(WriteAsync_EventsSpacedBeyondFlushInterval_FlushSeparately),
            flushIntervalMs: 50);
        await using var _ = writer;

        await writer.WriteAsync(NewEvent());
        await Task.Delay(300); // well beyond the 50ms flush window
        await writer.WriteAsync(NewEvent());

        using var connection = OpenVerifierConnection(dataSource);
        using var cmd = connection.CreateCommand();
        cmd.CommandText = "SELECT COUNT(*) FROM audit_event;";
        Assert.Equal(2, Convert.ToInt64(cmd.ExecuteScalar()));

        Assert.Equal(2, writer.FlushCountForTests);
    }

    [Fact]
    public async Task WriteAsync_DuplicateEventId_FirstWriteWins_NoException()
    {
        var (writer, dataSource) = CreateWriter(nameof(WriteAsync_DuplicateEventId_FirstWriteWins_NoException));
        await using var _ = writer;

        var sharedId = Guid.NewGuid();
        var first = NewEvent(sharedId) with { Target = "first" };
        var second = NewEvent(sharedId) with { Target = "second" };

        await writer.WriteAsync(first);
        await writer.WriteAsync(second);

        using var connection = OpenVerifierConnection(dataSource);
        using var countCmd = connection.CreateCommand();
        countCmd.CommandText = "SELECT COUNT(*) FROM audit_event WHERE EventId = $id;";
        countCmd.Parameters.AddWithValue("$id", sharedId.ToString());
        Assert.Equal(1, Convert.ToInt64(countCmd.ExecuteScalar()));

        // The sidecar likewise gained exactly one row (the canonical PK throws
        // before the sidecar insert runs, so neither table double-inserts).
        using var sidecarCmd = connection.CreateCommand();
        sidecarCmd.CommandText = "SELECT COUNT(*) FROM audit_forward_state WHERE EventId = $id;";
        sidecarCmd.Parameters.AddWithValue("$id", sharedId.ToString());
        Assert.Equal(1, Convert.ToInt64(sidecarCmd.ExecuteScalar()));

        using var targetCmd = connection.CreateCommand();
        targetCmd.CommandText = "SELECT Target FROM audit_event WHERE EventId = $id;";
        targetCmd.Parameters.AddWithValue("$id", sharedId.ToString());
        Assert.Equal("first", targetCmd.ExecuteScalar() as string);
    }

    [Fact]
    public async Task WriteAsync_ForcesSidecarForwardStatePending()
    {
        var (writer, dataSource) = CreateWriter(nameof(WriteAsync_ForcesSidecarForwardStatePending));
        await using var _ = writer;

        // C4 (Task 2.5): ForwardState is not a field on the canonical record; a
        // fresh event's sidecar row defaults to Pending on INSERT.
        var evt = NewEvent();
        await writer.WriteAsync(evt);

        Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, evt.EventId));
    }

    // ----- IsCachedKind drain split (precomputed at insert) ----- //

    [Fact]
    public async Task WriteAsync_CachedKind_SetsIsCachedKind_1_NonCached_0()
    {
        var (writer, dataSource) = CreateWriter(nameof(WriteAsync_CachedKind_SetsIsCachedKind_1_NonCached_0));
        await using var _ = writer;

        var cached = NewCachedEvent();      // ApiCallCached → cached
        var nonCached = NewEvent();          // ApiCall → not cached
        await writer.WriteAsync(cached);
        await writer.WriteAsync(nonCached);

        using var connection = OpenVerifierConnection(dataSource);
        using var cmd = connection.CreateCommand();
        cmd.CommandText = "SELECT IsCachedKind FROM audit_forward_state WHERE EventId = $id;";
        var p = cmd.Parameters.Add("$id", SqliteType.Text);

        p.Value = cached.EventId.ToString();
        Assert.Equal(1L, Convert.ToInt64(cmd.ExecuteScalar()));

        p.Value = nonCached.EventId.ToString();
        Assert.Equal(0L, Convert.ToInt64(cmd.ExecuteScalar()));
    }

    [Theory]
    [InlineData(AuditKind.CachedSubmit)]
    [InlineData(AuditKind.ApiCallCached)]
    [InlineData(AuditKind.DbWriteCached)]
    [InlineData(AuditKind.CachedResolve)]
    public async Task CachedKinds_DrainVia_ReadPendingCachedTelemetry_Not_ReadPending(AuditKind kind)
    {
        var (writer, _) = CreateWriter($"{nameof(CachedKinds_DrainVia_ReadPendingCachedTelemetry_Not_ReadPending)}-{kind}");
        await using var _w = writer;

        var cached = NewCachedEvent(kind: kind);
        await writer.WriteAsync(cached);

        // The cached kind appears in the cached read surface...
        var cachedRows = await writer.ReadPendingCachedTelemetryAsync(limit: 10);
        Assert.Single(cachedRows, r => r.EventId == cached.EventId);

        // ...and NOT in the audit-only read surface.
        var pendingRows = await writer.ReadPendingAsync(limit: 10);
        Assert.DoesNotContain(pendingRows, r => r.EventId == cached.EventId);
    }

    [Fact]
    public async Task NonCachedKind_DrainsVia_ReadPending_Not_ReadPendingCachedTelemetry()
    {
        var (writer, _) = CreateWriter(nameof(NonCachedKind_DrainsVia_ReadPending_Not_ReadPendingCachedTelemetry));
        await using var _w = writer;

        var nonCached = NewEvent(); // ApiCall — not a cached kind
        await writer.WriteAsync(nonCached);

        var pendingRows = await writer.ReadPendingAsync(limit: 10);
        Assert.Single(pendingRows, r => r.EventId == nonCached.EventId);

        var cachedRows = await writer.ReadPendingCachedTelemetryAsync(limit: 10);
        Assert.DoesNotContain(cachedRows, r => r.EventId == nonCached.EventId);
    }

    [Fact]
    public async Task ReadPendingAsync_Returns_OldestFirst_LimitedToN()
    {
        var (writer, _) = CreateWriter(nameof(ReadPendingAsync_Returns_OldestFirst_LimitedToN));
        await using var _writer = writer;

        var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
        var evts = new[]
        {
            NewEvent(occurredAtUtc: baseTime.AddSeconds(5)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(1)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(3)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(2)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(4)),
        };

        foreach (var e in evts)
        {
            await writer.WriteAsync(e);
        }

        var rows = await writer.ReadPendingAsync(limit: 3);

        Assert.Equal(3, rows.Count);
        Assert.Equal(baseTime.AddSeconds(1), rows[0].OccurredAtUtc);
        Assert.Equal(baseTime.AddSeconds(2), rows[1].OccurredAtUtc);
        Assert.Equal(baseTime.AddSeconds(3), rows[2].OccurredAtUtc);
    }

    [Fact]
    public async Task MarkForwardedAsync_FlipsSidecarRowsToForwarded()
    {
        var (writer, dataSource) = CreateWriter(nameof(MarkForwardedAsync_FlipsSidecarRowsToForwarded));
        await using var _ = writer;

        var ids = new[] { Guid.NewGuid(), Guid.NewGuid(), Guid.NewGuid() };
        foreach (var id in ids)
        {
            await writer.WriteAsync(NewEvent(id));
        }

        await writer.MarkForwardedAsync(ids);

        var byState = ForwardStateCounts(dataSource);
        Assert.Equal(3, byState[AuditForwardState.Forwarded.ToString()]);
        Assert.False(byState.ContainsKey(AuditForwardState.Pending.ToString()));
    }

    [Fact]
    public async Task MarkForwardedAsync_BumpsAttemptCount_And_StampsLastAttemptUtc()
    {
        var (writer, dataSource) = CreateWriter(nameof(MarkForwardedAsync_BumpsAttemptCount_And_StampsLastAttemptUtc));
        await using var _ = writer;

        var evt = NewEvent();
        await writer.WriteAsync(evt);

        await writer.MarkForwardedAsync(new[] { evt.EventId });

        using var connection = OpenVerifierConnection(dataSource);
        using var cmd = connection.CreateCommand();
        cmd.CommandText =
            "SELECT AttemptCount, LastAttemptUtc FROM audit_forward_state WHERE EventId = $id;";
        cmd.Parameters.AddWithValue("$id", evt.EventId.ToString());
        using var reader = cmd.ExecuteReader();
        Assert.True(reader.Read());
        Assert.Equal(1, reader.GetInt32(0));            // AttemptCount bumped 0 → 1
        Assert.False(reader.IsDBNull(1));               // LastAttemptUtc stamped
    }

    [Fact]
    public async Task MarkForwardedAsync_NonExistentId_NoThrow()
    {
        var (writer, _) = CreateWriter(nameof(MarkForwardedAsync_NonExistentId_NoThrow));
        await using var _writer = writer;

        var phantomIds = new[] { Guid.NewGuid(), Guid.NewGuid() };

        await writer.MarkForwardedAsync(phantomIds);
        // No assertion needed: the call must complete without throwing.
    }

    [Fact]
    public async Task ReadForwardedAsync_Returns_Only_Forwarded_Rows()
    {
        var (writer, _) = CreateWriter(nameof(ReadForwardedAsync_Returns_Only_Forwarded_Rows));
        await using var _w = writer;

        var forwarded = NewEvent();
        var pending = NewEvent();
        await writer.WriteAsync(forwarded);
        await writer.WriteAsync(pending);
        await writer.MarkForwardedAsync(new[] { forwarded.EventId });

        var rows = await writer.ReadForwardedAsync(limit: 10);
        var row = Assert.Single(rows);
        Assert.Equal(forwarded.EventId, row.EventId);
    }

    // ----- M6 reconciliation pull surface ----- //

    [Fact]
    public async Task ReadPendingSinceAsync_Returns_PendingAndForwarded_OldestFirst_LimitedToN()
    {
        var (writer, dataSource) = CreateWriter(nameof(ReadPendingSinceAsync_Returns_PendingAndForwarded_OldestFirst_LimitedToN));
        await using var _ = writer;

        var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
        var evts = new[]
        {
            NewEvent(occurredAtUtc: baseTime.AddSeconds(5)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(1)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(3)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(2)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(4)),
        };
        foreach (var e in evts) await writer.WriteAsync(e);

        // Flip half to Forwarded — they must still surface in the reconciliation pull
        // because central hasn't confirmed they were ingested yet.
        await writer.MarkForwardedAsync(new[] { evts[0].EventId, evts[2].EventId });

        var rows = await writer.ReadPendingSinceAsync(sinceUtc: DateTime.MinValue, batchSize: 3);

        Assert.Equal(3, rows.Count);
        Assert.Equal(baseTime.AddSeconds(1), rows[0].OccurredAtUtc);
        Assert.Equal(baseTime.AddSeconds(2), rows[1].OccurredAtUtc);
        Assert.Equal(baseTime.AddSeconds(3), rows[2].OccurredAtUtc);
    }

    [Fact]
    public async Task ReadPendingSinceAsync_ExcludesRowsOlderThanSinceUtc()
    {
        var (writer, _) = CreateWriter(nameof(ReadPendingSinceAsync_ExcludesRowsOlderThanSinceUtc));
        await using var _w = writer;

        var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
        var old = NewEvent(occurredAtUtc: baseTime.AddSeconds(-30));
        var newer1 = NewEvent(occurredAtUtc: baseTime.AddSeconds(10));
        var newer2 = NewEvent(occurredAtUtc: baseTime.AddSeconds(20));

        await writer.WriteAsync(old);
        await writer.WriteAsync(newer1);
        await writer.WriteAsync(newer2);

        var rows = await writer.ReadPendingSinceAsync(sinceUtc: baseTime, batchSize: 10);

        Assert.Equal(2, rows.Count);
        Assert.Contains(rows, r => r.EventId == newer1.EventId);
        Assert.Contains(rows, r => r.EventId == newer2.EventId);
        Assert.DoesNotContain(rows, r => r.EventId == old.EventId);
    }

    [Fact]
    public async Task ReadPendingSinceAsync_ExcludesReconciledRows()
    {
        var (writer, _) = CreateWriter(nameof(ReadPendingSinceAsync_ExcludesReconciledRows));
        await using var _w = writer;

        var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
        var pending = NewEvent(occurredAtUtc: baseTime);
        var reconciled = NewEvent(occurredAtUtc: baseTime.AddSeconds(1));

        await writer.WriteAsync(pending);
        await writer.WriteAsync(reconciled);
        await writer.MarkReconciledAsync(new[] { reconciled.EventId });

        var rows = await writer.ReadPendingSinceAsync(sinceUtc: DateTime.MinValue, batchSize: 10);

        Assert.Single(rows);
        Assert.Equal(pending.EventId, rows[0].EventId);
    }

    [Fact]
    public async Task ReadPendingSinceAsync_InvalidBatchSize_Throws()
    {
        var (writer, _) = CreateWriter(nameof(ReadPendingSinceAsync_InvalidBatchSize_Throws));
        await using var _w = writer;

        await Assert.ThrowsAsync<ArgumentOutOfRangeException>(
            () => writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 0));
        await Assert.ThrowsAsync<ArgumentOutOfRangeException>(
            () => writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: -3));
    }

    [Fact]
    public async Task MarkReconciledAsync_FlipsPendingAndForwarded_To_Reconciled()
    {
        var (writer, dataSource) = CreateWriter(nameof(MarkReconciledAsync_FlipsPendingAndForwarded_To_Reconciled));
        await using var _ = writer;

        var a = NewEvent();
        var b = NewEvent();
        var c = NewEvent();
        await writer.WriteAsync(a);
        await writer.WriteAsync(b);
        await writer.WriteAsync(c);

        // b is currently Forwarded; a and c are Pending.
        await writer.MarkForwardedAsync(new[] { b.EventId });

        await writer.MarkReconciledAsync(new[] { a.EventId, b.EventId, c.EventId });

        var byState = ForwardStateCounts(dataSource);
        Assert.Equal(3, byState[AuditForwardState.Reconciled.ToString()]);
        Assert.False(byState.ContainsKey(AuditForwardState.Pending.ToString()));
        Assert.False(byState.ContainsKey(AuditForwardState.Forwarded.ToString()));
    }

    [Fact]
    public async Task MarkReconciledAsync_Idempotent_LeavesAlreadyReconciledRowsUntouched()
    {
        var (writer, dataSource) = CreateWriter(nameof(MarkReconciledAsync_Idempotent_LeavesAlreadyReconciledRowsUntouched));
        await using var _ = writer;

        var a = NewEvent();
        await writer.WriteAsync(a);
        await writer.MarkReconciledAsync(new[] { a.EventId });
        // Re-call must not throw and must leave the single row Reconciled.
        await writer.MarkReconciledAsync(new[] { a.EventId });

        Assert.Equal(AuditForwardState.Reconciled.ToString(), ReadForwardState(dataSource, a.EventId));
    }

    [Fact]
    public async Task MarkReconciledAsync_NonExistentId_NoThrow()
    {
        var (writer, _) = CreateWriter(nameof(MarkReconciledAsync_NonExistentId_NoThrow));
        await using var _w = writer;

        await writer.MarkReconciledAsync(new[] { Guid.NewGuid(), Guid.NewGuid() });
        // Completes without throwing.
    }

    /// <summary>
    /// Fix 2 / M1 state guard: <see cref="SqliteAuditWriter.MarkForwardedAsync"/>
    /// must NOT demote a <see cref="AuditForwardState.Reconciled"/> row back to
    /// <see cref="AuditForwardState.Forwarded"/>. When a batch contains both a
    /// Pending ID and an already-Reconciled ID:
    /// <list type="bullet">
    ///   <item>the Pending row transitions to Forwarded (normal path)</item>
    ///   <item>the Reconciled row stays Reconciled (AttemptCount unchanged)</item>
    /// </list>
    /// This mirrors the idempotency guard already present on
    /// <see cref="SqliteAuditWriter.MarkReconciledAsync"/>.
    /// </summary>
    [Fact]
    public async Task MarkForwardedAsync_DoesNotDemoteReconciledRow_WhilePendingStillTransitions()
    {
        var (writer, dataSource) = CreateWriter(
            nameof(MarkForwardedAsync_DoesNotDemoteReconciledRow_WhilePendingStillTransitions));
        await using var _ = writer;

        var pending = NewEvent();
        var reconciled = NewEvent();

        await writer.WriteAsync(pending);
        await writer.WriteAsync(reconciled);

        // Advance reconciled through Forwarded → Reconciled so its AttemptCount = 1.
        await writer.MarkForwardedAsync(new[] { reconciled.EventId });
        await writer.MarkReconciledAsync(new[] { reconciled.EventId });

        // Verify the reconciled row's AttemptCount is 1 before the test call.
        using var conn = OpenVerifierConnection(dataSource);
        long reconciledAttemptBefore;
        using (var cmd = conn.CreateCommand())
        {
            cmd.CommandText =
                "SELECT AttemptCount FROM audit_forward_state WHERE EventId = $id;";
            cmd.Parameters.AddWithValue("$id", reconciled.EventId.ToString());
            reconciledAttemptBefore = Convert.ToInt64(cmd.ExecuteScalar());
        }
        Assert.Equal(1L, reconciledAttemptBefore);

        // Now call MarkForwardedAsync with BOTH IDs in the same batch.
        await writer.MarkForwardedAsync(new[] { pending.EventId, reconciled.EventId });

        // The Pending row must have transitioned to Forwarded.
        Assert.Equal(
            AuditForwardState.Forwarded.ToString(),
            ReadForwardState(dataSource, pending.EventId));

        // The Reconciled row must remain Reconciled — the state guard must have
        // excluded it from the UPDATE.
        Assert.Equal(
            AuditForwardState.Reconciled.ToString(),
            ReadForwardState(dataSource, reconciled.EventId));

        // AttemptCount on the Reconciled row must be unchanged (still 1, not 2).
        using (var cmd = conn.CreateCommand())
        {
            cmd.CommandText =
                "SELECT AttemptCount FROM audit_forward_state WHERE EventId = $id;";
            cmd.Parameters.AddWithValue("$id", reconciled.EventId.ToString());
            var attemptAfter = Convert.ToInt64(cmd.ExecuteScalar());
            Assert.Equal(reconciledAttemptBefore, attemptAfter);
        }
    }

    /// <summary>
    /// P8 (arch-review 04): a drain batch larger than SQLite's bound-parameter ceiling
    /// (999 on older builds) must still flip every row. The writer chunks the id list
    /// into ≤500-id UPDATEs inside one transaction, so 1200 ids in a single
    /// <see cref="SqliteAuditWriter.MarkForwardedAsync"/> call flip atomically — and
    /// <see cref="SqliteAuditWriter.MarkReconciledAsync"/> chunks the same way.
    /// </summary>
    [Fact]
    public async Task MarkForwardedThenReconciled_LargeBatch_ChunksParameters_AllFlip()
    {
        var (writer, dataSource) = CreateWriter(nameof(MarkForwardedThenReconciled_LargeBatch_ChunksParameters_AllFlip));
        await using var _ = writer;

        const int count = 1200; // > 999 and > the 500 chunk size, so multiple chunks fire
        var ids = new List<Guid>(count);
        for (int i = 0; i < count; i++)
        {
            var id = Guid.NewGuid();
            ids.Add(id);
            await writer.WriteAsync(NewEvent(id));
        }

        // One MarkForwardedAsync call over all 1200 ids — must not throw on the
        // parameter ceiling and must flip every sidecar row to Forwarded.
        await writer.MarkForwardedAsync(ids);

        var afterForward = ForwardStateCounts(dataSource);
        Assert.Equal(count, afterForward[AuditForwardState.Forwarded.ToString()]);
        Assert.False(afterForward.ContainsKey(AuditForwardState.Pending.ToString()));

        // Same for reconciliation over the full batch.
        await writer.MarkReconciledAsync(ids);

        var afterReconcile = ForwardStateCounts(dataSource);
        Assert.Equal(count, afterReconcile[AuditForwardState.Reconciled.ToString()]);
        Assert.False(afterReconcile.ContainsKey(AuditForwardState.Forwarded.ToString()));
    }

    // ----- ExecutionId (rides DetailsJson, recomposed via AsRow) ----- //

    [Fact]
    public async Task WriteAsync_NonNullExecutionId_RoundTrips()
    {
        var (writer, _) = CreateWriter(nameof(WriteAsync_NonNullExecutionId_RoundTrips));
        await using var _w = writer;

        var executionId = Guid.NewGuid();
        var evt = NewEvent(executionId: executionId);
        await writer.WriteAsync(evt);

        var rows = await writer.ReadPendingAsync(limit: 10);

        var row = Assert.Single(rows);
        Assert.Equal(executionId, row.AsRow().ExecutionId);
    }

    [Fact]
    public async Task WriteAsync_NullExecutionId_RoundTripsAsNull()
    {
        var (writer, _) = CreateWriter(nameof(WriteAsync_NullExecutionId_RoundTripsAsNull));
        await using var _w = writer;

        var evt = NewEvent(); // executionId defaults to null
        await writer.WriteAsync(evt);

        var rows = await writer.ReadPendingAsync(limit: 10);

        var row = Assert.Single(rows);
        Assert.Null(row.AsRow().ExecutionId);
    }

    // ----- SourceNode stamping (Tasks 11/12) ----- //

    [Fact]
    public async Task WriteAsync_StampsSourceNodeFromProvider_WhenEventHasNone()
    {
        var (writer, _) = CreateWriter(
            nameof(WriteAsync_StampsSourceNodeFromProvider_WhenEventHasNone),
            nodeIdentity: new FakeNodeIdentityProvider("node-a"));
        await using var _w = writer;

        var evt = NewEvent();
        Assert.Null(evt.SourceNode); // sanity check — fresh event has no SourceNode
        await writer.WriteAsync(evt);

        var rows = await writer.ReadPendingAsync(limit: 10);
        var row = Assert.Single(rows);
        Assert.Equal("node-a", row.SourceNode);
    }

    [Fact]
    public async Task WriteAsync_PreservesCallerProvidedSourceNode()
    {
        var (writer, _) = CreateWriter(
            nameof(WriteAsync_PreservesCallerProvidedSourceNode),
            nodeIdentity: new FakeNodeIdentityProvider("node-a"));
        await using var _w = writer;

        // Reconciled rows from another node arrive with their origin's
        // SourceNode already populated; the writer must preserve it.
        var evt = NewEvent(sourceNode: "node-z");
        await writer.WriteAsync(evt);

        var rows = await writer.ReadPendingAsync(limit: 10);
        var row = Assert.Single(rows);
        Assert.Equal("node-z", row.SourceNode);
    }

    [Fact]
    public async Task WriteAsync_LeavesSourceNodeNull_WhenProviderReturnsNull()
    {
        var (writer, _) = CreateWriter(
            nameof(WriteAsync_LeavesSourceNodeNull_WhenProviderReturnsNull),
            nodeIdentity: new FakeNodeIdentityProvider(nodeName: null));
        await using var _w = writer;

        var evt = NewEvent();
        await writer.WriteAsync(evt);

        var rows = await writer.ReadPendingAsync(limit: 10);
        var row = Assert.Single(rows);
        Assert.Null(row.SourceNode);
    }

    // ----- C4 hardening: safe enum-parse in MapRow ----- //

    /// <summary>
    /// C4 hardening (Task 2.5): a row whose stored <c>Outcome</c> column holds an
    /// unknown/renamed enum string must NOT fault the read path; it degrades
    /// gracefully to <see cref="AuditOutcome.Success"/> (the safe
    /// <see cref="AuditRowProjection.ParseEnum{TEnum}"/> fallback). The read is
    /// exercised via the public <see cref="SqliteAuditWriter.ReadPendingAsync"/>
    /// surface which calls the private <c>MapRow</c>.
    /// </summary>
    [Fact]
    public async Task ReadPendingAsync_UnknownOutcomeString_DoesNotThrow_YieldsFallback()
    {
        var (writer, dataSource) = CreateWriter(
            nameof(ReadPendingAsync_UnknownOutcomeString_DoesNotThrow_YieldsFallback));
        await using var _ = writer;

        var evt = NewEvent();
        await writer.WriteAsync(evt);

        // Tamper: overwrite the canonical Outcome column with a string that is not
        // a declared AuditOutcome member name.
        using (var conn = OpenVerifierConnection(dataSource))
        using (var cmd = conn.CreateCommand())
        {
            cmd.CommandText = "UPDATE audit_event SET Outcome = 'RenamedOutcome99' WHERE EventId = $id;";
            cmd.Parameters.AddWithValue("$id", evt.EventId.ToString());
            cmd.ExecuteNonQuery();
        }

        // Must not throw (a raw Enum.Parse would throw ArgumentException).
        var rows = await writer.ReadPendingAsync(limit: 10);

        var row = Assert.Single(rows);
        Assert.Equal(AuditOutcome.Success, row.Outcome);
    }

    // ── Task 24: pin the inclusive (>=) reconciliation cursor boundary ──

    [Fact]
    public async Task ReadPendingSince_BoundaryTimestamp_IsInclusive_EventIdDedupAbsorbsDuplicates()
    {
        // Two rows share one OccurredAtUtc; a pull whose sinceUtc equals that
        // timestamp MUST return both (inclusive >= read). Central reconciliation
        // dedups on EventId (insert-if-not-exists), so re-reading the boundary row
        // is a safe no-op — but an exclusive (>) read would silently and
        // permanently lose the same-timestamp row. This test pins the >= contract.
        var (writer, _) = CreateWriter(nameof(ReadPendingSince_BoundaryTimestamp_IsInclusive_EventIdDedupAbsorbsDuplicates));
        await using var _w = writer;

        var boundary = DateTime.UtcNow;
        await writer.WriteAsync(NewEvent(id: Guid.NewGuid(), occurredAtUtc: boundary));
        await writer.WriteAsync(NewEvent(id: Guid.NewGuid(), occurredAtUtc: boundary));

        var rows = await writer.ReadPendingSinceAsync(boundary, batchSize: 10);

        Assert.Equal(2, rows.Count);
    }
}