sysknife_daemon/

store.rs

//! Storage backend abstraction for the audit log.
//!
//! Two backends ship in the box:
//!
//! - [`SqliteStore`] — `rusqlite`-backed, single-file local database. The
//!   default for dev/test; explicitly **not recommended** for production
//!   because the audit log dies with the host (no off-box durability,
//!   nothing to forward to a SOC).
//! - [`PostgresStore`] — `sqlx`-backed, native async, TLS via rustls +
//!   webpki-roots. Wire-compatible with AWS RDS / Aurora, GCP Cloud SQL +
//!   AlloyDB, Azure Database for PostgreSQL Flexible Server, Supabase,
//!   Neon, and self-hosted PostgreSQL.
//!
//! The dispatcher does not know which backend is active. It calls into
//! [`AuditStore`] (an async trait) and the runtime resolves to the configured
//! impl through `state.audit: Arc<dyn AuditStore + Send + Sync>`.
//!
//! ## Cloud provider quick reference
//!
//! See `docs/storage-cloud.md` for the full table. Summary:
//!
//! These three rows said `require` until they were checked against the code.
//! `pg_tls::require_tls_for_remote` REFUSES `sslmode=require` on any connection
//! that crosses the network, because `require` completes TLS against any
//! certificate and so authenticates nothing. An operator following the old table
//! got a hard startup refusal, and the obvious way to "fix" that is to loosen the
//! check — which reopens the MITM hole the check exists to close. Supabase and
//! Neon both publish CA chains, so `verify-full` is what they support and what
//! this daemon accepts.
//!
//! | Provider             | URL hint                                        | TLS mode      | Statement cache |
//! |----------------------|-------------------------------------------------|---------------|-----------------|
//! | AWS RDS / Aurora     | `*.rds.amazonaws.com:5432`                      | `verify-full` | default         |
//! | GCP Cloud SQL        | via Cloud SQL Auth Proxy on `127.0.0.1`         | `disable`     | default         |
//! | Azure Flexible       | `*.postgres.database.azure.com:5432`            | `verify-full` | default         |
//! | Supabase (5432)      | `db.<ref>.supabase.co:5432`                     | `verify-full` | default         |
//! | Supabase (pooler)    | `*.pooler.supabase.com:6543`                    | `verify-full` | **0**           |
//! | Neon                 | `ep-*.neon.tech`                                | `verify-full` | default         |
//! | Self-hosted          | (operator chooses)                              | as needed     | default         |

use async_trait::async_trait;
use std::sync::Arc;
use sysknife_types::{JobState, PreviewEnvelope, TransactionRecord};

use crate::audit_chain::{AuditKey, ChainRow, EventRow, VerifyOutcome};
use crate::auth::CallerPrincipal;
use crate::transactions::{
    NewTransaction, RecordedPreviewedTransaction, TransactionStore, TransactionStoreError,
};

pub mod postgres;

/// Column list every `ChainRow` read shares, on both backends. The SQLite
/// path (`transactions.rs`) maps rows positionally and the Postgres path
/// (`store/postgres.rs`) maps them by name, but both SELECT this exact list
/// in this exact order, so the list, the column count, and the mapper order
/// are one invariant.
///
/// This used to be declared twice — once per backend — kept in sync only by
/// a "Mirrors" doc comment. Adding the caller-identity columns updated the
/// mapper and one of the two queries, and the miss showed up only as a
/// runtime "no column found for name: chain_version" from the live-Postgres
/// test — the unit tests, which never touch this SQL, stayed green. One
/// declaration means that class of drift is a compile-time impossibility
/// rather than a live-database surprise (#397).
///
/// Parameter placeholders stay per-backend on purpose: SQLite binds `?1`,
/// Postgres binds `$1`. Only the column list is shared.
pub(crate) const CHAIN_ROW_COLUMNS: &str =
    "seq, key_id, transaction_id, request_id, request_hash, \
     action_name, risk_level, summary, approval_id, warnings_json, \
     created_at, prev_chain_hash, chain_hash, chain_version, caller_role, event_tip, \
     caller_principal";

/// The two facts the status cross-check reads from `audit_events` for each
/// transaction: the kind of its newest `status_*` event, and whether an
/// `approval_consumed` event exists for it. Correlated subqueries over `t`, the
/// `transactions` alias, and valid in both dialects.
///
/// Declared once for the reason [`CHAIN_ROW_COLUMNS`] is: the Postgres status
/// check is a port of the SQLite one, and two copies of this SQL could drift
/// into two different verdicts over the same evidence.
pub(crate) const NEWEST_STATUS_EVENT_SQL: &str = "(SELECT e.kind FROM audit_events e \
     WHERE e.transaction_id = t.transaction_id AND e.kind LIKE 'status_%' \
     ORDER BY e.seq DESC LIMIT 1)";
/// See [`NEWEST_STATUS_EVENT_SQL`].
pub(crate) const APPROVAL_CONSUMED_SQL: &str = "EXISTS (SELECT 1 FROM audit_events e \
     WHERE e.transaction_id = t.transaction_id AND e.kind = 'approval_consumed')";

/// Async, polymorphic interface to the audit log. Implemented by
/// [`SqliteStore`] (rusqlite, blocking under the hood) and
/// [`postgres::PostgresStore`] (sqlx, native async).
///
/// All methods on this trait correspond 1:1 to methods on
/// [`TransactionStore`] for the SQLite path; the Postgres path implements
/// the same semantics over a different SQL dialect.
#[async_trait]
pub trait AuditStore: Send + Sync + std::fmt::Debug {
    async fn record(
        &self,
        transaction: NewTransaction,
    ) -> Result<TransactionRecord, TransactionStoreError>;

    async fn record_previewed(
        &self,
        transaction: NewTransaction,
        preview: PreviewEnvelope,
    ) -> Result<RecordedPreviewedTransaction, TransactionStoreError>;

    async fn get(
        &self,
        transaction_id: &str,
    ) -> Result<Option<TransactionRecord>, TransactionStoreError>;

    async fn get_preview(
        &self,
        transaction_id: &str,
    ) -> Result<Option<PreviewEnvelope>, TransactionStoreError>;

    async fn update_status(
        &self,
        transaction_id: &str,
        new_status: JobState,
    ) -> Result<(), TransactionStoreError>;

    async fn approve_transaction(
        &self,
        transaction_id: &str,
        approver: CallerPrincipal,
    ) -> Result<Option<String>, TransactionStoreError>;

    async fn revoke_unconsumed_approval(
        &self,
        transaction_id: &str,
        revoker: CallerPrincipal,
    ) -> Result<bool, TransactionStoreError>;

    async fn claim_approved_for_execution(
        &self,
        transaction_id: &str,
        receipt_digest: &str,
        executor: CallerPrincipal,
    ) -> Result<bool, TransactionStoreError>;

    async fn cleanup_stale_queued(&self) -> Result<u64, TransactionStoreError>;

    /// Reconcile transactions orphaned by a daemon crash or restart.
    ///
    /// A row still in `Running` means the daemon was mid-execution of that
    /// action when it exited. On the next start the in-flight work is orphaned
    /// and its true outcome is unknown, so each such row is transitioned
    /// `Running → Failed` (the conservative outcome — the operator must
    /// re-check) and the transition is recorded in the audit chain. Intended to
    /// run exactly once at startup, before the daemon begins accepting
    /// connections, so clients never observe a phantom in-flight row.
    ///
    /// Returns the number of transactions reconciled. The default implementation
    /// is backend-agnostic: it pages through `Running` rows via
    /// [`list_transactions`](Self::list_transactions) and applies
    /// [`update_status`](Self::update_status), both of which every backend
    /// already implements with identical semantics.
    async fn reconcile_interrupted_running(&self) -> Result<u64, TransactionStoreError> {
        let mut total = 0u64;
        loop {
            let running = self
                .list_transactions(100, Some("running"), None, None)
                .await?;
            if running.is_empty() {
                break;
            }
            let batch = running.len();
            for tx in running {
                self.update_status(&tx.transaction_id, JobState::Failed)
                    .await?;
            }
            total += batch as u64;
            // A short page means we drained every remaining Running row.
            if batch < 100 {
                break;
            }
        }
        Ok(total)
    }

    /// Cancel one still-`Queued` transaction (`Queued → Canceled`). Returns
    /// `true` iff a queued row was transitioned; a `Running` (in-flight) or
    /// terminal transaction is never cancelled. See
    /// [`crate::transactions::TransactionStore::cancel_queued`].
    async fn cancel_queued(
        &self,
        transaction_id: &str,
        canceller: CallerPrincipal,
    ) -> Result<bool, TransactionStoreError>;

    async fn list_transactions(
        &self,
        limit: u32,
        status_filter: Option<&str>,
        action_filter: Option<&str>,
        since: Option<crate::transactions::HistorySince>,
    ) -> Result<Vec<TransactionRecord>, TransactionStoreError>;

    /// Structured history rows (with `created_at` and typed `risk_level`) for
    /// programmatic clients. See [`crate::transactions::JobHistoryEntry`].
    async fn list_history(
        &self,
        limit: u32,
        status_filter: Option<&str>,
        action_filter: Option<&str>,
        since: Option<crate::transactions::HistorySince>,
    ) -> Result<Vec<crate::transactions::JobHistoryEntry>, TransactionStoreError>;

    async fn fetch_chain_row(
        &self,
        transaction_id: &str,
    ) -> Result<Option<ChainRow>, TransactionStoreError>;

    async fn fetch_chain_rows(&self) -> Result<Vec<ChainRow>, TransactionStoreError>;

    /// Every approval event, in seq order. Backs the second chain that records
    /// grant/consume/revoke — see `audit_chain::AuditEventKind`.
    async fn fetch_event_rows(&self) -> Result<Vec<EventRow>, TransactionStoreError>;

    async fn verify_audit_chain(
        &self,
        key: &AuditKey,
    ) -> Result<VerifyOutcome, TransactionStoreError>;

    /// Walk the approval-event chain.
    ///
    /// Provided rather than required: it is the same signature walk over rows
    /// the backend has already fetched, so a backend that implements
    /// `fetch_event_rows` correctly cannot get this wrong, and an override
    /// would be a second place for the two chains' verification to drift.
    async fn verify_event_chain(
        &self,
        key: &AuditKey,
    ) -> Result<VerifyOutcome, TransactionStoreError> {
        let rows = self.fetch_event_rows().await?;
        Ok(crate::audit_chain::verify_event_chain(key, &rows))
    }

    /// Check that every event tip committed by a transaction row still exists
    /// in the event chain. See `audit_chain::verify_event_binding`.
    async fn verify_event_binding(
        &self,
    ) -> Result<crate::audit_chain::BindingOutcome, TransactionStoreError> {
        let tx_rows = self.fetch_chain_rows().await?;
        let event_rows = self.fetch_event_rows().await?;
        Ok(crate::audit_chain::verify_event_binding(
            &tx_rows,
            &event_rows,
        ))
    }
}

/// Adapter that exposes the existing rusqlite-backed [`TransactionStore`]
/// through the async [`AuditStore`] trait. Each method runs the underlying
/// blocking work on `tokio::task::spawn_blocking` so the executor stays
/// responsive even when the SQLite write takes a millisecond on a slow disk.
#[derive(Debug, Clone)]
pub struct SqliteStore {
    inner: Arc<TransactionStore>,
}

impl SqliteStore {
    pub fn new(inner: TransactionStore) -> Self {
        Self {
            inner: Arc::new(inner),
        }
    }

    /// Borrow the wrapped sync `TransactionStore`. Used by call sites that
    /// genuinely need the blocking handle (test helpers, the audit-verify
    /// CLI subcommand which is itself sync over the result).
    pub fn inner(&self) -> &TransactionStore {
        &self.inner
    }
}

/// Run a blocking closure on the runtime's blocking thread pool.
///
/// `spawn_blocking` returns a `JoinHandle` that resolves to the closure's
/// output; we panic-propagate JoinErrors because they only happen when the
/// runtime is shutting down — at which point the daemon is also shutting
/// down and there's no audit log to write to anyway.
async fn blocking<F, R>(f: F) -> R
where
    F: FnOnce() -> R + Send + 'static,
    R: Send + 'static,
{
    tokio::task::spawn_blocking(f)
        .await
        .expect("audit-store blocking task panicked")
}

#[async_trait]
impl AuditStore for SqliteStore {
    async fn record(
        &self,
        transaction: NewTransaction,
    ) -> Result<TransactionRecord, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        blocking(move || inner.record(transaction)).await
    }

    async fn record_previewed(
        &self,
        transaction: NewTransaction,
        preview: PreviewEnvelope,
    ) -> Result<RecordedPreviewedTransaction, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        blocking(move || inner.record_previewed(transaction, preview)).await
    }

    async fn get(
        &self,
        transaction_id: &str,
    ) -> Result<Option<TransactionRecord>, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let id = transaction_id.to_string();
        blocking(move || inner.get(&id)).await
    }

    async fn get_preview(
        &self,
        transaction_id: &str,
    ) -> Result<Option<PreviewEnvelope>, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let id = transaction_id.to_string();
        blocking(move || inner.get_preview(&id)).await
    }

    async fn update_status(
        &self,
        transaction_id: &str,
        new_status: JobState,
    ) -> Result<(), TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let id = transaction_id.to_string();
        blocking(move || inner.update_status(&id, new_status)).await
    }

    async fn approve_transaction(
        &self,
        transaction_id: &str,
        approver: CallerPrincipal,
    ) -> Result<Option<String>, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let id = transaction_id.to_string();
        blocking(move || inner.approve_transaction(&id, approver)).await
    }

    async fn revoke_unconsumed_approval(
        &self,
        transaction_id: &str,
        revoker: CallerPrincipal,
    ) -> Result<bool, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let id = transaction_id.to_string();
        blocking(move || inner.revoke_unconsumed_approval(&id, revoker)).await
    }

    async fn claim_approved_for_execution(
        &self,
        transaction_id: &str,
        receipt_digest: &str,
        executor: CallerPrincipal,
    ) -> Result<bool, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let id = transaction_id.to_string();
        let digest = receipt_digest.to_string();
        blocking(move || inner.claim_approved_for_execution(&id, &digest, executor)).await
    }

    async fn cleanup_stale_queued(&self) -> Result<u64, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        blocking(move || inner.cleanup_stale_queued()).await
    }

    async fn cancel_queued(
        &self,
        transaction_id: &str,
        canceller: CallerPrincipal,
    ) -> Result<bool, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let id = transaction_id.to_string();
        blocking(move || inner.cancel_queued(&id, canceller)).await
    }

    async fn list_transactions(
        &self,
        limit: u32,
        status_filter: Option<&str>,
        action_filter: Option<&str>,
        since: Option<crate::transactions::HistorySince>,
    ) -> Result<Vec<TransactionRecord>, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let status = status_filter.map(str::to_string);
        let action = action_filter.map(str::to_string);
        blocking(move || {
            inner.list_transactions(limit, status.as_deref(), action.as_deref(), since)
        })
        .await
    }

    async fn list_history(
        &self,
        limit: u32,
        status_filter: Option<&str>,
        action_filter: Option<&str>,
        since: Option<crate::transactions::HistorySince>,
    ) -> Result<Vec<crate::transactions::JobHistoryEntry>, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let status = status_filter.map(str::to_string);
        let action = action_filter.map(str::to_string);
        blocking(move || inner.list_history(limit, status.as_deref(), action.as_deref(), since))
            .await
    }

    async fn fetch_chain_row(
        &self,
        transaction_id: &str,
    ) -> Result<Option<ChainRow>, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        let id = transaction_id.to_string();
        blocking(move || inner.fetch_chain_row(&id)).await
    }

    async fn fetch_chain_rows(&self) -> Result<Vec<ChainRow>, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        blocking(move || inner.fetch_chain_rows()).await
    }

    async fn fetch_event_rows(&self) -> Result<Vec<EventRow>, TransactionStoreError> {
        let inner = Arc::clone(&self.inner);
        blocking(move || inner.fetch_event_rows()).await
    }

    async fn verify_audit_chain(
        &self,
        key: &AuditKey,
    ) -> Result<VerifyOutcome, TransactionStoreError> {
        // `key` is borrowed from the caller; we can't move it into a
        // `'static` `spawn_blocking` closure. Verifying is a CPU-only signature
        // walk over rows already fetched, so doing the work on the async
        // thread (after the blocking fetch) is fine — no I/O risk.
        let rows = self.fetch_chain_rows().await?;
        Ok(crate::audit_chain::verify_chain(key, &rows))
    }
}

// Re-export the postgres impl so call sites can `use store::PostgresStore`.
pub use postgres::PostgresStore;

#[cfg(test)]
mod tests {
    use super::*;
    use crate::audit_chain::AuditKey;
    use crate::auth::CallerPrincipal;
    use crate::transactions::NewTransaction;
    use sysknife_types::{CallerRole, RiskLevel};
    use tempfile::tempdir;

    fn sqlite_store(path: std::path::PathBuf) -> SqliteStore {
        let key = Arc::new(AuditKey::from_bytes(vec![0x42; 32]));
        SqliteStore::new(crate::transactions::TransactionStore::open_with_key(path, key).unwrap())
    }

    /// `CHAIN_ROW_COLUMNS` is now declared once (store.rs), so there is no
    /// second copy left to compare — the drift test that justified the hoist
    /// ran green first, was mutated red on purpose (see #397's PR), and is
    /// replaced by this pin. The invariant that matters is the one the
    /// positional mapper (`chain_row_from_sqlite`) and the by-name mapper
    /// (`row_to_chain_row`) both depend on: 17 columns, in this exact order.
    /// Adding a column to the chain means updating this list deliberately,
    /// together with both mappers.
    #[test]
    fn chain_row_columns_pinned() {
        let columns: Vec<&str> = crate::store::CHAIN_ROW_COLUMNS
            .split(',')
            .map(str::trim)
            .collect();
        assert_eq!(
            columns,
            [
                "seq",
                "key_id",
                "transaction_id",
                "request_id",
                "request_hash",
                "action_name",
                "risk_level",
                "summary",
                "approval_id",
                "warnings_json",
                "created_at",
                "prev_chain_hash",
                "chain_hash",
                "chain_version",
                "caller_role",
                "event_tip",
                "caller_principal",
            ],
            "CHAIN_ROW_COLUMNS changed — both ChainRow mappers read these \
             columns in this order, so update them in the same change"
        );
        assert_eq!(columns.len(), 17, "column count drifted from 17");
    }

    fn new_tx(request_id: &str) -> NewTransaction {
        NewTransaction {
            request_id: request_id.to_string(),
            request_hash: format!("hash-{request_id}"),
            action_name: "UpdateSystem".to_string(),
            risk_level: RiskLevel::High,
            summary: "Upgrade the system".to_string(),
            warnings: vec![],
            caller_role: CallerRole::Dev,
            caller_principal: CallerPrincipal::Uid(1000),
        }
    }

    #[tokio::test]
    async fn reconcile_interrupted_running_fails_only_running_rows() {
        let dir = tempdir().unwrap();
        let store = sqlite_store(dir.path().join("tx.db"));

        // A row left Running by a crash, plus two rows in terminal/other states
        // that reconciliation must not touch.
        let running = store.record(new_tx("running")).await.unwrap();
        store
            .update_status(&running.transaction_id, JobState::Running)
            .await
            .unwrap();

        let queued = store.record(new_tx("queued")).await.unwrap();

        let succeeded = store.record(new_tx("succeeded")).await.unwrap();
        store
            .update_status(&succeeded.transaction_id, JobState::Running)
            .await
            .unwrap();
        store
            .update_status(&succeeded.transaction_id, JobState::Succeeded)
            .await
            .unwrap();

        let reconciled = store.reconcile_interrupted_running().await.unwrap();
        assert_eq!(reconciled, 1, "only the one Running row is reconciled");

        assert_eq!(
            store
                .get(&running.transaction_id)
                .await
                .unwrap()
                .unwrap()
                .status,
            JobState::Failed,
            "orphaned Running row must become Failed"
        );
        assert_eq!(
            store
                .get(&queued.transaction_id)
                .await
                .unwrap()
                .unwrap()
                .status,
            JobState::Queued,
            "Queued row is untouched"
        );
        assert_eq!(
            store
                .get(&succeeded.transaction_id)
                .await
                .unwrap()
                .unwrap()
                .status,
            JobState::Succeeded,
            "terminal Succeeded row is untouched"
        );

        // Idempotent: a second pass finds nothing left to reconcile.
        assert_eq!(store.reconcile_interrupted_running().await.unwrap(), 0);
    }
}