sysknife_daemon/store/

postgres.rs

//! Postgres backend for the audit log.
//!
//! Wire-compatible with AWS RDS / Aurora, GCP Cloud SQL + AlloyDB, Azure
//! Database for PostgreSQL Flexible Server, Supabase (direct + pooler),
//! Neon (direct + pooler), and self-hosted Postgres.
//!
//! ## Connection lifecycle
//!
//! [`PostgresStore::connect`] takes a [`PostgresConfig`] (URL plus pool +
//! statement-cache tuning) and returns a configured `sqlx::PgPool`. The
//! same pool is used for every request — sqlx handles connection reuse,
//! retry on broken connections, and TLS via rustls.
//!
//! On connect, `PostgresStore::initialize` runs ordered, transactional schema
//! migrations under a database advisory lock. The schema mirrors SQLite
//! field-for-field; the only dialect
//! difference is `BIGINT NOT NULL UNIQUE` for `seq`. `created_at` remains
//! `TEXT` (RFC 3339 with `Z` suffix) in both backends and is cast to
//! `timestamptz` for interval arithmetic.
//!
//! ## Concurrency
//!
//! Like the SQLite path, the chain-hash + seq computation is wrapped in a
//! `BEGIN ... COMMIT` block. Postgres serialises this naturally via
//! row-level locks on the `seq` write; we use a `SELECT … FOR UPDATE` on
//! the most-recent row so concurrent writers see the same predecessor and
//! the second commit fails atomically rather than corrupting the chain.

use async_trait::async_trait;
use sqlx_core::row::Row;
use sqlx_postgres::{PgConnectOptions, PgPool, PgPoolOptions};
use std::str::FromStr;
use std::time::Duration;
use subtle::ConstantTimeEq;
use sysknife_types::{JobState, PreviewEnvelope, TransactionRecord};
use uuid::Uuid;

use crate::audit_chain::{
    AttributionCensus, AuditEventKind, AuditKey, AuditVerification, BindingOutcome, ChainContent,
    ChainIdentity, ChainRow, EventContent, EventRow, VerifyOutcome, CURRENT_KEY_ID,
};
use crate::audit_watermark::emit_chain_tip_watermark;
use crate::store::AuditStore;
use crate::transactions::{
    NewTransaction, RecordedPreviewedTransaction, TransactionStoreError,
    APPROVAL_RECEIPT_TTL_MINUTES,
};

const MIGRATION_LOCK_ID: i64 = 0x5359_534b_4e49_4645;

/// Column list every `ChainRow` read shares, kept next to `row_to_chain_row`.
///
/// The two read paths each spelled the list out. 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. Mirrors `CHAIN_ROW_COLUMNS` in `transactions.rs`.
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";

struct Migration {
    version: i64,
    name: &'static str,
    statements: &'static [&'static str],
}

const MIGRATIONS: &[Migration] = &[Migration {
    version: 1,
    name: "initial_audit_schema",
    statements: &[
        r#"
        CREATE TABLE IF NOT EXISTS transactions (
            transaction_id TEXT PRIMARY KEY,
            request_id TEXT NOT NULL,
            request_hash TEXT NOT NULL,
            action_name TEXT NOT NULL,
            risk_level TEXT NOT NULL,
            status TEXT NOT NULL,
            approval_id TEXT,
            summary TEXT NOT NULL,
            warnings_json TEXT NOT NULL,
            created_at TEXT NOT NULL,
            seq BIGINT NOT NULL UNIQUE,
            key_id TEXT NOT NULL,
            chain_hash TEXT NOT NULL,
            prev_chain_hash TEXT NOT NULL DEFAULT ''
        )
        "#,
        r#"
        CREATE TABLE IF NOT EXISTS transaction_approvals (
            transaction_id TEXT PRIMARY KEY,
            receipt_digest TEXT NOT NULL,
            approved_at TEXT NOT NULL,
            consumed_at TEXT
        )
        "#,
        r#"
        CREATE TABLE IF NOT EXISTS transaction_previews (
            transaction_id TEXT PRIMARY KEY,
            preview_json TEXT NOT NULL
        )
        "#,
        "CREATE INDEX IF NOT EXISTS transactions_seq_idx ON transactions(seq)",
    ],
},
    // Mirrors SQLITE_MIGRATIONS v2 in `transactions.rs`. The columns are
    // nullable with a legacy default because rows written by an earlier binary
    // were signed over an encoding without them — see `ChainIdentity`.
    Migration {
        version: 2,
        name: "caller_identity_and_approval_events",
        statements: &[
            "ALTER TABLE transactions ADD COLUMN IF NOT EXISTS chain_version BIGINT NOT NULL DEFAULT 1",
            "ALTER TABLE transactions ADD COLUMN IF NOT EXISTS caller_role TEXT",
            "ALTER TABLE transactions ADD COLUMN IF NOT EXISTS event_tip TEXT",
            r#"
            CREATE TABLE IF NOT EXISTS audit_events (
                seq BIGINT PRIMARY KEY,
                key_id TEXT NOT NULL,
                kind TEXT NOT NULL,
                transaction_id TEXT NOT NULL,
                receipt_digest TEXT NOT NULL,
                created_at TEXT NOT NULL,
                chain_hash TEXT NOT NULL,
                prev_chain_hash TEXT NOT NULL DEFAULT ''
            )
            "#,
            "CREATE INDEX IF NOT EXISTS audit_events_transaction_idx ON audit_events(transaction_id)",
        ],
    },
    // Mirrors SQLite migration 3. Nullable for the same reason: rows written
    // before this step were signed over an encoding with no principal, so any
    // backfill would rewrite their message and report the chain as Broken.
    Migration {
        version: 3,
        name: "caller_principal",
        statements: &["ALTER TABLE transactions ADD COLUMN IF NOT EXISTS caller_principal TEXT"],
    },
];

/// Configuration for the Postgres backend. Built by `main.rs` from
/// `[storage]` in `config.toml`.
#[derive(Clone, Debug)]
pub struct PostgresConfig {
    /// `postgres://...` URL. SSL mode and other knobs are encoded here.
    pub url: String,
    /// Maximum pool size. 8 is comfortable for a single SysKnife daemon;
    /// raise if multiple shells run concurrently.
    pub max_connections: u32,
    /// `acquire_timeout` for getting a connection from the pool. Raise above
    /// the default 5s if you hit Neon cold starts (~600 ms first call).
    pub acquire_timeout: Duration,
    /// Set to 0 for transaction-mode PgBouncer (including some hosted
    /// poolers), which cannot use sqlx's per-connection prepared-statement
    /// cache. Default 100 covers direct Postgres connections.
    pub statement_cache_capacity: usize,
}

impl Default for PostgresConfig {
    fn default() -> Self {
        Self {
            url: String::new(),
            max_connections: 8,
            acquire_timeout: Duration::from_secs(10),
            statement_cache_capacity: 100,
        }
    }
}

/// Postgres-backed [`AuditStore`].
#[derive(Debug, Clone)]
pub struct PostgresStore {
    pool: PgPool,
    audit_key: std::sync::Arc<AuditKey>,
}

struct InsertedTransaction {
    record: TransactionRecord,
    seq: u64,
    chain_hash: String,
}

impl PostgresStore {
    async fn connect_pool(config: &PostgresConfig) -> Result<PgPool, TransactionStoreError> {
        let mut connect_opts = PgConnectOptions::from_str(&config.url).map_err(|e| {
            TransactionStoreError::Io(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                format!("invalid postgres URL: {e}"),
            ))
        })?;
        // Refuse a remote connection that could leak the credential (#149). This
        // is a hard configuration refusal, not a transient I/O error: it fails
        // daemon startup and is never retried. It is mapped onto Io(PermissionDenied)
        // only because that is the closest existing variant; nothing branches on
        // the ErrorKind for this path.
        crate::pg_tls::require_tls_for_remote(&connect_opts).map_err(|msg| {
            TransactionStoreError::Io(std::io::Error::new(
                std::io::ErrorKind::PermissionDenied,
                msg,
            ))
        })?;
        connect_opts = connect_opts.statement_cache_capacity(config.statement_cache_capacity);
        PgPoolOptions::new()
            .max_connections(config.max_connections)
            .acquire_timeout(config.acquire_timeout)
            .connect_with(connect_opts)
            .await
            .map_err(map_sqlx_err)
    }

    /// Connect to the configured Postgres database, run the migration,
    /// and bind an audit key for chain computation.
    pub async fn connect(
        config: &PostgresConfig,
        audit_key: std::sync::Arc<AuditKey>,
    ) -> Result<Self, TransactionStoreError> {
        let pool = Self::connect_pool(config).await?;

        let store = Self { pool, audit_key };
        store.initialize().await?;
        Ok(store)
    }

    /// Verify an existing Postgres chain — transaction chain, approval-event
    /// chain, and the binding between them — with only the exported public
    /// key. This path performs no migrations and never loads the signing key.
    pub async fn verify_all_with_pubkey(
        config: &PostgresConfig,
        verifying_key_hex: &str,
    ) -> Result<AuditVerification, TransactionStoreError> {
        let pool = Self::connect_pool(config).await?;
        let tx_rows = fetch_chain_rows_from_pool(&pool).await?;
        let event_rows = match fetch_event_rows_from_pool(&pool).await {
            Ok(rows) => rows,
            Err(err) => {
                // A chain written before the migration has no `audit_events`
                // table. That is an absent feature, not a missing verification:
                // report zero events rather than failing the whole verify.
                //
                // That is the ONLY reason to carry on. This used to be
                // `unwrap_or_default()`, which also swallowed a revoked SELECT,
                // a dropped table, a lost connection and a malformed row — each
                // becoming "zero approval events", which is either a clean bill
                // for a trail nobody read or, when a transaction row commits a
                // non-empty event tip, a `MissingEvent` binding published in the
                // words operators are taught to read as deleted approvals.
                //
                // So ask specifically whether the table is absent, and fail
                // closed on everything else, including a probe that cannot
                // answer: an audit tool that cannot read its evidence must say
                // so rather than grade it.
                match audit_events_table_absent(&pool).await {
                    Ok(true) => Vec::new(),
                    Ok(false) => {
                        return Ok(Self::unreadable_events(
                            verifying_key_hex,
                            &tx_rows,
                            format!("could not read the approval-event table: {err}"),
                        ))
                    }
                    Err(probe_err) => {
                        return Ok(Self::unreadable_events(
                            verifying_key_hex,
                            &tx_rows,
                            format!(
                                "could not read the approval-event table ({err}), and could not \
                                 determine whether it exists ({probe_err})"
                            ),
                        ))
                    }
                }
            }
        };
        Ok(crate::audit_chain::verify_all_with_pubkey(
            verifying_key_hex,
            &tx_rows,
            &event_rows,
        ))
    }

    /// The transaction chain still verifies on its own — report it — but the
    /// event chain was not read, so its verdict is inconclusive and the binding
    /// check between the two did not run. `NotChecked` is what keeps that
    /// distinct from a binding that ran over zero events and agreed.
    fn unreadable_events(
        verifying_key_hex: &str,
        tx_rows: &[ChainRow],
        reason: String,
    ) -> AuditVerification {
        AuditVerification {
            chain: crate::audit_chain::verify_chain_with_pubkey(verifying_key_hex, tx_rows),
            events: VerifyOutcome::CannotVerify { reason },
            binding: BindingOutcome::NotChecked,
            attribution: Some(AttributionCensus::of(tx_rows)),
        }
    }

    /// All three checks with the daemon's key.
    pub async fn verify_all(
        &self,
        key: &AuditKey,
    ) -> Result<AuditVerification, TransactionStoreError> {
        let tx_rows = fetch_chain_rows_from_pool(&self.pool).await?;
        let event_rows = fetch_event_rows_from_pool(&self.pool).await?;
        Ok(crate::audit_chain::verify_all(key, &tx_rows, &event_rows))
    }

    /// Apply pending schema migrations atomically. The advisory transaction
    /// lock serializes concurrent daemon starts against the same database.
    async fn initialize(&self) -> Result<(), TransactionStoreError> {
        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
        sqlx_core::query::query("SELECT pg_advisory_xact_lock($1)")
            .bind(MIGRATION_LOCK_ID)
            .execute(&mut *tx)
            .await
            .map_err(map_sqlx_err)?;

        sqlx_core::query::query(
            r#"
            CREATE TABLE IF NOT EXISTS schema_migrations (
                version BIGINT PRIMARY KEY,
                name TEXT NOT NULL,
                applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
            )
            "#,
        )
        .execute(&mut *tx)
        .await
        .map_err(map_sqlx_err)?;

        let current: i64 = sqlx_core::query_scalar::query_scalar(
            "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
        )
        .fetch_one(&mut *tx)
        .await
        .map_err(map_sqlx_err)?;
        let latest = MIGRATIONS.last().map_or(0, |migration| migration.version);
        if current > latest {
            return Err(TransactionStoreError::DatabaseInvariant(format!(
                "postgres schema version {current} is newer than this binary supports ({latest})"
            )));
        }

        for migration in MIGRATIONS
            .iter()
            .filter(|migration| migration.version > current)
        {
            // Run statements separately for transaction-mode poolers that
            // reject multi-statement query strings.
            for statement in migration.statements {
                sqlx_core::query::query(*statement)
                    .execute(&mut *tx)
                    .await
                    .map_err(map_sqlx_err)?;
            }
            sqlx_core::query::query(
                "INSERT INTO schema_migrations (version, name) VALUES ($1, $2)",
            )
            .bind(migration.version)
            .bind(migration.name)
            .execute(&mut *tx)
            .await
            .map_err(map_sqlx_err)?;
        }

        tx.commit().await.map_err(map_sqlx_err)
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

fn map_sqlx_err(e: sqlx_core::Error) -> TransactionStoreError {
    let message = format!("postgres: {e}");
    let invariant = match &e {
        sqlx_core::Error::Database(db) => db
            .code()
            .is_some_and(|code| code.starts_with("22") || code.starts_with("23")),
        sqlx_core::Error::TypeNotFound { .. }
        | sqlx_core::Error::ColumnIndexOutOfBounds { .. }
        | sqlx_core::Error::ColumnNotFound(_)
        | sqlx_core::Error::ColumnDecode { .. }
        | sqlx_core::Error::Decode(_) => true,
        _ => false,
    };
    if invariant {
        TransactionStoreError::DatabaseInvariant(message)
    } else {
        TransactionStoreError::Io(std::io::Error::other(message))
    }
}

fn serialize<T: serde::Serialize>(v: &T) -> Result<String, TransactionStoreError> {
    serde_json::to_string(v).map_err(TransactionStoreError::Json)
}

fn deserialize<T: serde::de::DeserializeOwned>(s: &str) -> Result<T, TransactionStoreError> {
    serde_json::from_str(s).map_err(TransactionStoreError::Json)
}

/// UTC timestamp in the same RFC 3339 millisecond shape the SQLite backend
/// produces via `strftime('%Y-%m-%dT%H:%M:%fZ','now')`.
///
/// `%S` is load-bearing: chrono's `%.3f` renders only the fractional part, so
/// omitting `%S` yields `2026-07-27T12:34:.567Z`. That value is signed into
/// `ChainContent.created_at` on every audit row and cast to `timestamptz` by
/// approve, claim-for-execution, the stale sweep, and the history filter.
fn now_iso() -> String {
    chrono::Utc::now()
        .format("%Y-%m-%dT%H:%M:%S%.3fZ")
        .to_string()
}

// ---------------------------------------------------------------------------
// AuditStore impl
// ---------------------------------------------------------------------------

#[async_trait]
impl AuditStore for PostgresStore {
    async fn record(
        &self,
        transaction: NewTransaction,
    ) -> Result<TransactionRecord, TransactionStoreError> {
        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
        let transaction_id = Uuid::new_v4().to_string();
        let inserted =
            insert_transaction(&mut tx, &self.audit_key, &transaction_id, transaction).await?;
        tx.commit().await.map_err(map_sqlx_err)?;
        emit_chain_tip_watermark(inserted.seq, &inserted.chain_hash);
        Ok(inserted.record)
    }

    async fn record_previewed(
        &self,
        transaction: NewTransaction,
        preview: PreviewEnvelope,
    ) -> Result<RecordedPreviewedTransaction, TransactionStoreError> {
        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
        let transaction_id = Uuid::new_v4().to_string();
        let inserted =
            insert_transaction(&mut tx, &self.audit_key, &transaction_id, transaction).await?;

        sqlx_core::query::query(
            "INSERT INTO transaction_previews (transaction_id, preview_json) VALUES ($1, $2)",
        )
        .bind(&inserted.record.transaction_id)
        .bind(serialize(&preview)?)
        .execute(&mut *tx)
        .await
        .map_err(map_sqlx_err)?;

        tx.commit().await.map_err(map_sqlx_err)?;
        emit_chain_tip_watermark(inserted.seq, &inserted.chain_hash);
        Ok(RecordedPreviewedTransaction {
            transaction: inserted.record,
            preview,
        })
    }

    async fn get(
        &self,
        transaction_id: &str,
    ) -> Result<Option<TransactionRecord>, TransactionStoreError> {
        let row = sqlx_core::query::query(
            "SELECT transaction_id, request_id, request_hash, action_name, risk_level, \
                    status, approval_id, summary, warnings_json \
             FROM transactions WHERE transaction_id = $1",
        )
        .bind(transaction_id)
        .fetch_optional(&self.pool)
        .await
        .map_err(map_sqlx_err)?;
        row.map(row_to_record).transpose()
    }

    async fn get_preview(
        &self,
        transaction_id: &str,
    ) -> Result<Option<PreviewEnvelope>, TransactionStoreError> {
        let row = sqlx_core::query::query(
            "SELECT preview_json FROM transaction_previews WHERE transaction_id = $1",
        )
        .bind(transaction_id)
        .fetch_optional(&self.pool)
        .await
        .map_err(map_sqlx_err)?;
        match row {
            Some(r) => {
                let s: String = r.try_get("preview_json").map_err(map_sqlx_err)?;
                Ok(Some(deserialize(&s)?))
            }
            None => Ok(None),
        }
    }

    async fn update_status(
        &self,
        transaction_id: &str,
        new_status: JobState,
    ) -> Result<(), TransactionStoreError> {
        // Read-validate-write with a state-machine guard, same shape as the
        // SQLite path but strictly stronger: `SELECT ... FOR UPDATE` takes a
        // row lock for the lifetime of this transaction, so no concurrent
        // writer can change `status` between our read and our `UPDATE` — the
        // read-modify-write here is atomic by construction. SQLite has no
        // equivalent row-level lock, so its `update_status` instead uses an
        // explicit compare-and-swap (`UPDATE ... WHERE status = <observed>`,
        // checking `rows_affected`) to get the same guarantee.
        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
        let current_str: Option<String> = sqlx_core::query_scalar::query_scalar(
            "SELECT status FROM transactions WHERE transaction_id = $1 FOR UPDATE",
        )
        .bind(transaction_id)
        .fetch_optional(&mut *tx)
        .await
        .map_err(map_sqlx_err)?;
        let current_str = current_str
            .ok_or_else(|| TransactionStoreError::NotFound(transaction_id.to_string()))?;
        let current: JobState = deserialize(&current_str)?;
        if !crate::jobs::allowed_transition(&current, &new_status) {
            return Err(TransactionStoreError::InvalidTransition {
                from: current,
                to: new_status,
            });
        }
        sqlx_core::query::query("UPDATE transactions SET status = $1 WHERE transaction_id = $2")
            .bind(serialize(&new_status)?)
            .bind(transaction_id)
            .execute(&mut *tx)
            .await
            .map_err(map_sqlx_err)?;
        tx.commit().await.map_err(map_sqlx_err)?;
        Ok(())
    }

    async fn approve_transaction(
        &self,
        transaction_id: &str,
    ) -> Result<Option<String>, TransactionStoreError> {
        let row = sqlx_core::query::query(
            "SELECT request_hash, approval_id FROM transactions WHERE transaction_id = $1",
        )
        .bind(transaction_id)
        .fetch_optional(&self.pool)
        .await
        .map_err(map_sqlx_err)?;
        let Some(row) = row else {
            return Ok(None);
        };
        let request_hash: String = row.try_get("request_hash").map_err(map_sqlx_err)?;
        let committed_digest: Option<String> = row.try_get("approval_id").map_err(map_sqlx_err)?;
        let receipt = self
            .audit_key
            .approval_receipt(transaction_id, &request_hash);
        let receipt_digest = crate::audit_chain::approval_receipt_digest(&receipt);
        let commitment_matches = committed_digest.as_deref().is_some_and(|committed| {
            bool::from(receipt_digest.as_bytes().ct_eq(committed.as_bytes()))
        });
        if !commitment_matches {
            return Err(TransactionStoreError::DatabaseInvariant(format!(
                "transaction {transaction_id} approval commitment does not match its signed preview"
            )));
        }
        let queued = serialize(&JobState::Queued)?;
        // Approval row and chained event commit together — an approval absent
        // from the event chain is exactly what that chain exists to prevent.
        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
        let result = sqlx_core::query::query(
            sqlx_core::sql_str::AssertSqlSafe(format!(
                "INSERT INTO transaction_approvals \
                     (transaction_id, receipt_digest, approved_at) \
                 SELECT transaction_id, $1, $2 FROM transactions \
                 WHERE transaction_id = $3 \
                   AND status = $4 \
                   AND created_at::timestamptz > now() - INTERVAL '{APPROVAL_RECEIPT_TTL_MINUTES} minutes' \
                 ON CONFLICT (transaction_id) DO NOTHING"
            )),
        )
        .bind(&receipt_digest)
        .bind(now_iso())
        .bind(transaction_id)
        .bind(&queued)
        .execute(&mut *tx)
        .await
        .map_err(map_sqlx_err)?;
        if result.rows_affected() > 0 {
            append_event(
                &mut tx,
                &self.audit_key,
                AuditEventKind::ApprovalGranted,
                transaction_id,
                &receipt_digest,
            )
            .await?;
        }
        tx.commit().await.map_err(map_sqlx_err)?;
        Ok((result.rows_affected() > 0).then_some(receipt))
    }

    async fn revoke_unconsumed_approval(
        &self,
        transaction_id: &str,
    ) -> Result<bool, TransactionStoreError> {
        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
        // Read the digest before the DELETE: the event names which receipt was
        // retracted, and after the delete there is nothing left to name.
        let digest: Option<String> = sqlx_core::query_scalar::query_scalar(
            "SELECT receipt_digest FROM transaction_approvals \
             WHERE transaction_id = $1 AND consumed_at IS NULL",
        )
        .bind(transaction_id)
        .fetch_optional(&mut *tx)
        .await
        .map_err(map_sqlx_err)?;
        let result = sqlx_core::query::query(
            "DELETE FROM transaction_approvals \
             WHERE transaction_id = $1 AND consumed_at IS NULL",
        )
        .bind(transaction_id)
        .execute(&mut *tx)
        .await
        .map_err(map_sqlx_err)?;
        if result.rows_affected() > 0 {
            let digest = digest.ok_or_else(|| {
                TransactionStoreError::DatabaseInvariant(format!(
                    "revoked an approval for {transaction_id} that had no receipt digest"
                ))
            })?;
            append_event(
                &mut tx,
                &self.audit_key,
                AuditEventKind::ApprovalRevoked,
                transaction_id,
                &digest,
            )
            .await?;
        }
        tx.commit().await.map_err(map_sqlx_err)?;
        Ok(result.rows_affected() > 0)
    }

    async fn claim_approved_for_execution(
        &self,
        transaction_id: &str,
        receipt_digest: &str,
    ) -> Result<bool, TransactionStoreError> {
        let queued = serialize(&JobState::Queued)?;
        let running = serialize(&JobState::Running)?;
        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
        let result = sqlx_core::query::query(
            sqlx_core::sql_str::AssertSqlSafe(format!(
                "UPDATE transactions SET status = $1 \
                 WHERE transaction_id = $2 \
                   AND status = $3 \
                   AND created_at::timestamptz > now() - INTERVAL '{APPROVAL_RECEIPT_TTL_MINUTES} minutes' \
                   AND EXISTS ( \
                       SELECT 1 FROM transaction_approvals \
                       WHERE transaction_id = $2 \
                         AND receipt_digest = $4 \
                         AND consumed_at IS NULL \
                   )"
            )),
        )
        .bind(&running)
        .bind(transaction_id)
        .bind(&queued)
        .bind(receipt_digest)
        .execute(&mut *tx)
        .await
        .map_err(map_sqlx_err)?;
        if result.rows_affected() > 0 {
            sqlx_core::query::query(
                "UPDATE transaction_approvals SET consumed_at = $1 \
                 WHERE transaction_id = $2 AND consumed_at IS NULL",
            )
            .bind(now_iso())
            .bind(transaction_id)
            .execute(&mut *tx)
            .await
            .map_err(map_sqlx_err)?;
            append_event(
                &mut tx,
                &self.audit_key,
                AuditEventKind::ApprovalConsumed,
                transaction_id,
                receipt_digest,
            )
            .await?;
        }
        tx.commit().await.map_err(map_sqlx_err)?;
        Ok(result.rows_affected() > 0)
    }

    async fn cleanup_stale_queued(&self) -> Result<u64, TransactionStoreError> {
        let queued = serialize(&JobState::Queued)?;
        let canceled = serialize(&JobState::Canceled)?;
        let result = sqlx_core::query::query(
            sqlx_core::sql_str::AssertSqlSafe(format!(
                "UPDATE transactions SET status = $1 \
                 WHERE status = $2 \
                   AND created_at::timestamptz <= now() - INTERVAL '{APPROVAL_RECEIPT_TTL_MINUTES} minutes'"
            )),
        )
        .bind(&canceled)
        .bind(&queued)
        .execute(&self.pool)
        .await
        .map_err(map_sqlx_err)?;
        Ok(result.rows_affected())
    }

    async fn cancel_queued(&self, transaction_id: &str) -> Result<bool, TransactionStoreError> {
        // Option A: the `status = $3` (Queued) guard means a Running (in-flight)
        // transaction is never cancelled.
        let queued = serialize(&JobState::Queued)?;
        let canceled = serialize(&JobState::Canceled)?;
        let result = sqlx_core::query::query(
            "UPDATE transactions SET status = $1 \
             WHERE transaction_id = $2 AND status = $3",
        )
        .bind(&canceled)
        .bind(transaction_id)
        .bind(&queued)
        .execute(&self.pool)
        .await
        .map_err(map_sqlx_err)?;
        Ok(result.rows_affected() > 0)
    }

    async fn list_transactions(
        &self,
        limit: u32,
        status_filter: Option<&str>,
        action_filter: Option<&str>,
        since_hours: Option<u32>,
    ) -> Result<Vec<TransactionRecord>, TransactionStoreError> {
        let limit = limit.min(100) as i64;

        // Validate the status filter against the JobState enum so a typo
        // produces an error instead of silent empty results.
        let validated_status: Option<String> = status_filter
            .map(|s| -> Result<String, TransactionStoreError> {
                let parsed: JobState = deserialize(&format!("\"{s}\""))?;
                serialize(&parsed)
            })
            .transpose()?;

        let mut sql = String::from(
            "SELECT transaction_id, request_id, request_hash, action_name, risk_level, \
                    status, approval_id, summary, warnings_json \
             FROM transactions WHERE TRUE",
        );
        let mut idx = 1;
        if validated_status.is_some() {
            sql.push_str(&format!(" AND status = ${idx}"));
            idx += 1;
        }
        if action_filter.is_some() {
            sql.push_str(&format!(" AND action_name = ${idx}"));
            idx += 1;
        }
        if since_hours.is_some() {
            sql.push_str(&format!(
                " AND created_at::timestamptz > now() - (${idx} || ' hours')::INTERVAL"
            ));
            idx += 1;
        }
        sql.push_str(&format!(" ORDER BY seq DESC LIMIT ${idx}"));

        let mut q = sqlx_core::query::query(sqlx_core::sql_str::AssertSqlSafe(sql.as_str()));
        if let Some(s) = &validated_status {
            q = q.bind(s);
        }
        if let Some(a) = action_filter {
            q = q.bind(a);
        }
        if let Some(h) = since_hours {
            q = q.bind(h.to_string());
        }
        q = q.bind(limit);

        let rows = q.fetch_all(&self.pool).await.map_err(map_sqlx_err)?;
        rows.into_iter().map(row_to_record).collect()
    }

    async fn list_history(
        &self,
        limit: u32,
        status_filter: Option<&str>,
        action_filter: Option<&str>,
        since_hours: Option<u32>,
    ) -> Result<Vec<crate::transactions::JobHistoryEntry>, TransactionStoreError> {
        // Parallels list_transactions but selects created_at and returns the
        // structured JobHistoryEntry DTO. Same filter/validation semantics.
        let limit = limit.min(100) as i64;
        let validated_status: Option<String> = status_filter
            .map(|s| -> Result<String, TransactionStoreError> {
                let parsed: JobState = deserialize(&format!("\"{s}\""))?;
                serialize(&parsed)
            })
            .transpose()?;

        let mut sql = String::from(
            "SELECT transaction_id, action_name, risk_level, status, summary, created_at \
             FROM transactions WHERE TRUE",
        );
        let mut idx = 1;
        if validated_status.is_some() {
            sql.push_str(&format!(" AND status = ${idx}"));
            idx += 1;
        }
        if action_filter.is_some() {
            sql.push_str(&format!(" AND action_name = ${idx}"));
            idx += 1;
        }
        if since_hours.is_some() {
            sql.push_str(&format!(
                " AND created_at::timestamptz > now() - (${idx} || ' hours')::INTERVAL"
            ));
            idx += 1;
        }
        sql.push_str(&format!(" ORDER BY seq DESC LIMIT ${idx}"));

        let mut q = sqlx_core::query::query(sqlx_core::sql_str::AssertSqlSafe(sql.as_str()));
        if let Some(s) = &validated_status {
            q = q.bind(s);
        }
        if let Some(a) = action_filter {
            q = q.bind(a);
        }
        if let Some(h) = since_hours {
            q = q.bind(h.to_string());
        }
        q = q.bind(limit);

        let rows = q.fetch_all(&self.pool).await.map_err(map_sqlx_err)?;
        rows.into_iter().map(row_to_history_entry).collect()
    }

    async fn fetch_chain_row(
        &self,
        transaction_id: &str,
    ) -> Result<Option<ChainRow>, TransactionStoreError> {
        let row = sqlx_core::query::query(sqlx_core::sql_str::AssertSqlSafe(format!(
            "SELECT {CHAIN_ROW_COLUMNS} FROM transactions WHERE transaction_id = $1"
        )))
        .bind(transaction_id)
        .fetch_optional(&self.pool)
        .await
        .map_err(map_sqlx_err)?;
        row.map(row_to_chain_row).transpose()
    }

    async fn fetch_event_rows(&self) -> Result<Vec<EventRow>, TransactionStoreError> {
        fetch_event_rows_from_pool(&self.pool).await
    }

    async fn fetch_chain_rows(&self) -> Result<Vec<ChainRow>, TransactionStoreError> {
        fetch_chain_rows_from_pool(&self.pool).await
    }

    async fn verify_audit_chain(
        &self,
        key: &AuditKey,
    ) -> Result<VerifyOutcome, TransactionStoreError> {
        let rows = self.fetch_chain_rows().await?;
        Ok(crate::audit_chain::verify_chain(key, &rows))
    }
}

async fn fetch_chain_rows_from_pool(pool: &PgPool) -> Result<Vec<ChainRow>, TransactionStoreError> {
    let rows = sqlx_core::query::query(sqlx_core::sql_str::AssertSqlSafe(format!(
        "SELECT {CHAIN_ROW_COLUMNS} FROM transactions ORDER BY seq ASC"
    )))
    .fetch_all(pool)
    .await
    .map_err(map_sqlx_err)?;
    rows.into_iter().map(row_to_chain_row).collect()
}

/// Is `audit_events` genuinely not there, as opposed to unreadable?
///
/// `to_regclass` resolves the name through the same `search_path` the queries
/// use and returns NULL when nothing matches. It answers existence, not access:
/// a table the caller may not SELECT still resolves, which is exactly the
/// distinction the caller needs.
async fn audit_events_table_absent(pool: &PgPool) -> Result<bool, TransactionStoreError> {
    let present: Option<String> =
        sqlx_core::query_scalar::query_scalar("SELECT to_regclass('audit_events')::text")
            .fetch_one(pool)
            .await
            .map_err(map_sqlx_err)?;
    Ok(present.is_none())
}

/// Fetch every approval event in seq order.
async fn fetch_event_rows_from_pool(pool: &PgPool) -> Result<Vec<EventRow>, TransactionStoreError> {
    let rows = sqlx_core::query::query(
        "SELECT seq, key_id, kind, transaction_id, receipt_digest, \
                created_at, prev_chain_hash, chain_hash \
         FROM audit_events ORDER BY seq ASC",
    )
    .fetch_all(pool)
    .await
    .map_err(map_sqlx_err)?;
    rows.into_iter().map(row_to_event_row).collect()
}

/// Append one signed approval event inside the caller's transaction.
///
/// `FOR UPDATE` on the current tip serialises concurrent appends the same way
/// `insert_transaction` serialises on the transaction chain tip; without it two
/// approvals could compute the same `seq` and collide on the primary key.
async fn append_event(
    tx: &mut sqlx_core::transaction::Transaction<'_, sqlx_postgres::Postgres>,
    key: &AuditKey,
    kind: AuditEventKind,
    transaction_id: &str,
    receipt_digest: &str,
) -> Result<(), TransactionStoreError> {
    let prev: Option<(i64, String)> = sqlx_core::query_as::query_as(
        "SELECT seq, chain_hash FROM audit_events ORDER BY seq DESC LIMIT 1 FOR UPDATE",
    )
    .fetch_optional(&mut **tx)
    .await
    .map_err(map_sqlx_err)?;
    let (seq, prev_chain_hash) = match prev {
        Some((s, h)) => ((s as u64) + 1, h),
        None => (1, String::new()),
    };
    let created_at = now_iso();
    let key_id = CURRENT_KEY_ID.to_string();
    let chain_hash = key.event_hash(
        &EventContent {
            seq,
            key_id: &key_id,
            kind,
            transaction_id,
            receipt_digest,
            created_at: &created_at,
        },
        &prev_chain_hash,
    );
    sqlx_core::query::query(
        "INSERT INTO audit_events ( \
            seq, key_id, kind, transaction_id, receipt_digest, \
            created_at, chain_hash, prev_chain_hash \
         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
    )
    .bind(seq as i64)
    .bind(&key_id)
    .bind(kind.as_str())
    .bind(transaction_id)
    .bind(receipt_digest)
    .bind(&created_at)
    .bind(&chain_hash)
    .bind(&prev_chain_hash)
    .execute(&mut **tx)
    .await
    .map_err(map_sqlx_err)?;
    Ok(())
}

// ---------------------------------------------------------------------------
// Insert path — chain-aware
// ---------------------------------------------------------------------------

async fn insert_transaction(
    tx: &mut sqlx_core::transaction::Transaction<'_, sqlx_postgres::Postgres>,
    key: &AuditKey,
    transaction_id: &str,
    transaction: NewTransaction,
) -> Result<InsertedTransaction, TransactionStoreError> {
    let warnings_json = serialize(&transaction.warnings)?;
    let status = JobState::Queued;
    let created_at = now_iso();
    let key_id = CURRENT_KEY_ID.to_string();
    // Always the store's own commitment — see the matching note in
    // `transactions.rs`; a caller may not supply this value.
    let approval_id = Some(key.approval_commitment(transaction_id, &transaction.request_hash));

    // SELECT … FOR UPDATE on the most-recent row (if any) so concurrent
    // writers serialise on the chain tip. Without this, two parallel
    // record() calls could both compute seq=N+1 and then race on the
    // UNIQUE(seq) constraint — the loser would get a unique-violation
    // error.
    let prev: Option<(i64, String)> = sqlx_core::query_as::query_as(
        "SELECT seq, chain_hash FROM transactions ORDER BY seq DESC LIMIT 1 FOR UPDATE",
    )
    .fetch_optional(&mut **tx)
    .await
    .map_err(map_sqlx_err)?;
    let (seq, prev_chain_hash) = match prev {
        Some((s, h)) => ((s as u64) + 1, h),
        None => (1, String::new()),
    };

    // Bind the row to the approval-event chain tip as of this insert. Read
    // inside the same DB transaction as the seq allocation above.
    let event_tip: String = sqlx_core::query_scalar::query_scalar(
        "SELECT COALESCE((SELECT chain_hash FROM audit_events ORDER BY seq DESC LIMIT 1), '')",
    )
    .fetch_one(&mut **tx)
    .await
    .map_err(map_sqlx_err)?;
    let caller_role = transaction.caller_role.as_str();
    let caller_principal = transaction.caller_principal.as_signed_str();

    // As in the SQLite path: derive the stored version from the identity that was
    // signed, so the column cannot disagree with the message.
    let identity = ChainIdentity::V3 {
        caller_role,
        event_tip: &event_tip,
        caller_principal: &caller_principal,
    };
    let chain_hash = key.chain_hash(
        &ChainContent {
            seq,
            key_id: &key_id,
            transaction_id,
            request_id: &transaction.request_id,
            request_hash: &transaction.request_hash,
            action_name: &transaction.action_name,
            risk_level: transaction.risk_level,
            summary: &transaction.summary,
            approval_id: approval_id.as_deref(),
            warnings_json: &warnings_json,
            created_at: &created_at,
            identity,
        },
        &prev_chain_hash,
    );

    sqlx_core::query::query(
        "INSERT INTO transactions ( \
            transaction_id, request_id, request_hash, action_name, risk_level, \
            status, approval_id, summary, warnings_json, created_at, \
            seq, key_id, chain_hash, prev_chain_hash, \
            chain_version, caller_role, event_tip, caller_principal \
         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)",
    )
    .bind(transaction_id)
    .bind(&transaction.request_id)
    .bind(&transaction.request_hash)
    .bind(&transaction.action_name)
    .bind(serialize(&transaction.risk_level)?)
    .bind(serialize(&status)?)
    .bind(&approval_id)
    .bind(&transaction.summary)
    .bind(&warnings_json)
    .bind(&created_at)
    .bind(seq as i64)
    .bind(&key_id)
    .bind(&chain_hash)
    .bind(&prev_chain_hash)
    .bind(i64::from(identity.version()))
    .bind(caller_role)
    .bind(&event_tip)
    .bind(&caller_principal)
    .execute(&mut **tx)
    .await
    .map_err(map_sqlx_err)?;

    Ok(InsertedTransaction {
        record: TransactionRecord {
            transaction_id: transaction_id.to_string(),
            request_id: transaction.request_id,
            request_hash: transaction.request_hash,
            action_name: transaction.action_name,
            risk_level: transaction.risk_level,
            status,
            approval_id,
            summary: transaction.summary,
            warnings: transaction.warnings,
        },
        seq,
        chain_hash,
    })
}

// ---------------------------------------------------------------------------
// Row mappers
// ---------------------------------------------------------------------------

fn row_to_history_entry(
    row: sqlx_postgres::PgRow,
) -> Result<crate::transactions::JobHistoryEntry, TransactionStoreError> {
    Ok(crate::transactions::JobHistoryEntry {
        transaction_id: row.try_get("transaction_id").map_err(map_sqlx_err)?,
        action_name: row.try_get("action_name").map_err(map_sqlx_err)?,
        risk_level: deserialize(
            &row.try_get::<String, _>("risk_level")
                .map_err(map_sqlx_err)?,
        )?,
        status: deserialize(&row.try_get::<String, _>("status").map_err(map_sqlx_err)?)?,
        summary: row.try_get("summary").map_err(map_sqlx_err)?,
        created_at: row.try_get("created_at").map_err(map_sqlx_err)?,
    })
}

fn row_to_record(row: sqlx_postgres::PgRow) -> Result<TransactionRecord, TransactionStoreError> {
    Ok(TransactionRecord {
        transaction_id: row.try_get("transaction_id").map_err(map_sqlx_err)?,
        request_id: row.try_get("request_id").map_err(map_sqlx_err)?,
        request_hash: row.try_get("request_hash").map_err(map_sqlx_err)?,
        action_name: row.try_get("action_name").map_err(map_sqlx_err)?,
        risk_level: deserialize(
            &row.try_get::<String, _>("risk_level")
                .map_err(map_sqlx_err)?,
        )?,
        status: deserialize(&row.try_get::<String, _>("status").map_err(map_sqlx_err)?)?,
        approval_id: row.try_get("approval_id").map_err(map_sqlx_err)?,
        summary: row.try_get("summary").map_err(map_sqlx_err)?,
        warnings: deserialize(
            &row.try_get::<String, _>("warnings_json")
                .map_err(map_sqlx_err)?,
        )?,
    })
}

fn row_to_chain_row(row: sqlx_postgres::PgRow) -> Result<ChainRow, TransactionStoreError> {
    Ok(ChainRow {
        seq: row.try_get::<i64, _>("seq").map_err(map_sqlx_err)? as u64,
        key_id: row.try_get("key_id").map_err(map_sqlx_err)?,
        transaction_id: row.try_get("transaction_id").map_err(map_sqlx_err)?,
        request_id: row.try_get("request_id").map_err(map_sqlx_err)?,
        request_hash: row.try_get("request_hash").map_err(map_sqlx_err)?,
        action_name: row.try_get("action_name").map_err(map_sqlx_err)?,
        risk_level: deserialize(
            &row.try_get::<String, _>("risk_level")
                .map_err(map_sqlx_err)?,
        )?,
        summary: row.try_get("summary").map_err(map_sqlx_err)?,
        approval_id: row.try_get("approval_id").map_err(map_sqlx_err)?,
        warnings_json: row.try_get("warnings_json").map_err(map_sqlx_err)?,
        created_at: row.try_get("created_at").map_err(map_sqlx_err)?,
        prev_chain_hash: row.try_get("prev_chain_hash").map_err(map_sqlx_err)?,
        chain_hash: row.try_get("chain_hash").map_err(map_sqlx_err)?,
        chain_version: row
            .try_get::<i64, _>("chain_version")
            .map_err(map_sqlx_err)? as u32,
        caller_role: row.try_get("caller_role").map_err(map_sqlx_err)?,
        event_tip: row.try_get("event_tip").map_err(map_sqlx_err)?,
        caller_principal: row.try_get("caller_principal").map_err(map_sqlx_err)?,
    })
}

fn row_to_event_row(row: sqlx_postgres::PgRow) -> Result<EventRow, TransactionStoreError> {
    Ok(EventRow {
        seq: row.try_get::<i64, _>("seq").map_err(map_sqlx_err)? as u64,
        key_id: row.try_get("key_id").map_err(map_sqlx_err)?,
        kind: row.try_get("kind").map_err(map_sqlx_err)?,
        transaction_id: row.try_get("transaction_id").map_err(map_sqlx_err)?,
        receipt_digest: row.try_get("receipt_digest").map_err(map_sqlx_err)?,
        created_at: row.try_get("created_at").map_err(map_sqlx_err)?,
        prev_chain_hash: row.try_get("prev_chain_hash").map_err(map_sqlx_err)?,
        chain_hash: row.try_get("chain_hash").map_err(map_sqlx_err)?,
    })
}

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

    /// Verifies that PostgresConfig defaults are sensible and don't accidentally
    /// disable the statement cache.
    #[test]
    fn default_config_keeps_statement_cache_enabled() {
        let c = PostgresConfig::default();
        assert!(c.statement_cache_capacity > 0);
        assert!(c.max_connections >= 4);
        assert!(c.acquire_timeout >= Duration::from_secs(5));
    }

    #[test]
    fn url_parsing_accepts_standard_postgres_url() {
        let opts = PgConnectOptions::from_str(
            "postgres://user:pass@host.example.com:5432/audit?sslmode=require",
        )
        .expect("standard URL parses");
        // Ensure binding via PgPoolOptions wouldn't reject it (just smoke-check).
        let _ = opts.statement_cache_capacity(0);
    }

    #[test]
    fn now_iso_produces_a_parseable_rfc3339_timestamp() {
        // `now_iso()` is signed into `ChainContent.created_at` on every
        // audit row and is cast to `timestamptz` by approve, claim-for-
        // execution, the stale sweep, and the history filter. The format
        // string previously omitted `%S`, yielding `2026-07-27T12:34:.567Z`:
        // seconds silently vanished, the chain attested to a broken value
        // forever, and same-minute ordering became unrecoverable.
        let ts = now_iso();
        let parsed = chrono::DateTime::parse_from_rfc3339(&ts);
        assert!(
            parsed.is_ok(),
            "now_iso() must emit a parseable RFC 3339 timestamp, got {ts:?} ({parsed:?})"
        );
        assert_eq!(
            ts.len(),
            "2026-07-27T12:34:56.789Z".len(),
            "timestamp shape changed unexpectedly: {ts}"
        );
    }

    #[test]
    fn now_iso_matches_the_sqlite_backends_timestamp_shape() {
        // Both backends feed the same signed chain, so a row written by
        // Postgres must be byte-comparable with one written by SQLite
        // (`strftime('%Y-%m-%dT%H:%M:%fZ','now')`, which yields
        // milliseconds). Divergent shapes make cross-backend verification
        // and time-ordered queries subtly wrong.
        let ts = now_iso();
        let (date, rest) = ts.split_once('T').expect("timestamp has a T separator");
        assert_eq!(date.len(), "2026-07-27".len(), "date part: {ts}");
        assert!(rest.ends_with('Z'), "must be UTC with a Z suffix: {ts}");
        let time = rest.trim_end_matches('Z');
        let fields: Vec<&str> = time.split(':').collect();
        assert_eq!(fields.len(), 3, "expected HH:MM:SS.mmm, got {time:?}");
        assert!(
            fields[2].contains('.'),
            "seconds field must carry fractional milliseconds: {time:?}"
        );
        let (secs, millis) = fields[2].split_once('.').expect("checked above");
        assert_eq!(secs.len(), 2, "seconds must be two digits: {time:?}");
        assert_eq!(millis.len(), 3, "expected millisecond precision: {time:?}");
    }

    #[test]
    fn schema_decode_errors_are_classified_as_invariant_failures() {
        let error = map_sqlx_err(sqlx_core::Error::ColumnNotFound("chain_hash".to_string()));
        assert!(matches!(error, TransactionStoreError::DatabaseInvariant(_)));
    }
}