ScadaBridge/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogger.cs

using System.Threading.Channels;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.LocalDb;

namespace ZB.MOM.WW.ScadaBridge.SiteEventLogging;

/// <summary>
/// Records operational events into the consolidated site database (LocalDb Phase 1).
/// Only the active node generates events, but <c>site_events</c> is a replicated table:
/// when the pair has a peer configured, the standby holds the same history and a failover
/// no longer starts a fresh log. With no peer configured the database is simply local —
/// replication is opt-in via <c>LocalDb:Replication:PeerAddress</c>.
/// </summary>
/// <remarks>
/// <para>
/// A single <see cref="SqliteConnection"/> is owned here and is NOT thread-safe.
/// All access — recording, querying, purging — must be funnelled through
/// <see cref="WithConnection"/>, which serialises callers on a shared lock.
/// </para>
/// <para>
/// Event recording is offloaded to a dedicated background writer thread (fed by a
/// <em>bounded</em> <see cref="Channel{T}"/>; capacity <see cref="SiteEventLogOptions.WriteQueueCapacity"/>,
/// default 10 000, overflow <see cref="BoundedChannelFullMode.DropOldest"/>).
/// <see cref="LogEventAsync"/> only validates its arguments and enqueues, so callers —
/// typically Akka actor threads on hot paths — never block on disk I/O or on
/// contention for the write lock. The returned <see cref="Task"/> completes once the
/// event is durably persisted and faults if the write fails. When a queued
/// event is evicted to make room for a newer one, that event's Task
/// is faulted with <see cref="InvalidOperationException"/> and
/// <see cref="FailedWriteCount"/> is incremented so the drop is observable.
/// </para>
/// </remarks>
public class SiteEventLogger : ISiteEventLogger, IDisposable
{
    private readonly SqliteConnection _connection;
    private readonly ILogger<SiteEventLogger> _logger;
    private readonly object _writeLock = new();
    private readonly Channel<PendingEvent> _writeQueue;
    private readonly Task _writerLoop;
    private long _failedWriteCount;
    private bool _disposed;

    /// <summary>
    /// Initializes the event logger over the consolidated site database and starts the
    /// background writer loop.
    /// </summary>
    /// <param name="options">Site event log configuration (retention settings, queue capacity).</param>
    /// <param name="logger">Logger for write-failure diagnostics.</param>
    /// <param name="localDb">
    /// The consolidated site database. The connection it hands out is already open and
    /// carries the <c>zb_hlc_next()</c> UDF that <c>site_events</c>' capture triggers
    /// call. This logger owns the connection for its lifetime and disposes it.
    /// </param>
    /// <remarks>
    /// The former <c>connectionStringOverride</c> seam is deliberately gone rather than
    /// preserved. <c>site_events</c> is a replicated table, so a raw connection lacking
    /// the UDF would fail closed on every insert — an override would only let a caller
    /// build a logger that cannot write. <c>SiteEventLogOptions.DatabasePath</c> is
    /// likewise no longer read: the location is <c>LocalDb:Path</c>.
    /// </remarks>
    public SiteEventLogger(
        IOptions<SiteEventLogOptions> options,
        ILogger<SiteEventLogger> logger,
        ILocalDb localDb)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentNullException.ThrowIfNull(localDb);

        _logger = logger;

        // Already open — CreateConnection returns a live, pragma-configured,
        // UDF-registered connection. Calling Open() on it again would throw.
        _connection = localDb.CreateConnection();

        InitializeSchema();

        // Bounded queue with DropOldest preserves the
        // "callers never block" guarantee while putting an
        // upper bound on memory under sustained writer slowness. Drops are
        // observable — itemDropped faults the evicted Task and increments
        // FailedWriteCount.
        var capacity = Math.Max(1, options.Value.WriteQueueCapacity);
        _writeQueue = Channel.CreateBounded<PendingEvent>(
            new BoundedChannelOptions(capacity)
            {
                SingleReader = true,
                SingleWriter = false,
                FullMode = BoundedChannelFullMode.DropOldest,
            },
            itemDropped: dropped =>
            {
                Interlocked.Increment(ref _failedWriteCount);
                dropped.Completion.TrySetException(
                    new InvalidOperationException(
                        $"Event was dropped because the write queue exceeded its bounded capacity ({capacity})."));
            });
        _writerLoop = Task.Run(ProcessWriteQueueAsync);
    }

    /// <inheritdoc />
    public long FailedWriteCount => Interlocked.Read(ref _failedWriteCount);

    /// <summary>
    /// Runs <paramref name="action"/> against the shared connection while holding the
    /// write lock, so purge / query / record callers on different threads never use
    /// the non-thread-safe <see cref="SqliteConnection"/> concurrently.
    /// Returns <see langword="false"/> without invoking the action if the logger has
    /// been disposed.
    /// </summary>
    /// <param name="action">The action to run against the shared connection.</param>
    /// <returns><see langword="true"/> if the action was executed; <see langword="false"/> if the logger is disposed.</returns>
    internal bool WithConnection(Action<SqliteConnection> action)
    {
        ArgumentNullException.ThrowIfNull(action);
        lock (_writeLock)
        {
            if (_disposed) return false;
            action(_connection);
            return true;
        }
    }

    /// <summary>
    /// Runs <paramref name="func"/> against the shared connection while holding the
    /// write lock. Throws <see cref="ObjectDisposedException"/> if the logger has
    /// been disposed (callers that need a result cannot proceed without the database).
    /// </summary>
    /// <typeparam name="T">The return type of the function.</typeparam>
    /// <param name="func">The function to run against the shared connection.</param>
    /// <returns>The value returned by <paramref name="func"/>.</returns>
    internal T WithConnection<T>(Func<SqliteConnection, T> func)
    {
        ArgumentNullException.ThrowIfNull(func);
        lock (_writeLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);
            return func(_connection);
        }
    }

    // Schema lives in SiteEventLogSchema so the Host's AddZbLocalDb onReady callback
    // can create this table in the consolidated site database before RegisterReplicated
    // installs its capture triggers. Applying it here too keeps a directly-constructed
    // logger (tests, tooling) self-sufficient; the DDL is idempotent.
    private void InitializeSchema() => SiteEventLogSchema.Apply(_connection);

    /// <summary>
    /// Closed set of allowed severities. Case-sensitive to
    /// match the SQLite default <c>BINARY</c> collation used by the query filter —
    /// a row stored as <c>"error"</c> would be invisible to a query filtering on
    /// <c>"Error"</c>, so the contract on the way in must match the contract on
    /// the way out.
    /// </summary>
    private static readonly HashSet<string> AllowedSeverities =
        new(StringComparer.Ordinal) { "Info", "Warning", "Error" };

    /// <inheritdoc />
    public Task LogEventAsync(
        string eventType,
        string severity,
        string? instanceId,
        string source,
        string message,
        string? details = null)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(eventType);
        ArgumentException.ThrowIfNullOrWhiteSpace(severity);
        ArgumentException.ThrowIfNullOrWhiteSpace(source);
        ArgumentException.ThrowIfNullOrWhiteSpace(message);

        // Reject unknown severities so the query-time filter
        // (case-sensitive BINARY collation) and the documented enum stay in sync.
        if (!AllowedSeverities.Contains(severity))
        {
            throw new ArgumentException(
                $"Severity '{severity}' is not one of the allowed values: Info, Warning, Error.",
                nameof(severity));
        }

        // The id is minted here, by the application, rather than by SQLite. Under
        // LocalDb replication the site pair converges by last-writer-wins on the
        // primary key, so a server-minted AUTOINCREMENT would have both nodes
        // independently issuing id 1, 2, 3... for unrelated events and silently
        // overwriting each other on sync. A GUID makes the event log a pure union.
        var pending = new PendingEvent(
            Guid.NewGuid().ToString("N"),
            DateTimeOffset.UtcNow.ToString("o"),
            eventType,
            severity,
            instanceId,
            source,
            message,
            details);

        // Enqueue only — the actual SQLite write happens on the background writer
        // thread so the caller (an Akka actor thread on a hot path) never blocks
        // on disk I/O or on contention for the write lock.
        if (!_writeQueue.Writer.TryWrite(pending))
        {
            // The channel is bounded with DropOldest, so TryWrite only returns false
            // once the channel is completed (logger disposed) — full-buffer pressure
            // evicts the oldest pending event instead of failing the write. The event
            // cannot be persisted — fault the Task rather than reporting false success,
            // so a caller that awaits a critical audit event can tell it was dropped.
            pending.Completion.TrySetException(
                new ObjectDisposedException(nameof(SiteEventLogger),
                    "Event could not be recorded: the event logger has been disposed."));
        }

        return pending.Completion.Task;
    }

    private async Task ProcessWriteQueueAsync()
    {
        await foreach (var pending in _writeQueue.Reader.ReadAllAsync().ConfigureAwait(false))
        {
            try
            {
                var written = WithConnection(connection =>
                {
                    using var cmd = connection.CreateCommand();
                    cmd.CommandText = """
                        INSERT INTO site_events (id, timestamp, event_type, severity, instance_id, source, message, details)
                        VALUES ($id, $timestamp, $event_type, $severity, $instance_id, $source, $message, $details)
                        """;
                    cmd.Parameters.AddWithValue("$id", pending.Id);
                    cmd.Parameters.AddWithValue("$timestamp", pending.Timestamp);
                    cmd.Parameters.AddWithValue("$event_type", pending.EventType);
                    cmd.Parameters.AddWithValue("$severity", pending.Severity);
                    cmd.Parameters.AddWithValue("$instance_id", (object?)pending.InstanceId ?? DBNull.Value);
                    cmd.Parameters.AddWithValue("$source", pending.Source);
                    cmd.Parameters.AddWithValue("$message", pending.Message);
                    cmd.Parameters.AddWithValue("$details", (object?)pending.Details ?? DBNull.Value);
                    cmd.ExecuteNonQuery();
                });

                if (written)
                {
                    pending.Completion.TrySetResult();
                }
                else
                {
                    // WithConnection returns false only when the logger has been
                    // disposed mid-drain; the event was not persisted. Fault the
                    // Task instead of reporting false
                    // success for a dropped audit event.
                    pending.Completion.TrySetException(
                        new ObjectDisposedException(nameof(SiteEventLogger),
                            "Event could not be recorded: the event logger was disposed before the write completed."));
                }
            }
            catch (Exception ex)
            {
                // A write failure must be observable. Count it
                // (Health Monitoring reads FailedWriteCount) and fault the caller's
                // Task instead of silently discarding the exception.
                Interlocked.Increment(ref _failedWriteCount);
                _logger.LogError(ex, "Failed to record event: {EventType} from {Source} (sqlite {SqliteError})",
                    pending.EventType, pending.Source, DescribeSqliteError(ex));
                pending.Completion.TrySetException(ex);
            }
        }
    }

    /// <summary>
    /// Stops accepting new events, drains the write queue, and disposes the SQLite connection.
    /// </summary>
    public void Dispose()
    {
        Task? writerLoop = null;
        lock (_writeLock)
        {
            if (_disposed) return;
            _disposed = true;
            // Stop accepting new events and let the writer loop drain.
            _writeQueue.Writer.TryComplete();
            writerLoop = _writerLoop;
        }

        // Wait for the writer loop to finish outside the lock — the loop itself
        // acquires the lock for each write.
        try
        {
            writerLoop?.Wait(TimeSpan.FromSeconds(5));
        }
        catch (AggregateException)
        {
            // A faulted writer loop has already been logged per event; nothing more
            // to do during disposal.
        }

        lock (_writeLock)
        {
            _connection.Dispose();
        }
    }

    /// <summary>
    /// "primary/extended" SQLite result codes (e.g. "10/522") for a
    /// <see cref="SqliteException"/> anywhere in the chain; "n/a" otherwise.
    /// The exception message alone carries only the primary code, which proved
    /// too generic to diagnose the 2026-07-20 disk-I/O incident
    /// (known-issues/2026-07-20-localdb-disk-io-error-under-load.md).
    /// </summary>
    private static string DescribeSqliteError(Exception ex)
    {
        for (Exception? e = ex; e is not null; e = e.InnerException)
        {
            if (e is SqliteException se)
            {
                return $"{se.SqliteErrorCode}/{se.SqliteExtendedErrorCode}";
            }
        }
        return "n/a";
    }

    /// <summary>An event awaiting persistence by the background writer.</summary>
    private sealed record PendingEvent(
        string Id,
        string Timestamp,
        string EventType,
        string Severity,
        string? InstanceId,
        string Source,
        string Message,
        string? Details)
    {
        /// <summary>Completes when the event has been durably persisted, or faults on write failure.</summary>
        public TaskCompletionSource Completion { get; } =
            new(TaskCreationOptions.RunContinuationsAsynchronously);
    }
}