solo-storage 0.8.1

// SPDX-License-Identifier: Apache-2.0

//! Audit log primitives (v0.8.0 P4).
//!
//! ## Two emission paths
//!
//! **Synchronous (mutating writer-actor handlers):** an audit row is
//! inserted as part of the same SQLite transaction that performs the
//! write. If the audit insert fails, the entire write rolls back —
//! strict ACID. Implemented via [`insert_audit_row_in_tx`].
//!
//! **Asynchronous (query path):** the query path runs on a `ReaderPool`
//! that doesn't own a writeable connection, and we don't want to pay an
//! extra round-trip on the hot read path. Instead, queries emit through
//! an [`AuditWriter`] backed by a bounded `tokio::sync::mpsc` channel.
//! A background task drains the channel, batches up to
//! [`AUDIT_BATCH_FLUSH_MAX_EVENTS`] events (or up to
//! [`AUDIT_BATCH_FLUSH_MAX_MILLIS`] milliseconds, whichever fires first),
//! and COMMITs each batch in one transaction.
//!
//! Backpressure: the mpsc capacity is [`AUDIT_QUEUE_CAPACITY`]. When the
//! channel is full, `AuditWriter::emit_async` drops the event with a
//! `tracing::warn!` line — the query path is never blocked. This is the
//! explicit trade-off documented in 0090 §"Concurrency contract": the
//! query path's latency budget dominates compliance completeness for the
//! "queries log my reads" case. Mutating writes have no such trade-off —
//! they go through the synchronous path with full ACID guarantees.

use rusqlite::{params, Connection, Transaction, TransactionBehavior};
use solo_core::{Error, Result};
use std::path::PathBuf;
use std::time::Duration;
use tokio::sync::mpsc;

use crate::init::open_sqlcipher;
use crate::key_material::KeyMaterial;

/// Bounded mpsc capacity for the async audit pipeline. 1024 lets a burst
/// of 1000 concurrent recalls land without dropping; sustained load past
/// this drops events with a `tracing::warn!` line. Tuned with the
/// `recall(1000)` stress test in mind.
pub const AUDIT_QUEUE_CAPACITY: usize = 1024;

/// Max events per flushed batch. Trading off larger batches (fewer
/// transactions, lower overhead) vs. tighter recency for the audit log.
/// 64 chosen so a sustained recall storm flushes ~16 batches/sec at
/// burst — well within SQLite's commit-rate envelope.
pub const AUDIT_BATCH_FLUSH_MAX_EVENTS: usize = 64;

/// Max millisecond delay before forcing a flush even with a partial
/// batch. Sets the bound on "how recent is the audit log". 50ms means
/// the audit log lags real time by at most 50ms in the absence of a
/// crash; on crash, any events still in the mpsc/batch are lost (this
/// is the explicit async trade-off — mutating writes use the
/// synchronous transactional path).
pub const AUDIT_BATCH_FLUSH_MAX_MILLIS: u64 = 50;

/// All audit operations Solo records. 1:1 with the 13 MCP tools + admin
/// tenant ops + redaction + GDPR. Display string matches the column
/// value persisted in `audit_events.operation`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuditOperation {
    // Per-tenant writes (synchronous emit inside writer-actor tx)
    MemoryRemember,
    MemoryForget,
    MemoryConsolidate,
    MemoryReembed,
    MemoryIngestDocument,
    MemoryForgetDocument,
    MemoryNormalizeSubjects,
    MemoryBackup,
    MemorySaveSnapshot,

    // Per-tenant queries (async emit via AuditWriter)
    MemoryRecall,
    MemoryInspect,
    MemoryThemes,
    MemoryFactsAbout,
    MemoryContradictions,
    MemoryInspectCluster,
    MemorySearchDocs,
    MemoryInspectDocument,
    MemoryListDocuments,

    // Per-tenant redaction (v0.8.0 P5; synchronous emit inside the same
    // writer-actor tx as the ingest/remember it summarises).
    RedactionApplied,

    // Admin (write to audit_events_admin in tenants_index.db; wired in P7)
    TenantCreate,
    TenantDelete,
    TenantBackup,
    TenantRestore,

    /// v0.8.1 P3: admin-tier op recording a quota change (set / change /
    /// clear). Written to `audit_events_admin` because it's a registry-
    /// scope change, not per-tenant data.
    TenantSetQuota,

    // GDPR (v0.8.0 P6; written to audit_events_admin because the subject
    // can no longer query their own per-tenant DB audit_events post-
    // deletion).
    GdprForgetUser,
}

impl AuditOperation {
    /// Canonical string form persisted in the `operation` column.
    pub const fn as_str(&self) -> &'static str {
        match self {
            Self::MemoryRemember => "memory.remember",
            Self::MemoryForget => "memory.forget",
            Self::MemoryConsolidate => "memory.consolidate",
            Self::MemoryReembed => "memory.reembed",
            Self::MemoryIngestDocument => "memory.ingest_document",
            Self::MemoryForgetDocument => "memory.forget_document",
            Self::MemoryNormalizeSubjects => "memory.normalize_subjects",
            Self::MemoryBackup => "memory.backup",
            Self::MemorySaveSnapshot => "memory.save_snapshot",
            Self::MemoryRecall => "memory.recall",
            Self::MemoryInspect => "memory.inspect",
            Self::MemoryThemes => "memory.themes",
            Self::MemoryFactsAbout => "memory.facts_about",
            Self::MemoryContradictions => "memory.contradictions",
            Self::MemoryInspectCluster => "memory.inspect_cluster",
            Self::MemorySearchDocs => "memory.search_docs",
            Self::MemoryInspectDocument => "memory.inspect_document",
            Self::MemoryListDocuments => "memory.list_documents",
            Self::RedactionApplied => "redaction.applied",
            Self::TenantCreate => "tenant.create",
            Self::TenantDelete => "tenant.delete",
            Self::TenantBackup => "tenant.backup",
            Self::TenantRestore => "tenant.restore",
            Self::TenantSetQuota => "tenant.set_quota",
            Self::GdprForgetUser => "gdpr.forget_user",
        }
    }
}

impl std::fmt::Display for AuditOperation {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

/// Persisted `result` column value. Mirrors the SQL CHECK domain.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuditResult {
    Ok,
    Error,
    Forbidden,
}

impl AuditResult {
    pub const fn as_str(&self) -> &'static str {
        match self {
            Self::Ok => "ok",
            Self::Error => "error",
            Self::Forbidden => "forbidden",
        }
    }
}

/// One audit event before persistence. Built by the caller; the audit
/// pipeline owns persistence + retention.
#[derive(Debug, Clone)]
pub struct AuditEvent {
    /// Epoch ms. Filled by callers via `chrono::Utc::now().timestamp_millis()`.
    pub ts_ms: i64,
    /// Authenticated principal's subject (JWT `sub` or `"bearer"`). `None`
    /// for CLI / no-auth / system-initiated paths.
    pub principal_subject: Option<String>,
    pub operation: AuditOperation,
    /// Subject of the operation: a memory_id, doc_id, cluster_id, query
    /// hash, etc. `None` when the operation isn't bound to a single id
    /// (e.g. `memory.themes`).
    pub target_id: Option<String>,
    pub result: AuditResult,
    /// Optional structured detail as a `serde_json::Value`. Serialized
    /// to TEXT for storage; `None` persists as NULL.
    pub details: Option<serde_json::Value>,
}

impl AuditEvent {
    /// Construct an `ok` event with current-time timestamp. Convenience
    /// for the synchronous mutating-handler emit path.
    pub fn ok_now(
        principal_subject: Option<String>,
        operation: AuditOperation,
        target_id: Option<String>,
    ) -> Self {
        Self {
            ts_ms: chrono::Utc::now().timestamp_millis(),
            principal_subject,
            operation,
            target_id,
            result: AuditResult::Ok,
            details: None,
        }
    }

    /// Construct an `error` event with current-time timestamp + a JSON
    /// details blob carrying the failure summary.
    pub fn error_now(
        principal_subject: Option<String>,
        operation: AuditOperation,
        target_id: Option<String>,
        error_message: impl Into<String>,
    ) -> Self {
        Self {
            ts_ms: chrono::Utc::now().timestamp_millis(),
            principal_subject,
            operation,
            target_id,
            result: AuditResult::Error,
            details: Some(serde_json::json!({ "error": error_message.into() })),
        }
    }
}

/// Persist one audit event inside an already-open SQLCipher transaction.
/// Synchronous emit path — used by the writer-actor for mutating ops, so
/// the audit row is atomic with the audited SQL write.
///
/// If this returns an error the caller MUST rollback the surrounding
/// transaction — strict ACID for the mutating write.
pub fn insert_audit_row_in_tx(tx: &Transaction<'_>, event: &AuditEvent) -> Result<()> {
    let details_json: Option<String> = match event.details.as_ref() {
        Some(v) => Some(serde_json::to_string(v).map_err(|e| {
            Error::storage(format!("serialize audit details: {e}"))
        })?),
        None => None,
    };
    tx.execute(
        "INSERT INTO audit_events (
            ts_ms, principal_subject, operation, target_id, result, details_json
         ) VALUES (?, ?, ?, ?, ?, ?)",
        params![
            event.ts_ms,
            event.principal_subject.as_deref(),
            event.operation.as_str(),
            event.target_id.as_deref(),
            event.result.as_str(),
            details_json,
        ],
    )
    .map_err(|e| Error::storage(format!("INSERT audit_events: {e}")))?;
    Ok(())
}

/// Persist one admin-tier audit event into the
/// `audit_events_admin` table in `tenants_index.db`. Schema differs
/// from per-tenant `audit_events`: the row is keyed by
/// `target_tenant_id` (the tenant the admin operation affects) rather
/// than `target_id` (which is per-tenant data scope).
///
/// Returns the last_insert_rowid for the inserted row — callers (GDPR
/// + backup/restore) bubble that to their reports for traceability.
///
/// Synchronous — wrap in a tx if the caller wants atomicity with
/// surrounding work. The standalone helper just executes the INSERT on
/// the supplied connection.
pub fn insert_audit_admin_row(
    conn: &Connection,
    ts_ms: i64,
    principal_subject: Option<&str>,
    operation: AuditOperation,
    target_tenant_id: Option<&str>,
    result: AuditResult,
    details: Option<&serde_json::Value>,
) -> Result<i64> {
    let details_json: Option<String> = match details {
        Some(v) => Some(serde_json::to_string(v).map_err(|e| {
            Error::storage(format!("serialize admin audit details: {e}"))
        })?),
        None => None,
    };
    conn.execute(
        "INSERT INTO audit_events_admin (
            ts_ms, principal_subject, operation, target_tenant_id, result, details_json
         ) VALUES (?, ?, ?, ?, ?, ?)",
        params![
            ts_ms,
            principal_subject,
            operation.as_str(),
            target_tenant_id,
            result.as_str(),
            details_json,
        ],
    )
    .map_err(|e| Error::storage(format!("INSERT audit_events_admin: {e}")))?;
    Ok(conn.last_insert_rowid())
}

/// Persist one audit event on a non-transactional connection. Used by
/// the async batch drainer (which opens its own SQLCipher connection
/// per tenant; the writer-actor's connection is unavailable from the
/// query side).
#[allow(dead_code)]
fn insert_audit_row_one_off(conn: &Connection, event: &AuditEvent) -> Result<()> {
    let details_json: Option<String> = match event.details.as_ref() {
        Some(v) => Some(serde_json::to_string(v).map_err(|e| {
            Error::storage(format!("serialize audit details: {e}"))
        })?),
        None => None,
    };
    conn.execute(
        "INSERT INTO audit_events (
            ts_ms, principal_subject, operation, target_id, result, details_json
         ) VALUES (?, ?, ?, ?, ?, ?)",
        params![
            event.ts_ms,
            event.principal_subject.as_deref(),
            event.operation.as_str(),
            event.target_id.as_deref(),
            event.result.as_str(),
            details_json,
        ],
    )
    .map_err(|e| Error::storage(format!("INSERT audit_events (async): {e}")))?;
    Ok(())
}

/// Cheaply-cloneable handle for async audit emit.
///
/// Cloned freely across query handlers (one clone per outstanding query
/// is fine — the channel sender is itself cheap). Dropping the LAST
/// clone closes the channel and lets the background drainer flush + exit
/// cleanly.
#[derive(Clone)]
pub struct AuditWriter {
    tx: mpsc::Sender<AuditEvent>,
}

/// Handle to the spawned audit drainer task. Held by `TenantHandle` so a
/// graceful shutdown can wait for the drainer to finish flushing the
/// pending batch.
pub struct AuditWriterShutdown {
    drainer: tokio::task::JoinHandle<()>,
}

impl AuditWriterShutdown {
    /// Await the drainer's exit. Call AFTER dropping every `AuditWriter`
    /// clone so the mpsc channel closes and the drainer drains-and-exits.
    pub async fn join(self) {
        if let Err(e) = self.drainer.await {
            tracing::warn!(error = %e, "audit drainer task join error");
        }
    }
}

impl AuditWriter {
    /// Spawn a new audit drainer for one tenant's DB. Returns a cheap-
    /// to-clone writer plus a shutdown handle the caller stores alongside.
    ///
    /// `db_path` + `key` are used by the drainer to open its own
    /// SQLCipher connection (separate from the writer-actor's so the
    /// async emit path doesn't contend on the writer's mutex). The
    /// connection is opened lazily on first event so cold tenants don't
    /// pay the cost.
    pub fn spawn(
        db_path: PathBuf,
        key: Option<KeyMaterial>,
    ) -> (Self, AuditWriterShutdown) {
        let (tx, rx) = mpsc::channel::<AuditEvent>(AUDIT_QUEUE_CAPACITY);
        let drainer = tokio::spawn(audit_drainer_loop(rx, db_path, key));
        (
            Self { tx },
            AuditWriterShutdown { drainer },
        )
    }

    /// Spawn a no-op writer that drops every event. Used by tests that
    /// don't care about audit emission, and by code paths that want to
    /// thread an `AuditWriter` argument without actually wiring storage
    /// (e.g. the legacy single-tenant test harnesses).
    pub fn noop() -> Self {
        // 1-capacity channel whose receiver is dropped immediately — every
        // send fails, which `emit_async` swallows.
        let (tx, _rx) = mpsc::channel::<AuditEvent>(1);
        // Drop the receiver in a background task to keep the channel open
        // for `try_send` to succeed in capacity-1 mode (the receiver is
        // immediately dropped here so try_send fails — the warn-on-drop
        // path covers this). Either way the event is intentionally lost.
        Self { tx }
    }

    /// Try to enqueue an event for async persistence. Non-blocking: if
    /// the channel is full, the event is dropped + a single
    /// `tracing::warn!` line records the loss. The query path is never
    /// blocked.
    ///
    /// Returns `true` if enqueued, `false` if dropped.
    pub fn emit_async(&self, event: AuditEvent) -> bool {
        match self.tx.try_send(event) {
            Ok(()) => true,
            Err(mpsc::error::TrySendError::Full(ev)) => {
                tracing::warn!(
                    operation = %ev.operation,
                    "audit: mpsc full, dropping event (queue capacity {AUDIT_QUEUE_CAPACITY})"
                );
                false
            }
            Err(mpsc::error::TrySendError::Closed(ev)) => {
                tracing::debug!(
                    operation = %ev.operation,
                    "audit: writer closed, dropping event (shutdown in progress?)"
                );
                false
            }
        }
    }

    /// Convenience: build + emit an `ok` event for `operation`.
    pub fn emit_ok(
        &self,
        principal_subject: Option<String>,
        operation: AuditOperation,
        target_id: Option<String>,
    ) {
        let _ = self.emit_async(AuditEvent::ok_now(
            principal_subject,
            operation,
            target_id,
        ));
    }

    /// Convenience: build + emit an `error` event for `operation` with
    /// the failure message in details.
    pub fn emit_error(
        &self,
        principal_subject: Option<String>,
        operation: AuditOperation,
        target_id: Option<String>,
        err: impl std::fmt::Display,
    ) {
        let _ = self.emit_async(AuditEvent::error_now(
            principal_subject,
            operation,
            target_id,
            err.to_string(),
        ));
    }
}

/// Background loop: read up to `AUDIT_BATCH_FLUSH_MAX_EVENTS` events or
/// wait `AUDIT_BATCH_FLUSH_MAX_MILLIS`, whichever first; flush the batch
/// to SQLite in one transaction; repeat.
///
/// Exits when the mpsc channel is closed (every `AuditWriter` clone has
/// been dropped). Flushes any partial batch before exit.
async fn audit_drainer_loop(
    mut rx: mpsc::Receiver<AuditEvent>,
    db_path: PathBuf,
    key: Option<KeyMaterial>,
) {
    // Lazy connection — opened on first event so cold tenants don't pay
    // the cost. `None` until we have something to write.
    let mut conn: Option<Connection> = None;
    let flush_interval = Duration::from_millis(AUDIT_BATCH_FLUSH_MAX_MILLIS);

    loop {
        // Block until we have at least one event (or shutdown).
        let first = match rx.recv().await {
            Some(e) => e,
            None => {
                // Channel closed; no more events will arrive. Done.
                tracing::debug!("audit drainer: mpsc closed, exiting");
                return;
            }
        };
        let mut batch: Vec<AuditEvent> = Vec::with_capacity(AUDIT_BATCH_FLUSH_MAX_EVENTS);
        batch.push(first);

        // Greedily pull more events without waiting (drain whatever else
        // is already buffered), up to the batch cap.
        while batch.len() < AUDIT_BATCH_FLUSH_MAX_EVENTS {
            match rx.try_recv() {
                Ok(e) => batch.push(e),
                Err(mpsc::error::TryRecvError::Empty) => break,
                Err(mpsc::error::TryRecvError::Disconnected) => break,
            }
        }

        // If we still have budget for more events AND the batch isn't
        // full, wait up to flush_interval for stragglers. Done with
        // `tokio::time::timeout` so the wait is bounded.
        if batch.len() < AUDIT_BATCH_FLUSH_MAX_EVENTS {
            let deadline = tokio::time::Instant::now() + flush_interval;
            while batch.len() < AUDIT_BATCH_FLUSH_MAX_EVENTS {
                let now = tokio::time::Instant::now();
                if now >= deadline {
                    break;
                }
                let remaining = deadline - now;
                match tokio::time::timeout(remaining, rx.recv()).await {
                    Ok(Some(e)) => batch.push(e),
                    Ok(None) => break, // channel closed
                    Err(_) => break,    // deadline reached
                }
            }
        }

        // Ensure the connection is open (lazy first-event open).
        if conn.is_none() {
            match open_drainer_conn(&db_path, key.as_ref()) {
                Ok(c) => conn = Some(c),
                Err(e) => {
                    tracing::error!(
                        error = %e,
                        path = %db_path.display(),
                        dropped = batch.len(),
                        "audit drainer: failed to open connection, dropping batch"
                    );
                    // Don't retry on this iteration — we'd loop forever.
                    // The next batch may succeed (e.g. transient I/O).
                    continue;
                }
            }
        }
        let c = conn.as_mut().expect("conn just set");

        if let Err(e) = flush_batch(c, &batch) {
            tracing::error!(
                error = %e,
                dropped = batch.len(),
                "audit drainer: flush failed, dropping batch"
            );
        }
    }
}

fn open_drainer_conn(
    db_path: &std::path::Path,
    key: Option<&KeyMaterial>,
) -> Result<Connection> {
    if let Some(k) = key {
        open_sqlcipher(db_path, k)
    } else {
        // Test/legacy path: open plain SQLite. Same shape as
        // `ReaderPool::new(..., None, ...)`.
        let conn = Connection::open(db_path).map_err(|e| {
            Error::storage(format!("audit drainer open {}: {e}", db_path.display()))
        })?;
        conn.execute_batch(
            "PRAGMA journal_mode = wal;
             PRAGMA busy_timeout = 5000;",
        )
        .map_err(|e| Error::storage(format!("audit drainer pragmas: {e}")))?;
        Ok(conn)
    }
}

fn flush_batch(conn: &mut Connection, batch: &[AuditEvent]) -> Result<()> {
    let tx = conn
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(|e| Error::storage(format!("audit drainer BEGIN IMMEDIATE: {e}")))?;
    for event in batch {
        insert_audit_row_in_tx(&tx, event)?;
    }
    tx.commit()
        .map_err(|e| Error::storage(format!("audit drainer COMMIT: {e}")))?;
    Ok(())
}

/// Delete every audit row whose `ts_ms` is older than `cutoff_ms`.
/// Returns the number of rows deleted.
///
/// Idempotent: re-running on the same cutoff is a no-op once the rows
/// are gone. Single-transaction so a crash mid-purge leaves the table
/// either fully purged or fully intact.
pub fn purge_older_than(conn: &mut Connection, cutoff_ms: i64) -> Result<usize> {
    let tx = conn
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(|e| Error::storage(format!("BEGIN IMMEDIATE for purge: {e}")))?;
    let rows = tx
        .execute("DELETE FROM audit_events WHERE ts_ms < ?", params![cutoff_ms])
        .map_err(|e| Error::storage(format!("DELETE audit_events: {e}")))?;
    tx.commit()
        .map_err(|e| Error::storage(format!("COMMIT purge: {e}")))?;
    Ok(rows)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn open_in_memory_with_audit() -> Connection {
        let mut conn = Connection::open_in_memory().expect("open in-memory db");
        crate::migration::run_migrations(&mut conn).expect("migrations");
        conn
    }

    #[test]
    fn audit_operation_display_matches_canonical_string() {
        assert_eq!(AuditOperation::MemoryRemember.to_string(), "memory.remember");
        assert_eq!(AuditOperation::MemoryRecall.to_string(), "memory.recall");
        assert_eq!(AuditOperation::TenantCreate.to_string(), "tenant.create");
    }

    #[test]
    fn audit_result_check_constraint_rejects_unknown_value() {
        let conn = open_in_memory_with_audit();
        // Direct INSERT bypasses our enum so we can test the SQL CHECK.
        let res = conn.execute(
            "INSERT INTO audit_events (ts_ms, operation, result) VALUES (?, ?, ?)",
            params![0i64, "memory.remember", "bogus"],
        );
        assert!(res.is_err(), "result='bogus' must violate CHECK");
    }

    #[test]
    fn insert_then_select_round_trip() {
        let mut conn = open_in_memory_with_audit();
        let tx = conn.transaction().unwrap();
        let event = AuditEvent::ok_now(
            Some("alice".into()),
            AuditOperation::MemoryRemember,
            Some("00000000-0000-0000-0000-000000000001".into()),
        );
        insert_audit_row_in_tx(&tx, &event).unwrap();
        tx.commit().unwrap();

        let (op, principal, target, result): (String, Option<String>, Option<String>, String) =
            conn.query_row(
                "SELECT operation, principal_subject, target_id, result FROM audit_events",
                [],
                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
            )
            .unwrap();
        assert_eq!(op, "memory.remember");
        assert_eq!(principal.as_deref(), Some("alice"));
        assert_eq!(target.as_deref(), Some("00000000-0000-0000-0000-000000000001"));
        assert_eq!(result, "ok");
    }

    #[test]
    fn purge_older_than_drops_old_rows() {
        let mut conn = open_in_memory_with_audit();
        // Three rows at ts=100/200/300; purge anything ts < 250.
        for ts in [100i64, 200, 300] {
            conn.execute(
                "INSERT INTO audit_events (ts_ms, operation, result) VALUES (?, ?, ?)",
                params![ts, "memory.remember", "ok"],
            )
            .unwrap();
        }
        let purged = purge_older_than(&mut conn, 250).unwrap();
        assert_eq!(purged, 2, "purge should drop ts=100 and ts=200");
        let remaining: i64 = conn
            .query_row("SELECT COUNT(*) FROM audit_events", [], |r| r.get(0))
            .unwrap();
        assert_eq!(remaining, 1);
    }

    #[test]
    fn purge_older_than_idempotent() {
        let mut conn = open_in_memory_with_audit();
        conn.execute(
            "INSERT INTO audit_events (ts_ms, operation, result) VALUES (100, 'memory.remember', 'ok')",
            [],
        )
        .unwrap();
        assert_eq!(purge_older_than(&mut conn, 200).unwrap(), 1);
        assert_eq!(purge_older_than(&mut conn, 200).unwrap(), 0);
    }

    #[test]
    fn noop_writer_drops_events_without_blocking() {
        let writer = AuditWriter::noop();
        // try_send on a closed-receiver channel fails immediately, but
        // emit_async swallows that — should never panic / block.
        for _ in 0..100 {
            writer.emit_ok(None, AuditOperation::MemoryRecall, None);
        }
    }

    #[test]
    fn migration_0005_applied_once_on_repeated_open() {
        // Build a fresh in-memory DB, run migrations TWICE; the audit
        // schema_migrations row count for version 5 should be exactly 1.
        let mut conn = Connection::open_in_memory().expect("in-memory");
        crate::migration::run_migrations(&mut conn).unwrap();
        crate::migration::run_migrations(&mut conn).unwrap();
        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM schema_migrations WHERE version = 5",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(count, 1, "migration 0005 must apply at most once");
    }

    #[test]
    fn audit_table_present_and_indices_exist() {
        let conn = open_in_memory_with_audit();
        let table: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='audit_events'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(table, 1);
        for idx in ["idx_audit_ts", "idx_audit_principal"] {
            let exists: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name=?",
                    params![idx],
                    |r| r.get(0),
                )
                .unwrap();
            assert_eq!(exists, 1, "missing index: {idx}");
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn batched_load_does_not_drop_under_burst() {
        // Spawn the drainer; emit 1000 events; verify all 1000 land.
        let tmp = tempfile::TempDir::new().unwrap();
        let db_path = tmp.path().join("burst.db");
        // Bootstrap the schema on a separate handle (the drainer opens
        // its own lazily, but the audit_events table must exist before
        // the first batch lands).
        let mut conn = rusqlite::Connection::open(&db_path).unwrap();
        conn.execute_batch(
            "PRAGMA journal_mode = wal;
             PRAGMA foreign_keys = ON;
             PRAGMA busy_timeout = 5000;",
        )
        .unwrap();
        crate::migration::run_migrations(&mut conn).unwrap();
        drop(conn);

        let (audit, shutdown) = AuditWriter::spawn(db_path.clone(), None);
        // Burst: capacity is 1024; 1000 should fit without dropping. We
        // emit synchronously from a single task — the drainer pulls in
        // parallel — so we exercise the queue's drain rate.
        for i in 0..1000 {
            audit.emit_ok(
                Some(format!("user-{i}")),
                AuditOperation::MemoryRecall,
                None,
            );
        }
        // Give the drainer time to flush.
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        drop(audit);
        shutdown.join().await;

        let conn = rusqlite::Connection::open(&db_path).unwrap();
        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM audit_events WHERE operation = 'memory.recall'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(count, 1000, "all 1000 audit rows must land");
    }

    #[test]
    fn ok_now_and_error_now_construct_expected_shapes() {
        let ok = AuditEvent::ok_now(
            Some("u".into()),
            AuditOperation::MemoryRecall,
            Some("tid".into()),
        );
        assert_eq!(ok.result, AuditResult::Ok);
        assert!(ok.details.is_none());

        let err = AuditEvent::error_now(
            None,
            AuditOperation::MemoryRecall,
            None,
            "boom",
        );
        assert_eq!(err.result, AuditResult::Error);
        let details = err.details.expect("error event carries details");
        assert_eq!(details["error"], "boom");
    }
}