udb 0.4.27

//! SQLite implementation of [`AdminAuditStore`].
//!
//! ## Dialect choices
//!
//! - **`audit_id`** is a TEXT UUID (generated Rust-side).
//! - **`request_json`** is TEXT (parsed via `serde_json` on read).
//! - **`created_at`** is RFC-3339 TEXT via `strftime`.
//! - **Atomic chain append** uses `BEGIN IMMEDIATE` to acquire the
//!   write lock. SQLite is single-writer; this gives the same
//!   serialization guarantee that `pg_advisory_xact_lock` gives on
//!   Postgres or `GET_LOCK` gives on MySQL.
//! - **Chain verification** streams rows in pages of 1,000 ordered
//!   by `rowid ASC` (the monotonic insertion key, immune to
//!   same-timestamp UUID tie-break reordering) so the audit log can
//!   be arbitrarily large without buffering it all into memory.

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use sqlx::{Row, sqlite::SqliteConnection};
use uuid::Uuid;

use super::dialect::{SqlDialect, build_eq_where, normalize_limit_offset};
use super::sqlite::SqliteCanonicalStore;
use super::system_store::{
    AdminAuditChainReport, AdminAuditInsert, AdminAuditListFilter, AdminAuditRow, AdminAuditStore,
    SystemStoreError, SystemStoreResult, compute_admin_audit_hash, verify_admin_audit_chain_step,
};

const TABLE: &str = "udb_admin_audit_log";

fn parse_iso(s: &str) -> DateTime<Utc> {
    DateTime::parse_from_rfc3339(s)
        .map(|dt| dt.with_timezone(&Utc))
        .unwrap_or_else(|_| Utc::now())
}

fn row_to_audit(row: sqlx::sqlite::SqliteRow) -> SystemStoreResult<AdminAuditRow> {
    let audit_id_str: String = row
        .try_get("audit_id")
        .map_err(|e| SystemStoreError::query("sqlite", "SELECT audit_id", e))?;
    let audit_id = Uuid::parse_str(&audit_id_str).map_err(|e| {
        SystemStoreError::InvalidInput(format!(
            "audit_id '{audit_id_str}' is not a valid UUID: {e}"
        ))
    })?;
    let request_json_text: String = row.try_get("request_json").unwrap_or_default();
    let request_json = if request_json_text.is_empty() {
        serde_json::Value::Null
    } else {
        serde_json::from_str(&request_json_text).map_err(|e| {
            SystemStoreError::InvalidInput(format!(
                "request_json is not valid JSON: {e} (raw: '{request_json_text}')"
            ))
        })?
    };
    Ok(AdminAuditRow {
        audit_id,
        actor: row.try_get("actor").unwrap_or_default(),
        operation: row.try_get("operation").unwrap_or_default(),
        target: row.try_get("target").unwrap_or_default(),
        request_json,
        result: row.try_get("result").unwrap_or_default(),
        tenant_id: row.try_get("tenant_id").unwrap_or_default(),
        project_id: row.try_get("project_id").unwrap_or_default(),
        correlation_id: row.try_get("correlation_id").unwrap_or_default(),
        previous_hash: row.try_get("previous_hash").unwrap_or_default(),
        current_hash: row.try_get("current_hash").unwrap_or_default(),
        signer_key_id: row.try_get("signer_key_id").unwrap_or_default(),
        external_anchor: row.try_get("external_anchor").unwrap_or_default(),
        created_at: row
            .try_get::<String, _>("created_at")
            .map(|s| parse_iso(&s))
            .unwrap_or_else(|_| Utc::now()),
    })
}

#[async_trait]
impl AdminAuditStore for SqliteCanonicalStore {
    fn backend_label(&self) -> &'static str {
        "sqlite"
    }

    async fn ensure_admin_audit_tables(&self) -> SystemStoreResult<()> {
        // B.7: DDL strings come from the shared `sql_schema` renderer (single
        // source of truth across SQL backends); the execute loop below is
        // unchanged.
        for sql in super::sql_schema::sqlite_admin_audit_ddl(TABLE) {
            sqlx::query(&sql)
                .execute(self.pool_ref())
                .await
                .map_err(|e| SystemStoreError::query("sqlite", sql.clone(), e))?;
        }
        Ok(())
    }

    async fn latest_admin_audit_hash(&self) -> SystemStoreResult<String> {
        // Order by the monotonic insertion key (rowid), not (created_at,
        // audit_id): same-timestamp rows have random-UUID tie-breaks that can
        // mis-order the chain head. rowid reflects true append order.
        let sql = format!(
            "SELECT current_hash FROM {TABLE} \
             WHERE current_hash <> '' \
             ORDER BY rowid DESC LIMIT 1"
        );
        let hash: Option<String> = sqlx::query_scalar(&sql)
            .fetch_optional(self.pool_ref())
            .await
            .map_err(|e| SystemStoreError::query("sqlite", sql.clone(), e))?;
        Ok(hash.unwrap_or_default())
    }

    async fn append_admin_audit(&self, entry: &AdminAuditInsert) -> SystemStoreResult<Uuid> {
        // The read-latest + insert must be atomic against concurrent appends or
        // the hash chain forks. sqlx's `begin()` issues `BEGIN DEFERRED`, which
        // only takes the write lock lazily — two appends can both read the same
        // latest hash before either writes. Pin one connection and start the tx
        // with an explicit `BEGIN IMMEDIATE` so the write lock is held for the
        // entire critical section, then COMMIT (or ROLLBACK on failure).
        let mut conn = self
            .pool_ref()
            .acquire()
            .await
            .map_err(|e| SystemStoreError::io("sqlite", e))?;
        sqlx::query("BEGIN IMMEDIATE")
            .execute(&mut *conn)
            .await
            .map_err(|e| SystemStoreError::query("sqlite", "BEGIN IMMEDIATE", e))?;
        let result = Self::append_admin_audit_locked(&mut conn, entry).await;
        match &result {
            Ok(_) => {
                sqlx::query("COMMIT")
                    .execute(&mut *conn)
                    .await
                    .map_err(|e| SystemStoreError::io("sqlite", e))?;
            }
            Err(_) => {
                let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
            }
        }
        result
    }

    async fn list_admin_audit(
        &self,
        filter: &AdminAuditListFilter,
    ) -> SystemStoreResult<Vec<AdminAuditRow>> {
        let w = build_eq_where(
            SqlDialect::SQLITE,
            &[
                ("operation", filter.operation.is_some()),
                ("actor", filter.actor.is_some()),
                ("tenant_id", filter.tenant_id.is_some()),
                ("project_id", filter.project_id.is_some()),
            ],
        );
        let where_sql = &w.where_sql;
        let limit_placeholder = &w.limit_placeholder;
        let offset_placeholder = &w.offset_placeholder;
        let (limit, offset) = normalize_limit_offset(filter.limit, filter.offset);
        let sql = format!(
            "SELECT audit_id, actor, operation, target, request_json, result,
                    tenant_id, project_id, correlation_id,
                    previous_hash, current_hash, signer_key_id, external_anchor,
                    created_at
             FROM {TABLE}
             {where_sql}
             ORDER BY created_at DESC
             LIMIT {limit_placeholder} OFFSET {offset_placeholder}"
        );
        let mut q = sqlx::query(&sql);
        if let Some(o) = &filter.operation {
            q = q.bind(o.clone());
        }
        if let Some(a) = &filter.actor {
            q = q.bind(a.clone());
        }
        if let Some(t) = &filter.tenant_id {
            q = q.bind(t.clone());
        }
        if let Some(p) = &filter.project_id {
            q = q.bind(p.clone());
        }
        q = q.bind(limit).bind(offset);
        let rows = q
            .fetch_all(self.pool_ref())
            .await
            .map_err(|e| SystemStoreError::query("sqlite", sql.clone(), e))?;
        let mut out = Vec::with_capacity(rows.len());
        for r in rows {
            let mut row = row_to_audit(r)?;
            if filter.redact_request_json {
                row.request_json = serde_json::json!({"redacted": true});
            }
            out.push(row);
        }
        Ok(out)
    }

    async fn verify_admin_audit_chain(
        &self,
        limit: Option<i64>,
    ) -> SystemStoreResult<AdminAuditChainReport> {
        let mut conn = self
            .pool_ref()
            .acquire()
            .await
            .map_err(|e| SystemStoreError::io("sqlite", e))?;
        sqlx::query("BEGIN IMMEDIATE")
            .execute(&mut *conn)
            .await
            .map_err(|e| SystemStoreError::query("sqlite", "BEGIN IMMEDIATE", e))?;
        let mut previous_hash = String::new();
        let mut checked: i64 = 0;
        let mut offset: i64 = 0;

        // Streaming-friendly: pull rows page by page in oldest-first
        // order, feed each into the shared verify helper, stop on
        // the first failure or when the limit is hit.
        let final_report = loop {
            let remaining = match limit {
                Some(n) if n > 0 => (n - checked).max(0),
                _ => i64::MAX,
            };
            if remaining == 0 {
                break AdminAuditChainReport::Passed {
                    checked_count: checked,
                    last_hash: previous_hash.clone(),
                };
            }
            let page = remaining.min(super::dialect::admin_audit_verify_page_size());
            let sql = format!(
                "SELECT audit_id, actor, operation, target, request_json, result,
                        tenant_id, project_id, correlation_id,
                        previous_hash, current_hash, signer_key_id, external_anchor,
                        created_at
                 FROM {TABLE}
                 ORDER BY rowid ASC
                 LIMIT ? OFFSET ?"
            );
            let rows = match sqlx::query(&sql)
                .bind(page)
                .bind(offset)
                .fetch_all(&mut *conn)
                .await
            {
                Ok(rows) => rows,
                Err(e) => {
                    let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
                    return Err(SystemStoreError::query("sqlite", sql.clone(), e));
                }
            };
            if rows.is_empty() {
                break AdminAuditChainReport::Passed {
                    checked_count: checked,
                    last_hash: previous_hash.clone(),
                };
            }
            let n_rows = rows.len() as i64;
            let mut tamper: Option<AdminAuditChainReport> = None;
            for r in rows {
                let row = match row_to_audit(r) {
                    Ok(row) => row,
                    Err(err) => {
                        let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
                        return Err(err);
                    }
                };
                match verify_admin_audit_chain_step(&row, &previous_hash, checked) {
                    Ok(next) => {
                        previous_hash = next;
                        checked += 1;
                    }
                    Err(report) => {
                        tamper = Some(report);
                        break;
                    }
                }
            }
            if let Some(report) = tamper {
                break report;
            }
            offset += n_rows;
            if n_rows < page {
                break AdminAuditChainReport::Passed {
                    checked_count: checked,
                    last_hash: previous_hash.clone(),
                };
            }
        };
        sqlx::query("COMMIT")
            .execute(&mut *conn)
            .await
            .map_err(|e| SystemStoreError::query("sqlite", "COMMIT", e))?;
        Ok(final_report)
    }
}

impl SqliteCanonicalStore {
    /// Read-latest + insert on a connection already inside a `BEGIN IMMEDIATE`
    /// transaction (caller commits/rolls back). Kept out of the trait impl so
    /// the `#[async_trait]` macro doesn't process it as a trait method.
    async fn append_admin_audit_locked(
        conn: &mut SqliteConnection,
        entry: &AdminAuditInsert,
    ) -> SystemStoreResult<Uuid> {
        // Order by rowid (monotonic insertion key), not (created_at, audit_id):
        // same-timestamp rows have random-UUID tie-breaks that can fork the chain.
        let latest_sql = format!(
            "SELECT current_hash FROM {TABLE} \
             WHERE current_hash <> '' \
             ORDER BY rowid DESC LIMIT 1"
        );
        let previous_hash: String = sqlx::query_scalar(&latest_sql)
            .fetch_optional(&mut *conn)
            .await
            .map_err(|e| SystemStoreError::query("sqlite", latest_sql.clone(), e))?
            .unwrap_or_default();
        let current_hash = compute_admin_audit_hash(
            &previous_hash,
            &entry.actor,
            &entry.operation,
            &entry.target,
            &entry.request_json,
            &entry.result,
            &entry.tenant_id,
            &entry.project_id,
            &entry.correlation_id,
            &entry.signer_key_id,
            &entry.external_anchor,
        );
        let audit_id = Uuid::new_v4();
        let request_json_text = serde_json::to_string(&entry.request_json)
            .map_err(|e| SystemStoreError::InvalidInput(format!("request_json: {e}")))?;
        let insert_sql = format!(
            "INSERT INTO {TABLE} (
                audit_id, actor, operation, target, request_json, result,
                tenant_id, project_id, correlation_id,
                previous_hash, current_hash, signer_key_id, external_anchor
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        );
        sqlx::query(&insert_sql)
            .bind(audit_id.to_string())
            .bind(&entry.actor)
            .bind(&entry.operation)
            .bind(&entry.target)
            .bind(&request_json_text)
            .bind(&entry.result)
            .bind(&entry.tenant_id)
            .bind(&entry.project_id)
            .bind(&entry.correlation_id)
            .bind(&previous_hash)
            .bind(&current_hash)
            .bind(&entry.signer_key_id)
            .bind(&entry.external_anchor)
            .execute(&mut *conn)
            .await
            .map_err(|e| SystemStoreError::query("sqlite", insert_sql.clone(), e))?;
        Ok(audit_id)
    }
}

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

    async fn fresh_store() -> SqliteCanonicalStore {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("in-memory sqlite");
        let store = SqliteCanonicalStore::new(pool, "test", "udb_outbox_events");
        AdminAuditStore::ensure_admin_audit_tables(&store)
            .await
            .expect("DDL");
        store
    }

    fn sample_insert(operation: &str, actor: &str) -> AdminAuditInsert {
        AdminAuditInsert {
            actor: actor.to_string(),
            operation: operation.to_string(),
            target: "project-alpha".to_string(),
            request_json: serde_json::json!({"target": "catalog", "ver": 1}),
            result: "ok".to_string(),
            tenant_id: "tenant-1".to_string(),
            project_id: "project-alpha".to_string(),
            correlation_id: "corr-1".to_string(),
            signer_key_id: "default".to_string(),
            external_anchor: String::new(),
        }
    }

    /// Pin: empty table → latest hash is empty string. The chain
    /// starts here.
    #[tokio::test]
    async fn latest_hash_on_empty_table_is_empty_string() {
        let store = fresh_store().await;
        let h = store.latest_admin_audit_hash().await.expect("latest");
        assert_eq!(h, "");
    }

    /// Pin: append + verify end-to-end with one row, including the
    /// shared canonical hash.
    #[tokio::test]
    async fn append_one_row_then_verify_chain_passes() {
        let store = fresh_store().await;
        let id = store
            .append_admin_audit(&sample_insert("ActivateCatalog", "op-1"))
            .await
            .expect("append");
        assert_ne!(id, Uuid::nil());
        let latest = store.latest_admin_audit_hash().await.expect("latest");
        assert_eq!(latest.len(), 64, "latest is the SHA-256 hex of the row");
        let report = store.verify_admin_audit_chain(None).await.expect("verify");
        match report {
            AdminAuditChainReport::Passed {
                checked_count,
                last_hash,
            } => {
                assert_eq!(checked_count, 1);
                assert_eq!(last_hash, latest);
            }
            other => panic!("expected Passed, got: {other:?}"),
        }
    }

    /// Pin: three rows link together and verify end-to-end. Each
    /// row's `previous_hash` matches the prior row's `current_hash`.
    /// Brief sleeps between inserts so the strftime-millisecond
    /// `created_at` ordering is unambiguous; in production, admin
    /// audit events arrive at human pace (seconds apart).
    #[tokio::test]
    async fn three_rows_chain_links_and_verifies() {
        let store = fresh_store().await;
        let _id1 = store
            .append_admin_audit(&sample_insert("Op1", "op-1"))
            .await
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(2)).await;
        let _id2 = store
            .append_admin_audit(&sample_insert("Op2", "op-2"))
            .await
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(2)).await;
        let _id3 = store
            .append_admin_audit(&sample_insert("Op3", "op-3"))
            .await
            .unwrap();
        let report = store.verify_admin_audit_chain(None).await.unwrap();
        match report {
            AdminAuditChainReport::Passed { checked_count, .. } => {
                assert_eq!(checked_count, 3);
            }
            other => panic!("expected Passed, got: {other:?}"),
        }
    }

    /// Pin: tampering with a row's `operation` (which feeds the
    /// canonical hash) is detected by verify.
    #[tokio::test]
    async fn tampering_with_row_breaks_chain() {
        let store = fresh_store().await;
        store
            .append_admin_audit(&sample_insert("LegitOp", "op-1"))
            .await
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(2)).await;
        store
            .append_admin_audit(&sample_insert("LegitOp2", "op-2"))
            .await
            .unwrap();
        // Mutate the operation of the first row directly in the DB,
        // simulating tampering.
        sqlx::query(&format!(
            "UPDATE {TABLE} SET operation = 'TamperedOp' \
             WHERE audit_id = (SELECT audit_id FROM {TABLE} ORDER BY created_at ASC, audit_id ASC LIMIT 1)"
        ))
        .execute(store.pool_ref())
        .await
        .unwrap();

        let report = store.verify_admin_audit_chain(None).await.unwrap();
        match report {
            AdminAuditChainReport::Failed {
                reason,
                checked_count,
                ..
            } => {
                assert_eq!(
                    reason,
                    super::super::system_store::AdminAuditBreakReason::CurrentHashMismatch
                );
                assert_eq!(
                    checked_count, 0,
                    "first row is the tampered one; checked_count=0 at failure"
                );
            }
            other => panic!("expected Failed, got: {other:?}"),
        }
    }

    /// Pin: chain link tampering — fudging `previous_hash` on row N
    /// is caught as PreviousHashMismatch.
    #[tokio::test]
    async fn tampering_with_chain_link_is_detected() {
        let store = fresh_store().await;
        store
            .append_admin_audit(&sample_insert("Op1", "op-1"))
            .await
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(2)).await;
        store
            .append_admin_audit(&sample_insert("Op2", "op-2"))
            .await
            .unwrap();
        // Tamper with row 2's previous_hash.
        sqlx::query(&format!(
            "UPDATE {TABLE} SET previous_hash = 'forged' \
             WHERE audit_id = (SELECT audit_id FROM {TABLE} ORDER BY created_at ASC, audit_id ASC LIMIT 1 OFFSET 1)"
        ))
        .execute(store.pool_ref())
        .await
        .unwrap();
        let report = store.verify_admin_audit_chain(None).await.unwrap();
        match report {
            AdminAuditChainReport::Failed {
                reason,
                checked_count,
                ..
            } => {
                assert_eq!(
                    reason,
                    super::super::system_store::AdminAuditBreakReason::PreviousHashMismatch
                );
                assert_eq!(checked_count, 1, "row 1 passed; row 2 is broken");
            }
            other => panic!("expected Failed, got: {other:?}"),
        }
    }

    /// Pin: list_admin_audit honours every filter axis + redacts
    /// when requested.
    #[tokio::test]
    async fn list_admin_audit_filters_and_redacts() {
        let store = fresh_store().await;
        store
            .append_admin_audit(&sample_insert("Op1", "alice"))
            .await
            .unwrap();
        store
            .append_admin_audit(&sample_insert("Op2", "bob"))
            .await
            .unwrap();
        let mut sample = sample_insert("Op3", "carol");
        sample.tenant_id = "other-tenant".to_string();
        store.append_admin_audit(&sample).await.unwrap();

        // Filter by operation.
        let only_op2 = store
            .list_admin_audit(&AdminAuditListFilter {
                operation: Some("Op2".to_string()),
                limit: 100,
                ..AdminAuditListFilter::default()
            })
            .await
            .unwrap();
        assert_eq!(only_op2.len(), 1);
        assert_eq!(only_op2[0].actor, "bob");

        // Filter by actor.
        let only_carol = store
            .list_admin_audit(&AdminAuditListFilter {
                actor: Some("carol".to_string()),
                limit: 100,
                ..AdminAuditListFilter::default()
            })
            .await
            .unwrap();
        assert_eq!(only_carol.len(), 1);

        // Filter by tenant.
        let only_other = store
            .list_admin_audit(&AdminAuditListFilter {
                tenant_id: Some("other-tenant".to_string()),
                limit: 100,
                ..AdminAuditListFilter::default()
            })
            .await
            .unwrap();
        assert_eq!(only_other.len(), 1);
        assert_eq!(only_other[0].actor, "carol");

        // Redact.
        let redacted = store
            .list_admin_audit(&AdminAuditListFilter {
                limit: 100,
                redact_request_json: true,
                ..AdminAuditListFilter::default()
            })
            .await
            .unwrap();
        for row in redacted {
            assert_eq!(row.request_json, serde_json::json!({"redacted": true}));
        }
    }

    /// Pin: limit (`Some(n)`) on verify stops after `n` rows. Useful
    /// for incremental verification during long-running audits.
    #[tokio::test]
    async fn verify_with_limit_stops_early() {
        let store = fresh_store().await;
        for i in 0..5 {
            store
                .append_admin_audit(&sample_insert(&format!("Op{i}"), "op"))
                .await
                .unwrap();
            tokio::time::sleep(std::time::Duration::from_millis(2)).await;
        }
        let report = store.verify_admin_audit_chain(Some(3)).await.unwrap();
        match report {
            AdminAuditChainReport::Passed { checked_count, .. } => {
                assert_eq!(checked_count, 3);
            }
            other => panic!("expected Passed, got: {other:?}"),
        }
    }
}