mxaccessgw/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/SqliteCanonicalAuditStore.cs

using System.Globalization;
using Microsoft.Data.Sqlite;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.Auth.ApiKeys.Sqlite;

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

/// <summary>
/// MxGateway-owned, append-only SQLite store for canonical
/// <see cref="AuditEvent"/>s. It writes to a NEW <c>audit_event</c> table in the
/// SAME database file as the shared <c>ZB.MOM.WW.Auth.ApiKeys</c> stores: both share
/// the library's <see cref="AuthSqliteConnectionFactory"/> (so they target the same
/// <c>ApiKeyOptions.SqlitePath</c> with the same WAL/busy-timeout connection config).
/// </summary>
/// <remarks>
/// This store is the canonical sink for ALL MxGateway audit. The library's own
/// <c>api_key_audit</c> table is left in place but UNUSED after adoption — the library's
/// <c>IApiKeyAuditStore</c> registration is overridden by
/// <see cref="CanonicalForwardingApiKeyAuditStore"/>, which forwards onto this store via
/// <see cref="CanonicalAuditWriter"/>. The library's <c>schema_version</c> /
/// <c>api_key_audit</c> tables are not touched here.
/// <para>
/// The <c>audit_event</c> table is created idempotently, but the <c>CREATE TABLE IF NOT
/// EXISTS</c> is <em>latched</em>: <see cref="AuditDrainService"/> runs
/// <see cref="EnsureInitializedAsync"/> once at startup, and every later insert/list/delete
/// then skips the DDL round-trip. Keeping the (now free) check on each path rather than
/// dropping it means the store still self-bootstraps for callers that use it without the
/// hosted drain — the <c>apikey</c> CLI and the DI-only tests — regardless of migration
/// ordering. The latch is deliberately racy: a lost race merely re-runs an idempotent
/// <c>CREATE TABLE IF NOT EXISTS</c>, and a failure leaves the latch open so the next call
/// retries.
/// </para>
/// </remarks>
/// <param name="connectionFactory">Factory for connections to the shared auth database file.</param>
/// <param name="logger">
/// Optional logger for row-level read diagnostics. Optional because the store is also constructed
/// directly by the <c>apikey</c> CLI path and by DI-free tests, which have no logger to hand.
/// </param>
public sealed class SqliteCanonicalAuditStore(
    AuthSqliteConnectionFactory connectionFactory,
    ILogger<SqliteCanonicalAuditStore>? logger = null) : IAuditEventSink
{
    private const string CreateTableSql =
        """
        CREATE TABLE IF NOT EXISTS audit_event (
            event_id        TEXT PRIMARY KEY,
            occurred_at_utc TEXT NOT NULL,
            actor           TEXT NOT NULL,
            action          TEXT NOT NULL,
            outcome         TEXT NOT NULL,
            category        TEXT NULL,
            target          TEXT NULL,
            source_node     TEXT NULL,
            correlation_id  TEXT NULL,
            details_json    TEXT NULL
        );
        """;

    private const string InsertSql =
        """
        INSERT INTO audit_event
            (event_id, occurred_at_utc, actor, action, outcome,
             category, target, source_node, correlation_id, details_json)
        VALUES
            ($event_id, $occurred_at_utc, $actor, $action, $outcome,
             $category, $target, $source_node, $correlation_id, $details_json);
        """;

    /// <summary>0 until the <c>audit_event</c> table has been created at least once by this instance.</summary>
    private int _tableEnsured;

    /// <inheritdoc />
    public async Task EnsureInitializedAsync(CancellationToken cancellationToken)
    {
        await using SqliteConnection connection =
            await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);

        await EnsureTableAsync(connection, cancellationToken).ConfigureAwait(false);
    }

    /// <inheritdoc />
    public Task InsertAsync(AuditEvent auditEvent, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(auditEvent);

        return InsertBatchAsync([auditEvent], cancellationToken);
    }

    /// <inheritdoc />
    // One connection, one transaction and one prepared command for the whole batch: the drain
    // pays a single commit for up to AuditDrainService.MaxBatchSize events rather
    // than one round-trip per event.
    public async Task InsertBatchAsync(IReadOnlyList<AuditEvent> auditEvents, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(auditEvents);

        if (auditEvents.Count == 0)
        {
            return;
        }

        await using SqliteConnection connection =
            await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);

        await EnsureTableAsync(connection, cancellationToken).ConfigureAwait(false);

        await using SqliteTransaction transaction =
            (SqliteTransaction)await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);

        await using (SqliteCommand command = connection.CreateCommand())
        {
            command.Transaction = transaction;
            command.CommandText = InsertSql;

            SqliteParameter eventId = command.Parameters.Add("$event_id", SqliteType.Text);
            SqliteParameter occurredAtUtc = command.Parameters.Add("$occurred_at_utc", SqliteType.Text);
            SqliteParameter actor = command.Parameters.Add("$actor", SqliteType.Text);
            SqliteParameter action = command.Parameters.Add("$action", SqliteType.Text);
            SqliteParameter outcome = command.Parameters.Add("$outcome", SqliteType.Text);
            SqliteParameter category = command.Parameters.Add("$category", SqliteType.Text);
            SqliteParameter target = command.Parameters.Add("$target", SqliteType.Text);
            SqliteParameter sourceNode = command.Parameters.Add("$source_node", SqliteType.Text);
            SqliteParameter correlationId = command.Parameters.Add("$correlation_id", SqliteType.Text);
            SqliteParameter detailsJson = command.Parameters.Add("$details_json", SqliteType.Text);

            foreach (AuditEvent auditEvent in auditEvents)
            {
                ArgumentNullException.ThrowIfNull(auditEvent);

                eventId.Value = auditEvent.EventId.ToString();
                occurredAtUtc.Value = auditEvent.OccurredAtUtc.ToString("O", CultureInfo.InvariantCulture);
                actor.Value = auditEvent.Actor;
                action.Value = auditEvent.Action;
                outcome.Value = auditEvent.Outcome.ToString();
                category.Value = (object?)auditEvent.Category ?? DBNull.Value;
                target.Value = (object?)auditEvent.Target ?? DBNull.Value;
                sourceNode.Value = (object?)auditEvent.SourceNode ?? DBNull.Value;
                correlationId.Value = (object?)auditEvent.CorrelationId?.ToString() ?? DBNull.Value;
                detailsJson.Value = (object?)auditEvent.DetailsJson ?? DBNull.Value;

                await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
            }
        }

        await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
    }

    /// <inheritdoc />
    // The comparison goes through SQLite's datetime() rather than comparing the stored
    // ISO-8601 text directly. Text comparison is only correct while every row is UTC-normalized
    // ISO-8601 — which AuditEvent.OccurredAtUtc guarantees for rows written through
    // this store, but not for rows that entered the table any other way (a repair script, an
    // older schema, a future producer). On a mixed-format column a text comparison silently
    // deletes live audit: 2026-05-17T09:00:00-05:00 is two hours AFTER a
    // 2026-05-17T12:00:00+00:00 cutoff yet sorts before it. Comparing instants is correct
    // regardless of how the text got there, and anything datetime() cannot parse yields
    // NULL and is therefore never deleted — audit that cannot be dated is kept, not swept.
    public async Task<int> DeleteOlderThanAsync(DateTimeOffset cutoffUtc, CancellationToken cancellationToken)
    {
        await using SqliteConnection connection =
            await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);

        await EnsureTableAsync(connection, cancellationToken).ConfigureAwait(false);

        await using SqliteCommand command = connection.CreateCommand();
        command.CommandText =
            """
            DELETE FROM audit_event
            WHERE datetime(occurred_at_utc) < datetime($cutoff);
            """;
        command.Parameters.AddWithValue(
            "$cutoff",
            cutoffUtc.ToUniversalTime().ToString("O", CultureInfo.InvariantCulture));

        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
    }

    /// <summary>Returns the most recent canonical audit events, newest first.</summary>
    /// <param name="limit">Maximum number of events to return.</param>
    /// <param name="cancellationToken">Token to observe for cancellation.</param>
    /// <returns>The most recent audit events, up to <paramref name="limit"/>, newest first.</returns>
    public async Task<IReadOnlyList<AuditEvent>> ListRecentAsync(int limit, CancellationToken cancellationToken)
    {
        if (limit <= 0)
        {
            return [];
        }

        await using SqliteConnection connection =
            await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);

        await EnsureTableAsync(connection, cancellationToken).ConfigureAwait(false);

        await using SqliteCommand command = connection.CreateCommand();
        command.CommandText =
            """
            SELECT event_id, occurred_at_utc, actor, action, outcome,
                   category, target, source_node, correlation_id, details_json
            FROM audit_event
            ORDER BY rowid DESC
            LIMIT $limit;
            """;
        command.Parameters.AddWithValue("$limit", limit);

        List<AuditEvent> events = [];

        await using SqliteDataReader reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);

        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
        {
            events.Add(new AuditEvent
            {
                EventId = Guid.Parse(reader.GetString(0)),
                OccurredAtUtc = ParseUtcOrMinValue(reader.GetString(1), reader.GetString(0)),
                Actor = reader.GetString(2),
                Action = reader.GetString(3),
                Outcome = Enum.Parse<AuditOutcome>(reader.GetString(4)),
                Category = reader.IsDBNull(5) ? null : reader.GetString(5),
                Target = reader.IsDBNull(6) ? null : reader.GetString(6),
                SourceNode = reader.IsDBNull(7) ? null : reader.GetString(7),
                CorrelationId = reader.IsDBNull(8) ? null : Guid.Parse(reader.GetString(8)),
                DetailsJson = reader.IsDBNull(9) ? null : reader.GetString(9),
            });
        }

        return events;
    }

    // Latched bootstrap: after the first success this is a single volatile read, so the DDL
    // round-trip is paid once per process rather than once per audit write.
    private async Task EnsureTableAsync(SqliteConnection connection, CancellationToken cancellationToken)
    {
        if (Volatile.Read(ref _tableEnsured) == 1)
        {
            return;
        }

        await using SqliteCommand command = connection.CreateCommand();
        command.CommandText = CreateTableSql;
        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);

        Volatile.Write(ref _tableEnsured, 1);
    }

    // Reading is defensive where writing is not: an insert always round-trips "O", but the table is
    // append-only shared state that an operator (or a future migration) can put an unparseable
    // timestamp into, and the retention sweep deliberately keeps such a row — SQLite's datetime()
    // yields NULL for it, so the DELETE's comparison is never true. A throwing Parse here would let
    // that single row take out the dashboard's whole recent-audit view. MinValue instead sorts the
    // row to the far past and keeps every other column readable, which is what an operator looking
    // at the view actually needs.
    private DateTimeOffset ParseUtcOrMinValue(string value, string eventId)
    {
        if (DateTimeOffset.TryParse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTimeOffset parsed))
        {
            return parsed;
        }

        // Debug, not warning: the row is still returned and the timestamp text itself is not logged
        // (audit rows are not secrets, but the value is attacker-influenceable in the worst case).
        logger?.LogDebug(
            "Audit event {EventId} has an unparseable occurred_at_utc; reporting it as DateTimeOffset.MinValue.",
            eventId);

        return DateTimeOffset.MinValue;
    }
}