sysknife_daemon/store/
postgres.rs

1//! Postgres backend for the audit log.
2//!
3//! Wire-compatible with AWS RDS / Aurora, GCP Cloud SQL + AlloyDB, Azure
4//! Database for PostgreSQL Flexible Server, Supabase (direct + pooler),
5//! Neon (direct + pooler), and self-hosted Postgres.
6//!
7//! ## Connection lifecycle
8//!
9//! [`PostgresStore::connect`] takes a [`PostgresConfig`] (URL plus pool +
10//! statement-cache tuning) and returns a configured `sqlx::PgPool`. The
11//! same pool is used for every request — sqlx handles connection reuse,
12//! retry on broken connections, and TLS via rustls.
13//!
14//! On connect, `PostgresStore::initialize` runs ordered, transactional schema
15//! migrations under a database advisory lock. The schema mirrors SQLite
16//! field-for-field; the only dialect
17//! difference is `BIGINT NOT NULL UNIQUE` for `seq`. `created_at` remains
18//! `TEXT` (RFC 3339 with `Z` suffix) in both backends and is cast to
19//! `timestamptz` for interval arithmetic.
20//!
21//! ## Concurrency
22//!
23//! Like the SQLite path, the chain-hash + seq computation is wrapped in a
24//! `BEGIN ... COMMIT` block. Postgres serialises this naturally via
25//! row-level locks on the `seq` write; we use a `SELECT … FOR UPDATE` on
26//! the most-recent row so concurrent writers see the same predecessor and
27//! the second commit fails atomically rather than corrupting the chain.
28
29use async_trait::async_trait;
30use sqlx_core::row::Row;
31use sqlx_postgres::{PgConnectOptions, PgPool, PgPoolOptions};
32use std::str::FromStr;
33use std::time::Duration;
34use subtle::ConstantTimeEq;
35use sysknife_types::{JobState, PreviewEnvelope, TransactionRecord};
36use uuid::Uuid;
37
38use crate::audit_chain::{
39    AuditEventKind, AuditKey, AuditVerification, ChainContent, ChainIdentity, ChainRow,
40    EventContent, EventRow, VerifyOutcome, CHAIN_VERSION_CURRENT, CURRENT_KEY_ID,
41};
42use crate::audit_watermark::emit_chain_tip_watermark;
43use crate::store::AuditStore;
44use crate::transactions::{
45    NewTransaction, RecordedPreviewedTransaction, TransactionStoreError,
46    APPROVAL_RECEIPT_TTL_MINUTES,
47};
48
49const MIGRATION_LOCK_ID: i64 = 0x5359_534b_4e49_4645;
50
51/// Column list every `ChainRow` read shares, kept next to `row_to_chain_row`.
52///
53/// The two read paths each spelled the list out. Adding the caller-identity
54/// columns updated the mapper and one of the two queries, and the miss showed
55/// up only as a runtime "no column found for name: chain_version" from the
56/// live-Postgres test — the unit tests, which never touch this SQL, stayed
57/// green. Mirrors `CHAIN_ROW_COLUMNS` in `transactions.rs`.
58const CHAIN_ROW_COLUMNS: &str = "seq, key_id, transaction_id, request_id, request_hash, \
59     action_name, risk_level, summary, approval_id, warnings_json, \
60     created_at, prev_chain_hash, chain_hash, chain_version, caller_role, event_tip";
61
62struct Migration {
63    version: i64,
64    name: &'static str,
65    statements: &'static [&'static str],
66}
67
68const MIGRATIONS: &[Migration] = &[Migration {
69    version: 1,
70    name: "initial_audit_schema",
71    statements: &[
72        r#"
73        CREATE TABLE IF NOT EXISTS transactions (
74            transaction_id TEXT PRIMARY KEY,
75            request_id TEXT NOT NULL,
76            request_hash TEXT NOT NULL,
77            action_name TEXT NOT NULL,
78            risk_level TEXT NOT NULL,
79            status TEXT NOT NULL,
80            approval_id TEXT,
81            summary TEXT NOT NULL,
82            warnings_json TEXT NOT NULL,
83            created_at TEXT NOT NULL,
84            seq BIGINT NOT NULL UNIQUE,
85            key_id TEXT NOT NULL,
86            chain_hash TEXT NOT NULL,
87            prev_chain_hash TEXT NOT NULL DEFAULT ''
88        )
89        "#,
90        r#"
91        CREATE TABLE IF NOT EXISTS transaction_approvals (
92            transaction_id TEXT PRIMARY KEY,
93            receipt_digest TEXT NOT NULL,
94            approved_at TEXT NOT NULL,
95            consumed_at TEXT
96        )
97        "#,
98        r#"
99        CREATE TABLE IF NOT EXISTS transaction_previews (
100            transaction_id TEXT PRIMARY KEY,
101            preview_json TEXT NOT NULL
102        )
103        "#,
104        "CREATE INDEX IF NOT EXISTS transactions_seq_idx ON transactions(seq)",
105    ],
106},
107    // Mirrors SQLITE_MIGRATIONS v2 in `transactions.rs`. The columns are
108    // nullable with a legacy default because rows written by an earlier binary
109    // were signed over an encoding without them — see `ChainIdentity`.
110    Migration {
111        version: 2,
112        name: "caller_identity_and_approval_events",
113        statements: &[
114            "ALTER TABLE transactions ADD COLUMN IF NOT EXISTS chain_version BIGINT NOT NULL DEFAULT 1",
115            "ALTER TABLE transactions ADD COLUMN IF NOT EXISTS caller_role TEXT",
116            "ALTER TABLE transactions ADD COLUMN IF NOT EXISTS event_tip TEXT",
117            r#"
118            CREATE TABLE IF NOT EXISTS audit_events (
119                seq BIGINT PRIMARY KEY,
120                key_id TEXT NOT NULL,
121                kind TEXT NOT NULL,
122                transaction_id TEXT NOT NULL,
123                receipt_digest TEXT NOT NULL,
124                created_at TEXT NOT NULL,
125                chain_hash TEXT NOT NULL,
126                prev_chain_hash TEXT NOT NULL DEFAULT ''
127            )
128            "#,
129            "CREATE INDEX IF NOT EXISTS audit_events_transaction_idx ON audit_events(transaction_id)",
130        ],
131    },
132];
133
134/// Configuration for the Postgres backend. Built by `main.rs` from
135/// `[storage]` in `config.toml`.
136#[derive(Clone, Debug)]
137pub struct PostgresConfig {
138    /// `postgres://...` URL. SSL mode and other knobs are encoded here.
139    pub url: String,
140    /// Maximum pool size. 8 is comfortable for a single SysKnife daemon;
141    /// raise if multiple shells run concurrently.
142    pub max_connections: u32,
143    /// `acquire_timeout` for getting a connection from the pool. Raise above
144    /// the default 5s if you hit Neon cold starts (~600 ms first call).
145    pub acquire_timeout: Duration,
146    /// Set to 0 for transaction-mode PgBouncer (including some hosted
147    /// poolers), which cannot use sqlx's per-connection prepared-statement
148    /// cache. Default 100 covers direct Postgres connections.
149    pub statement_cache_capacity: usize,
150}
151
152impl Default for PostgresConfig {
153    fn default() -> Self {
154        Self {
155            url: String::new(),
156            max_connections: 8,
157            acquire_timeout: Duration::from_secs(10),
158            statement_cache_capacity: 100,
159        }
160    }
161}
162
163/// Postgres-backed [`AuditStore`].
164#[derive(Debug, Clone)]
165pub struct PostgresStore {
166    pool: PgPool,
167    audit_key: std::sync::Arc<AuditKey>,
168}
169
170struct InsertedTransaction {
171    record: TransactionRecord,
172    seq: u64,
173    chain_hash: String,
174}
175
176impl PostgresStore {
177    async fn connect_pool(config: &PostgresConfig) -> Result<PgPool, TransactionStoreError> {
178        let mut connect_opts = PgConnectOptions::from_str(&config.url).map_err(|e| {
179            TransactionStoreError::Io(std::io::Error::new(
180                std::io::ErrorKind::InvalidInput,
181                format!("invalid postgres URL: {e}"),
182            ))
183        })?;
184        connect_opts = connect_opts.statement_cache_capacity(config.statement_cache_capacity);
185        PgPoolOptions::new()
186            .max_connections(config.max_connections)
187            .acquire_timeout(config.acquire_timeout)
188            .connect_with(connect_opts)
189            .await
190            .map_err(map_sqlx_err)
191    }
192
193    /// Connect to the configured Postgres database, run the migration,
194    /// and bind an audit key for chain computation.
195    pub async fn connect(
196        config: &PostgresConfig,
197        audit_key: std::sync::Arc<AuditKey>,
198    ) -> Result<Self, TransactionStoreError> {
199        let pool = Self::connect_pool(config).await?;
200
201        let store = Self { pool, audit_key };
202        store.initialize().await?;
203        Ok(store)
204    }
205
206    /// Verify an existing Postgres chain — transaction chain, approval-event
207    /// chain, and the binding between them — with only the exported public
208    /// key. This path performs no migrations and never loads the signing key.
209    pub async fn verify_all_with_pubkey(
210        config: &PostgresConfig,
211        verifying_key_hex: &str,
212    ) -> Result<AuditVerification, TransactionStoreError> {
213        let pool = Self::connect_pool(config).await?;
214        let tx_rows = fetch_chain_rows_from_pool(&pool).await?;
215        // A chain written before the migration has no `audit_events` table.
216        // That is an absent feature, not a missing verification: report zero
217        // events rather than failing the whole verify.
218        let event_rows = fetch_event_rows_from_pool(&pool).await.unwrap_or_default();
219        Ok(crate::audit_chain::verify_all_with_pubkey(
220            verifying_key_hex,
221            &tx_rows,
222            &event_rows,
223        ))
224    }
225
226    /// All three checks with the daemon's key.
227    pub async fn verify_all(
228        &self,
229        key: &AuditKey,
230    ) -> Result<AuditVerification, TransactionStoreError> {
231        let tx_rows = fetch_chain_rows_from_pool(&self.pool).await?;
232        let event_rows = fetch_event_rows_from_pool(&self.pool).await?;
233        Ok(crate::audit_chain::verify_all(key, &tx_rows, &event_rows))
234    }
235
236    /// Apply pending schema migrations atomically. The advisory transaction
237    /// lock serializes concurrent daemon starts against the same database.
238    async fn initialize(&self) -> Result<(), TransactionStoreError> {
239        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
240        sqlx_core::query::query("SELECT pg_advisory_xact_lock($1)")
241            .bind(MIGRATION_LOCK_ID)
242            .execute(&mut *tx)
243            .await
244            .map_err(map_sqlx_err)?;
245
246        sqlx_core::query::query(
247            r#"
248            CREATE TABLE IF NOT EXISTS schema_migrations (
249                version BIGINT PRIMARY KEY,
250                name TEXT NOT NULL,
251                applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
252            )
253            "#,
254        )
255        .execute(&mut *tx)
256        .await
257        .map_err(map_sqlx_err)?;
258
259        let current: i64 = sqlx_core::query_scalar::query_scalar(
260            "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
261        )
262        .fetch_one(&mut *tx)
263        .await
264        .map_err(map_sqlx_err)?;
265        let latest = MIGRATIONS.last().map_or(0, |migration| migration.version);
266        if current > latest {
267            return Err(TransactionStoreError::DatabaseInvariant(format!(
268                "postgres schema version {current} is newer than this binary supports ({latest})"
269            )));
270        }
271
272        for migration in MIGRATIONS
273            .iter()
274            .filter(|migration| migration.version > current)
275        {
276            // Run statements separately for transaction-mode poolers that
277            // reject multi-statement query strings.
278            for statement in migration.statements {
279                sqlx_core::query::query(*statement)
280                    .execute(&mut *tx)
281                    .await
282                    .map_err(map_sqlx_err)?;
283            }
284            sqlx_core::query::query(
285                "INSERT INTO schema_migrations (version, name) VALUES ($1, $2)",
286            )
287            .bind(migration.version)
288            .bind(migration.name)
289            .execute(&mut *tx)
290            .await
291            .map_err(map_sqlx_err)?;
292        }
293
294        tx.commit().await.map_err(map_sqlx_err)
295    }
296}
297
298// ---------------------------------------------------------------------------
299// Helpers
300// ---------------------------------------------------------------------------
301
302fn map_sqlx_err(e: sqlx_core::Error) -> TransactionStoreError {
303    let message = format!("postgres: {e}");
304    let invariant = match &e {
305        sqlx_core::Error::Database(db) => db
306            .code()
307            .is_some_and(|code| code.starts_with("22") || code.starts_with("23")),
308        sqlx_core::Error::TypeNotFound { .. }
309        | sqlx_core::Error::ColumnIndexOutOfBounds { .. }
310        | sqlx_core::Error::ColumnNotFound(_)
311        | sqlx_core::Error::ColumnDecode { .. }
312        | sqlx_core::Error::Decode(_) => true,
313        _ => false,
314    };
315    if invariant {
316        TransactionStoreError::DatabaseInvariant(message)
317    } else {
318        TransactionStoreError::Io(std::io::Error::other(message))
319    }
320}
321
322fn serialize<T: serde::Serialize>(v: &T) -> Result<String, TransactionStoreError> {
323    serde_json::to_string(v).map_err(TransactionStoreError::Json)
324}
325
326fn deserialize<T: serde::de::DeserializeOwned>(s: &str) -> Result<T, TransactionStoreError> {
327    serde_json::from_str(s).map_err(TransactionStoreError::Json)
328}
329
330/// UTC timestamp in the same RFC 3339 millisecond shape the SQLite backend
331/// produces via `strftime('%Y-%m-%dT%H:%M:%fZ','now')`.
332///
333/// `%S` is load-bearing: chrono's `%.3f` renders only the fractional part, so
334/// omitting `%S` yields `2026-07-27T12:34:.567Z`. That value is signed into
335/// `ChainContent.created_at` on every audit row and cast to `timestamptz` by
336/// approve, claim-for-execution, the stale sweep, and the history filter.
337fn now_iso() -> String {
338    chrono::Utc::now()
339        .format("%Y-%m-%dT%H:%M:%S%.3fZ")
340        .to_string()
341}
342
343// ---------------------------------------------------------------------------
344// AuditStore impl
345// ---------------------------------------------------------------------------
346
347#[async_trait]
348impl AuditStore for PostgresStore {
349    async fn record(
350        &self,
351        transaction: NewTransaction,
352    ) -> Result<TransactionRecord, TransactionStoreError> {
353        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
354        let transaction_id = Uuid::new_v4().to_string();
355        let inserted =
356            insert_transaction(&mut tx, &self.audit_key, &transaction_id, transaction).await?;
357        tx.commit().await.map_err(map_sqlx_err)?;
358        emit_chain_tip_watermark(inserted.seq, &inserted.chain_hash);
359        Ok(inserted.record)
360    }
361
362    async fn record_previewed(
363        &self,
364        transaction: NewTransaction,
365        preview: PreviewEnvelope,
366    ) -> Result<RecordedPreviewedTransaction, TransactionStoreError> {
367        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
368        let transaction_id = Uuid::new_v4().to_string();
369        let inserted =
370            insert_transaction(&mut tx, &self.audit_key, &transaction_id, transaction).await?;
371
372        sqlx_core::query::query(
373            "INSERT INTO transaction_previews (transaction_id, preview_json) VALUES ($1, $2)",
374        )
375        .bind(&inserted.record.transaction_id)
376        .bind(serialize(&preview)?)
377        .execute(&mut *tx)
378        .await
379        .map_err(map_sqlx_err)?;
380
381        tx.commit().await.map_err(map_sqlx_err)?;
382        emit_chain_tip_watermark(inserted.seq, &inserted.chain_hash);
383        Ok(RecordedPreviewedTransaction {
384            transaction: inserted.record,
385            preview,
386        })
387    }
388
389    async fn get(
390        &self,
391        transaction_id: &str,
392    ) -> Result<Option<TransactionRecord>, TransactionStoreError> {
393        let row = sqlx_core::query::query(
394            "SELECT transaction_id, request_id, request_hash, action_name, risk_level, \
395                    status, approval_id, summary, warnings_json \
396             FROM transactions WHERE transaction_id = $1",
397        )
398        .bind(transaction_id)
399        .fetch_optional(&self.pool)
400        .await
401        .map_err(map_sqlx_err)?;
402        row.map(row_to_record).transpose()
403    }
404
405    async fn get_preview(
406        &self,
407        transaction_id: &str,
408    ) -> Result<Option<PreviewEnvelope>, TransactionStoreError> {
409        let row = sqlx_core::query::query(
410            "SELECT preview_json FROM transaction_previews WHERE transaction_id = $1",
411        )
412        .bind(transaction_id)
413        .fetch_optional(&self.pool)
414        .await
415        .map_err(map_sqlx_err)?;
416        match row {
417            Some(r) => {
418                let s: String = r.try_get("preview_json").map_err(map_sqlx_err)?;
419                Ok(Some(deserialize(&s)?))
420            }
421            None => Ok(None),
422        }
423    }
424
425    async fn update_status(
426        &self,
427        transaction_id: &str,
428        new_status: JobState,
429    ) -> Result<(), TransactionStoreError> {
430        // Read-validate-write with a state-machine guard, same shape as the
431        // SQLite path but strictly stronger: `SELECT ... FOR UPDATE` takes a
432        // row lock for the lifetime of this transaction, so no concurrent
433        // writer can change `status` between our read and our `UPDATE` — the
434        // read-modify-write here is atomic by construction. SQLite has no
435        // equivalent row-level lock, so its `update_status` instead uses an
436        // explicit compare-and-swap (`UPDATE ... WHERE status = <observed>`,
437        // checking `rows_affected`) to get the same guarantee.
438        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
439        let current_str: Option<String> = sqlx_core::query_scalar::query_scalar(
440            "SELECT status FROM transactions WHERE transaction_id = $1 FOR UPDATE",
441        )
442        .bind(transaction_id)
443        .fetch_optional(&mut *tx)
444        .await
445        .map_err(map_sqlx_err)?;
446        let current_str = current_str
447            .ok_or_else(|| TransactionStoreError::NotFound(transaction_id.to_string()))?;
448        let current: JobState = deserialize(&current_str)?;
449        if !crate::jobs::allowed_transition(&current, &new_status) {
450            return Err(TransactionStoreError::InvalidTransition {
451                from: current,
452                to: new_status,
453            });
454        }
455        sqlx_core::query::query("UPDATE transactions SET status = $1 WHERE transaction_id = $2")
456            .bind(serialize(&new_status)?)
457            .bind(transaction_id)
458            .execute(&mut *tx)
459            .await
460            .map_err(map_sqlx_err)?;
461        tx.commit().await.map_err(map_sqlx_err)?;
462        Ok(())
463    }
464
465    async fn approve_transaction(
466        &self,
467        transaction_id: &str,
468    ) -> Result<Option<String>, TransactionStoreError> {
469        let row = sqlx_core::query::query(
470            "SELECT request_hash, approval_id FROM transactions WHERE transaction_id = $1",
471        )
472        .bind(transaction_id)
473        .fetch_optional(&self.pool)
474        .await
475        .map_err(map_sqlx_err)?;
476        let Some(row) = row else {
477            return Ok(None);
478        };
479        let request_hash: String = row.try_get("request_hash").map_err(map_sqlx_err)?;
480        let committed_digest: Option<String> = row.try_get("approval_id").map_err(map_sqlx_err)?;
481        let receipt = self
482            .audit_key
483            .approval_receipt(transaction_id, &request_hash);
484        let receipt_digest = crate::audit_chain::approval_receipt_digest(&receipt);
485        let commitment_matches = committed_digest.as_deref().is_some_and(|committed| {
486            bool::from(receipt_digest.as_bytes().ct_eq(committed.as_bytes()))
487        });
488        if !commitment_matches {
489            return Err(TransactionStoreError::DatabaseInvariant(format!(
490                "transaction {transaction_id} approval commitment does not match its signed preview"
491            )));
492        }
493        let queued = serialize(&JobState::Queued)?;
494        // Approval row and chained event commit together — an approval absent
495        // from the event chain is exactly what that chain exists to prevent.
496        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
497        let result = sqlx_core::query::query(
498            sqlx_core::sql_str::AssertSqlSafe(format!(
499                "INSERT INTO transaction_approvals \
500                     (transaction_id, receipt_digest, approved_at) \
501                 SELECT transaction_id, $1, $2 FROM transactions \
502                 WHERE transaction_id = $3 \
503                   AND status = $4 \
504                   AND created_at::timestamptz > now() - INTERVAL '{APPROVAL_RECEIPT_TTL_MINUTES} minutes' \
505                 ON CONFLICT (transaction_id) DO NOTHING"
506            )),
507        )
508        .bind(&receipt_digest)
509        .bind(now_iso())
510        .bind(transaction_id)
511        .bind(&queued)
512        .execute(&mut *tx)
513        .await
514        .map_err(map_sqlx_err)?;
515        if result.rows_affected() > 0 {
516            append_event(
517                &mut tx,
518                &self.audit_key,
519                AuditEventKind::ApprovalGranted,
520                transaction_id,
521                &receipt_digest,
522            )
523            .await?;
524        }
525        tx.commit().await.map_err(map_sqlx_err)?;
526        Ok((result.rows_affected() > 0).then_some(receipt))
527    }
528
529    async fn revoke_unconsumed_approval(
530        &self,
531        transaction_id: &str,
532    ) -> Result<bool, TransactionStoreError> {
533        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
534        // Read the digest before the DELETE: the event names which receipt was
535        // retracted, and after the delete there is nothing left to name.
536        let digest: Option<String> = sqlx_core::query_scalar::query_scalar(
537            "SELECT receipt_digest FROM transaction_approvals \
538             WHERE transaction_id = $1 AND consumed_at IS NULL",
539        )
540        .bind(transaction_id)
541        .fetch_optional(&mut *tx)
542        .await
543        .map_err(map_sqlx_err)?;
544        let result = sqlx_core::query::query(
545            "DELETE FROM transaction_approvals \
546             WHERE transaction_id = $1 AND consumed_at IS NULL",
547        )
548        .bind(transaction_id)
549        .execute(&mut *tx)
550        .await
551        .map_err(map_sqlx_err)?;
552        if result.rows_affected() > 0 {
553            let digest = digest.ok_or_else(|| {
554                TransactionStoreError::DatabaseInvariant(format!(
555                    "revoked an approval for {transaction_id} that had no receipt digest"
556                ))
557            })?;
558            append_event(
559                &mut tx,
560                &self.audit_key,
561                AuditEventKind::ApprovalRevoked,
562                transaction_id,
563                &digest,
564            )
565            .await?;
566        }
567        tx.commit().await.map_err(map_sqlx_err)?;
568        Ok(result.rows_affected() > 0)
569    }
570
571    async fn claim_approved_for_execution(
572        &self,
573        transaction_id: &str,
574        receipt_digest: &str,
575    ) -> Result<bool, TransactionStoreError> {
576        let queued = serialize(&JobState::Queued)?;
577        let running = serialize(&JobState::Running)?;
578        let mut tx = self.pool.begin().await.map_err(map_sqlx_err)?;
579        let result = sqlx_core::query::query(
580            sqlx_core::sql_str::AssertSqlSafe(format!(
581                "UPDATE transactions SET status = $1 \
582                 WHERE transaction_id = $2 \
583                   AND status = $3 \
584                   AND created_at::timestamptz > now() - INTERVAL '{APPROVAL_RECEIPT_TTL_MINUTES} minutes' \
585                   AND EXISTS ( \
586                       SELECT 1 FROM transaction_approvals \
587                       WHERE transaction_id = $2 \
588                         AND receipt_digest = $4 \
589                         AND consumed_at IS NULL \
590                   )"
591            )),
592        )
593        .bind(&running)
594        .bind(transaction_id)
595        .bind(&queued)
596        .bind(receipt_digest)
597        .execute(&mut *tx)
598        .await
599        .map_err(map_sqlx_err)?;
600        if result.rows_affected() > 0 {
601            sqlx_core::query::query(
602                "UPDATE transaction_approvals SET consumed_at = $1 \
603                 WHERE transaction_id = $2 AND consumed_at IS NULL",
604            )
605            .bind(now_iso())
606            .bind(transaction_id)
607            .execute(&mut *tx)
608            .await
609            .map_err(map_sqlx_err)?;
610            append_event(
611                &mut tx,
612                &self.audit_key,
613                AuditEventKind::ApprovalConsumed,
614                transaction_id,
615                receipt_digest,
616            )
617            .await?;
618        }
619        tx.commit().await.map_err(map_sqlx_err)?;
620        Ok(result.rows_affected() > 0)
621    }
622
623    async fn cleanup_stale_queued(&self) -> Result<u64, TransactionStoreError> {
624        let queued = serialize(&JobState::Queued)?;
625        let canceled = serialize(&JobState::Canceled)?;
626        let result = sqlx_core::query::query(
627            sqlx_core::sql_str::AssertSqlSafe(format!(
628                "UPDATE transactions SET status = $1 \
629                 WHERE status = $2 \
630                   AND created_at::timestamptz <= now() - INTERVAL '{APPROVAL_RECEIPT_TTL_MINUTES} minutes'"
631            )),
632        )
633        .bind(&canceled)
634        .bind(&queued)
635        .execute(&self.pool)
636        .await
637        .map_err(map_sqlx_err)?;
638        Ok(result.rows_affected())
639    }
640
641    async fn cancel_queued(&self, transaction_id: &str) -> Result<bool, TransactionStoreError> {
642        // Option A: the `status = $3` (Queued) guard means a Running (in-flight)
643        // transaction is never cancelled.
644        let queued = serialize(&JobState::Queued)?;
645        let canceled = serialize(&JobState::Canceled)?;
646        let result = sqlx_core::query::query(
647            "UPDATE transactions SET status = $1 \
648             WHERE transaction_id = $2 AND status = $3",
649        )
650        .bind(&canceled)
651        .bind(transaction_id)
652        .bind(&queued)
653        .execute(&self.pool)
654        .await
655        .map_err(map_sqlx_err)?;
656        Ok(result.rows_affected() > 0)
657    }
658
659    async fn list_transactions(
660        &self,
661        limit: u32,
662        status_filter: Option<&str>,
663        action_filter: Option<&str>,
664        since_hours: Option<u32>,
665    ) -> Result<Vec<TransactionRecord>, TransactionStoreError> {
666        let limit = limit.min(100) as i64;
667
668        // Validate the status filter against the JobState enum so a typo
669        // produces an error instead of silent empty results.
670        let validated_status: Option<String> = status_filter
671            .map(|s| -> Result<String, TransactionStoreError> {
672                let parsed: JobState = deserialize(&format!("\"{s}\""))?;
673                serialize(&parsed)
674            })
675            .transpose()?;
676
677        let mut sql = String::from(
678            "SELECT transaction_id, request_id, request_hash, action_name, risk_level, \
679                    status, approval_id, summary, warnings_json \
680             FROM transactions WHERE TRUE",
681        );
682        let mut idx = 1;
683        if validated_status.is_some() {
684            sql.push_str(&format!(" AND status = ${idx}"));
685            idx += 1;
686        }
687        if action_filter.is_some() {
688            sql.push_str(&format!(" AND action_name = ${idx}"));
689            idx += 1;
690        }
691        if since_hours.is_some() {
692            sql.push_str(&format!(
693                " AND created_at::timestamptz > now() - (${idx} || ' hours')::INTERVAL"
694            ));
695            idx += 1;
696        }
697        sql.push_str(&format!(" ORDER BY seq DESC LIMIT ${idx}"));
698
699        let mut q = sqlx_core::query::query(sqlx_core::sql_str::AssertSqlSafe(sql.as_str()));
700        if let Some(s) = &validated_status {
701            q = q.bind(s);
702        }
703        if let Some(a) = action_filter {
704            q = q.bind(a);
705        }
706        if let Some(h) = since_hours {
707            q = q.bind(h.to_string());
708        }
709        q = q.bind(limit);
710
711        let rows = q.fetch_all(&self.pool).await.map_err(map_sqlx_err)?;
712        rows.into_iter().map(row_to_record).collect()
713    }
714
715    async fn list_history(
716        &self,
717        limit: u32,
718        status_filter: Option<&str>,
719        action_filter: Option<&str>,
720        since_hours: Option<u32>,
721    ) -> Result<Vec<crate::transactions::JobHistoryEntry>, TransactionStoreError> {
722        // Parallels list_transactions but selects created_at and returns the
723        // structured JobHistoryEntry DTO. Same filter/validation semantics.
724        let limit = limit.min(100) as i64;
725        let validated_status: Option<String> = status_filter
726            .map(|s| -> Result<String, TransactionStoreError> {
727                let parsed: JobState = deserialize(&format!("\"{s}\""))?;
728                serialize(&parsed)
729            })
730            .transpose()?;
731
732        let mut sql = String::from(
733            "SELECT transaction_id, action_name, risk_level, status, summary, created_at \
734             FROM transactions WHERE TRUE",
735        );
736        let mut idx = 1;
737        if validated_status.is_some() {
738            sql.push_str(&format!(" AND status = ${idx}"));
739            idx += 1;
740        }
741        if action_filter.is_some() {
742            sql.push_str(&format!(" AND action_name = ${idx}"));
743            idx += 1;
744        }
745        if since_hours.is_some() {
746            sql.push_str(&format!(
747                " AND created_at::timestamptz > now() - (${idx} || ' hours')::INTERVAL"
748            ));
749            idx += 1;
750        }
751        sql.push_str(&format!(" ORDER BY seq DESC LIMIT ${idx}"));
752
753        let mut q = sqlx_core::query::query(sqlx_core::sql_str::AssertSqlSafe(sql.as_str()));
754        if let Some(s) = &validated_status {
755            q = q.bind(s);
756        }
757        if let Some(a) = action_filter {
758            q = q.bind(a);
759        }
760        if let Some(h) = since_hours {
761            q = q.bind(h.to_string());
762        }
763        q = q.bind(limit);
764
765        let rows = q.fetch_all(&self.pool).await.map_err(map_sqlx_err)?;
766        rows.into_iter().map(row_to_history_entry).collect()
767    }
768
769    async fn fetch_chain_row(
770        &self,
771        transaction_id: &str,
772    ) -> Result<Option<ChainRow>, TransactionStoreError> {
773        let row = sqlx_core::query::query(sqlx_core::sql_str::AssertSqlSafe(format!(
774            "SELECT {CHAIN_ROW_COLUMNS} FROM transactions WHERE transaction_id = $1"
775        )))
776        .bind(transaction_id)
777        .fetch_optional(&self.pool)
778        .await
779        .map_err(map_sqlx_err)?;
780        row.map(row_to_chain_row).transpose()
781    }
782
783    async fn fetch_event_rows(&self) -> Result<Vec<EventRow>, TransactionStoreError> {
784        fetch_event_rows_from_pool(&self.pool).await
785    }
786
787    async fn fetch_chain_rows(&self) -> Result<Vec<ChainRow>, TransactionStoreError> {
788        fetch_chain_rows_from_pool(&self.pool).await
789    }
790
791    async fn verify_audit_chain(
792        &self,
793        key: &AuditKey,
794    ) -> Result<VerifyOutcome, TransactionStoreError> {
795        let rows = self.fetch_chain_rows().await?;
796        Ok(crate::audit_chain::verify_chain(key, &rows))
797    }
798}
799
800async fn fetch_chain_rows_from_pool(pool: &PgPool) -> Result<Vec<ChainRow>, TransactionStoreError> {
801    let rows = sqlx_core::query::query(sqlx_core::sql_str::AssertSqlSafe(format!(
802        "SELECT {CHAIN_ROW_COLUMNS} FROM transactions ORDER BY seq ASC"
803    )))
804    .fetch_all(pool)
805    .await
806    .map_err(map_sqlx_err)?;
807    rows.into_iter().map(row_to_chain_row).collect()
808}
809
810/// Fetch every approval event in seq order.
811async fn fetch_event_rows_from_pool(pool: &PgPool) -> Result<Vec<EventRow>, TransactionStoreError> {
812    let rows = sqlx_core::query::query(
813        "SELECT seq, key_id, kind, transaction_id, receipt_digest, \
814                created_at, prev_chain_hash, chain_hash \
815         FROM audit_events ORDER BY seq ASC",
816    )
817    .fetch_all(pool)
818    .await
819    .map_err(map_sqlx_err)?;
820    rows.into_iter().map(row_to_event_row).collect()
821}
822
823/// Append one signed approval event inside the caller's transaction.
824///
825/// `FOR UPDATE` on the current tip serialises concurrent appends the same way
826/// `insert_transaction` serialises on the transaction chain tip; without it two
827/// approvals could compute the same `seq` and collide on the primary key.
828async fn append_event(
829    tx: &mut sqlx_core::transaction::Transaction<'_, sqlx_postgres::Postgres>,
830    key: &AuditKey,
831    kind: AuditEventKind,
832    transaction_id: &str,
833    receipt_digest: &str,
834) -> Result<(), TransactionStoreError> {
835    let prev: Option<(i64, String)> = sqlx_core::query_as::query_as(
836        "SELECT seq, chain_hash FROM audit_events ORDER BY seq DESC LIMIT 1 FOR UPDATE",
837    )
838    .fetch_optional(&mut **tx)
839    .await
840    .map_err(map_sqlx_err)?;
841    let (seq, prev_chain_hash) = match prev {
842        Some((s, h)) => ((s as u64) + 1, h),
843        None => (1, String::new()),
844    };
845    let created_at = now_iso();
846    let key_id = CURRENT_KEY_ID.to_string();
847    let chain_hash = key.event_hash(
848        &EventContent {
849            seq,
850            key_id: &key_id,
851            kind,
852            transaction_id,
853            receipt_digest,
854            created_at: &created_at,
855        },
856        &prev_chain_hash,
857    );
858    sqlx_core::query::query(
859        "INSERT INTO audit_events ( \
860            seq, key_id, kind, transaction_id, receipt_digest, \
861            created_at, chain_hash, prev_chain_hash \
862         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
863    )
864    .bind(seq as i64)
865    .bind(&key_id)
866    .bind(kind.as_str())
867    .bind(transaction_id)
868    .bind(receipt_digest)
869    .bind(&created_at)
870    .bind(&chain_hash)
871    .bind(&prev_chain_hash)
872    .execute(&mut **tx)
873    .await
874    .map_err(map_sqlx_err)?;
875    Ok(())
876}
877
878// ---------------------------------------------------------------------------
879// Insert path — chain-aware
880// ---------------------------------------------------------------------------
881
882async fn insert_transaction(
883    tx: &mut sqlx_core::transaction::Transaction<'_, sqlx_postgres::Postgres>,
884    key: &AuditKey,
885    transaction_id: &str,
886    transaction: NewTransaction,
887) -> Result<InsertedTransaction, TransactionStoreError> {
888    let warnings_json = serialize(&transaction.warnings)?;
889    let status = JobState::Queued;
890    let created_at = now_iso();
891    let key_id = CURRENT_KEY_ID.to_string();
892    // Always the store's own commitment — see the matching note in
893    // `transactions.rs`; a caller may not supply this value.
894    let approval_id = Some(key.approval_commitment(transaction_id, &transaction.request_hash));
895
896    // SELECT … FOR UPDATE on the most-recent row (if any) so concurrent
897    // writers serialise on the chain tip. Without this, two parallel
898    // record() calls could both compute seq=N+1 and then race on the
899    // UNIQUE(seq) constraint — the loser would get a unique-violation
900    // error.
901    let prev: Option<(i64, String)> = sqlx_core::query_as::query_as(
902        "SELECT seq, chain_hash FROM transactions ORDER BY seq DESC LIMIT 1 FOR UPDATE",
903    )
904    .fetch_optional(&mut **tx)
905    .await
906    .map_err(map_sqlx_err)?;
907    let (seq, prev_chain_hash) = match prev {
908        Some((s, h)) => ((s as u64) + 1, h),
909        None => (1, String::new()),
910    };
911
912    // Bind the row to the approval-event chain tip as of this insert. Read
913    // inside the same DB transaction as the seq allocation above.
914    let event_tip: String = sqlx_core::query_scalar::query_scalar(
915        "SELECT COALESCE((SELECT chain_hash FROM audit_events ORDER BY seq DESC LIMIT 1), '')",
916    )
917    .fetch_one(&mut **tx)
918    .await
919    .map_err(map_sqlx_err)?;
920    let caller_role = transaction.caller_role.as_str();
921
922    let chain_hash = key.chain_hash(
923        &ChainContent {
924            seq,
925            key_id: &key_id,
926            transaction_id,
927            request_id: &transaction.request_id,
928            request_hash: &transaction.request_hash,
929            action_name: &transaction.action_name,
930            risk_level: transaction.risk_level,
931            summary: &transaction.summary,
932            approval_id: approval_id.as_deref(),
933            warnings_json: &warnings_json,
934            created_at: &created_at,
935            identity: ChainIdentity::V2 {
936                caller_role,
937                event_tip: &event_tip,
938            },
939        },
940        &prev_chain_hash,
941    );
942
943    sqlx_core::query::query(
944        "INSERT INTO transactions ( \
945            transaction_id, request_id, request_hash, action_name, risk_level, \
946            status, approval_id, summary, warnings_json, created_at, \
947            seq, key_id, chain_hash, prev_chain_hash, \
948            chain_version, caller_role, event_tip \
949         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)",
950    )
951    .bind(transaction_id)
952    .bind(&transaction.request_id)
953    .bind(&transaction.request_hash)
954    .bind(&transaction.action_name)
955    .bind(serialize(&transaction.risk_level)?)
956    .bind(serialize(&status)?)
957    .bind(&approval_id)
958    .bind(&transaction.summary)
959    .bind(&warnings_json)
960    .bind(&created_at)
961    .bind(seq as i64)
962    .bind(&key_id)
963    .bind(&chain_hash)
964    .bind(&prev_chain_hash)
965    .bind(i64::from(CHAIN_VERSION_CURRENT))
966    .bind(caller_role)
967    .bind(&event_tip)
968    .execute(&mut **tx)
969    .await
970    .map_err(map_sqlx_err)?;
971
972    Ok(InsertedTransaction {
973        record: TransactionRecord {
974            transaction_id: transaction_id.to_string(),
975            request_id: transaction.request_id,
976            request_hash: transaction.request_hash,
977            action_name: transaction.action_name,
978            risk_level: transaction.risk_level,
979            status,
980            approval_id,
981            summary: transaction.summary,
982            warnings: transaction.warnings,
983        },
984        seq,
985        chain_hash,
986    })
987}
988
989// ---------------------------------------------------------------------------
990// Row mappers
991// ---------------------------------------------------------------------------
992
993fn row_to_history_entry(
994    row: sqlx_postgres::PgRow,
995) -> Result<crate::transactions::JobHistoryEntry, TransactionStoreError> {
996    Ok(crate::transactions::JobHistoryEntry {
997        transaction_id: row.try_get("transaction_id").map_err(map_sqlx_err)?,
998        action_name: row.try_get("action_name").map_err(map_sqlx_err)?,
999        risk_level: deserialize(
1000            &row.try_get::<String, _>("risk_level")
1001                .map_err(map_sqlx_err)?,
1002        )?,
1003        status: deserialize(&row.try_get::<String, _>("status").map_err(map_sqlx_err)?)?,
1004        summary: row.try_get("summary").map_err(map_sqlx_err)?,
1005        created_at: row.try_get("created_at").map_err(map_sqlx_err)?,
1006    })
1007}
1008
1009fn row_to_record(row: sqlx_postgres::PgRow) -> Result<TransactionRecord, TransactionStoreError> {
1010    Ok(TransactionRecord {
1011        transaction_id: row.try_get("transaction_id").map_err(map_sqlx_err)?,
1012        request_id: row.try_get("request_id").map_err(map_sqlx_err)?,
1013        request_hash: row.try_get("request_hash").map_err(map_sqlx_err)?,
1014        action_name: row.try_get("action_name").map_err(map_sqlx_err)?,
1015        risk_level: deserialize(
1016            &row.try_get::<String, _>("risk_level")
1017                .map_err(map_sqlx_err)?,
1018        )?,
1019        status: deserialize(&row.try_get::<String, _>("status").map_err(map_sqlx_err)?)?,
1020        approval_id: row.try_get("approval_id").map_err(map_sqlx_err)?,
1021        summary: row.try_get("summary").map_err(map_sqlx_err)?,
1022        warnings: deserialize(
1023            &row.try_get::<String, _>("warnings_json")
1024                .map_err(map_sqlx_err)?,
1025        )?,
1026    })
1027}
1028
1029fn row_to_chain_row(row: sqlx_postgres::PgRow) -> Result<ChainRow, TransactionStoreError> {
1030    Ok(ChainRow {
1031        seq: row.try_get::<i64, _>("seq").map_err(map_sqlx_err)? as u64,
1032        key_id: row.try_get("key_id").map_err(map_sqlx_err)?,
1033        transaction_id: row.try_get("transaction_id").map_err(map_sqlx_err)?,
1034        request_id: row.try_get("request_id").map_err(map_sqlx_err)?,
1035        request_hash: row.try_get("request_hash").map_err(map_sqlx_err)?,
1036        action_name: row.try_get("action_name").map_err(map_sqlx_err)?,
1037        risk_level: deserialize(
1038            &row.try_get::<String, _>("risk_level")
1039                .map_err(map_sqlx_err)?,
1040        )?,
1041        summary: row.try_get("summary").map_err(map_sqlx_err)?,
1042        approval_id: row.try_get("approval_id").map_err(map_sqlx_err)?,
1043        warnings_json: row.try_get("warnings_json").map_err(map_sqlx_err)?,
1044        created_at: row.try_get("created_at").map_err(map_sqlx_err)?,
1045        prev_chain_hash: row.try_get("prev_chain_hash").map_err(map_sqlx_err)?,
1046        chain_hash: row.try_get("chain_hash").map_err(map_sqlx_err)?,
1047        chain_version: row
1048            .try_get::<i64, _>("chain_version")
1049            .map_err(map_sqlx_err)? as u32,
1050        caller_role: row.try_get("caller_role").map_err(map_sqlx_err)?,
1051        event_tip: row.try_get("event_tip").map_err(map_sqlx_err)?,
1052    })
1053}
1054
1055fn row_to_event_row(row: sqlx_postgres::PgRow) -> Result<EventRow, TransactionStoreError> {
1056    Ok(EventRow {
1057        seq: row.try_get::<i64, _>("seq").map_err(map_sqlx_err)? as u64,
1058        key_id: row.try_get("key_id").map_err(map_sqlx_err)?,
1059        kind: row.try_get("kind").map_err(map_sqlx_err)?,
1060        transaction_id: row.try_get("transaction_id").map_err(map_sqlx_err)?,
1061        receipt_digest: row.try_get("receipt_digest").map_err(map_sqlx_err)?,
1062        created_at: row.try_get("created_at").map_err(map_sqlx_err)?,
1063        prev_chain_hash: row.try_get("prev_chain_hash").map_err(map_sqlx_err)?,
1064        chain_hash: row.try_get("chain_hash").map_err(map_sqlx_err)?,
1065    })
1066}
1067
1068#[cfg(test)]
1069mod tests {
1070    use super::*;
1071
1072    /// Verifies that PostgresConfig defaults are sensible and don't accidentally
1073    /// disable the statement cache.
1074    #[test]
1075    fn default_config_keeps_statement_cache_enabled() {
1076        let c = PostgresConfig::default();
1077        assert!(c.statement_cache_capacity > 0);
1078        assert!(c.max_connections >= 4);
1079        assert!(c.acquire_timeout >= Duration::from_secs(5));
1080    }
1081
1082    #[test]
1083    fn url_parsing_accepts_standard_postgres_url() {
1084        let opts = PgConnectOptions::from_str(
1085            "postgres://user:pass@host.example.com:5432/audit?sslmode=require",
1086        )
1087        .expect("standard URL parses");
1088        // Ensure binding via PgPoolOptions wouldn't reject it (just smoke-check).
1089        let _ = opts.statement_cache_capacity(0);
1090    }
1091
1092    #[test]
1093    fn now_iso_produces_a_parseable_rfc3339_timestamp() {
1094        // `now_iso()` is signed into `ChainContent.created_at` on every
1095        // audit row and is cast to `timestamptz` by approve, claim-for-
1096        // execution, the stale sweep, and the history filter. The format
1097        // string previously omitted `%S`, yielding `2026-07-27T12:34:.567Z`:
1098        // seconds silently vanished, the chain attested to a broken value
1099        // forever, and same-minute ordering became unrecoverable.
1100        let ts = now_iso();
1101        let parsed = chrono::DateTime::parse_from_rfc3339(&ts);
1102        assert!(
1103            parsed.is_ok(),
1104            "now_iso() must emit a parseable RFC 3339 timestamp, got {ts:?} ({parsed:?})"
1105        );
1106        assert_eq!(
1107            ts.len(),
1108            "2026-07-27T12:34:56.789Z".len(),
1109            "timestamp shape changed unexpectedly: {ts}"
1110        );
1111    }
1112
1113    #[test]
1114    fn now_iso_matches_the_sqlite_backends_timestamp_shape() {
1115        // Both backends feed the same signed chain, so a row written by
1116        // Postgres must be byte-comparable with one written by SQLite
1117        // (`strftime('%Y-%m-%dT%H:%M:%fZ','now')`, which yields
1118        // milliseconds). Divergent shapes make cross-backend verification
1119        // and time-ordered queries subtly wrong.
1120        let ts = now_iso();
1121        let (date, rest) = ts.split_once('T').expect("timestamp has a T separator");
1122        assert_eq!(date.len(), "2026-07-27".len(), "date part: {ts}");
1123        assert!(rest.ends_with('Z'), "must be UTC with a Z suffix: {ts}");
1124        let time = rest.trim_end_matches('Z');
1125        let fields: Vec<&str> = time.split(':').collect();
1126        assert_eq!(fields.len(), 3, "expected HH:MM:SS.mmm, got {time:?}");
1127        assert!(
1128            fields[2].contains('.'),
1129            "seconds field must carry fractional milliseconds: {time:?}"
1130        );
1131        let (secs, millis) = fields[2].split_once('.').expect("checked above");
1132        assert_eq!(secs.len(), 2, "seconds must be two digits: {time:?}");
1133        assert_eq!(millis.len(), 3, "expected millisecond precision: {time:?}");
1134    }
1135
1136    #[test]
1137    fn schema_decode_errors_are_classified_as_invariant_failures() {
1138        let error = map_sqlx_err(sqlx_core::Error::ColumnNotFound("chain_hash".to_string()));
1139        assert!(matches!(error, TransactionStoreError::DatabaseInvariant(_)));
1140    }
1141}
sysknife_daemon/store/postgres.rs

sysknife_daemon/store/
postgres.rs
