gatekeep_sqlx/audit/
postgres.rs

1use async_trait::async_trait;
2use gatekeep::AuditEntry;
3use sqlx::Transaction;
4
5use super::support::{
6    deny_shape_label, effect_label, position_i32, presence_label, records_from_json_rows,
7};
8use super::{DecisionAuditRecord, SqlxAuditError, SqlxAuditStore, SqlxDecisionAuditRepository};
9
10/// Postgres-backed decision audit repository.
11pub type PgDecisionAuditRepository = SqlxDecisionAuditRepository<crate::PostgresBackend>;
12
13impl PgDecisionAuditRepository {
14    /// Creates a repository from a Postgres pool.
15    #[must_use]
16    pub const fn new(pool: sqlx::PgPool) -> Self {
17        Self::from_pool(pool)
18    }
19}
20
21#[async_trait]
22impl SqlxAuditStore<crate::PostgresBackend> for PgDecisionAuditRepository {
23    async fn record_decision_audit(&self, entry: &AuditEntry) -> Result<i64, SqlxAuditError> {
24        let mut tx = self.pool.begin().await?;
25        let denial_reason_json = entry
26            .denial_reason
27            .as_ref()
28            .map(serde_json::to_value)
29            .transpose()?;
30        let id = sqlx::query_scalar::<_, i64>(
31            r"
32            insert into gatekeep_audit_decisions
33              (request_id, policy_id, policy_hash, effect, trace, decisive_clause,
34               denial_reason_code, denial_reason_shape, denial_reason, entry)
35            values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
36            returning id
37            ",
38        )
39        .bind(entry.request_id.as_ref().map(gatekeep::RequestId::as_str))
40        .bind(entry.anchor.policy_id.as_str())
41        .bind(entry.anchor.policy_hash.as_str())
42        .bind(effect_label(entry))
43        .bind(serde_json::to_value(&entry.trace)?)
44        .bind(serde_json::to_value(&entry.decisive)?)
45        .bind(
46            entry
47                .denial_reason
48                .as_ref()
49                .map(|reason| reason.code.as_str()),
50        )
51        .bind(
52            entry
53                .denial_reason
54                .as_ref()
55                .map(|reason| deny_shape_label(reason.shape)),
56        )
57        .bind(denial_reason_json)
58        .bind(serde_json::to_value(entry)?)
59        .fetch_one(&mut *tx)
60        .await?;
61        insert_children(&mut tx, id, entry).await?;
62        insert_outbox(&mut tx, id, entry).await?;
63        tx.commit().await?;
64        Ok(id)
65    }
66
67    async fn decision_audit_records(
68        &self,
69        after_id: Option<i64>,
70        limit: i64,
71    ) -> Result<Vec<DecisionAuditRecord>, SqlxAuditError> {
72        let rows = sqlx::query(
73            "select id, entry from gatekeep_audit_decisions where ($1::bigint is null or id > $1) order by id limit $2",
74        )
75        .bind(after_id)
76        .bind(limit)
77        .fetch_all(&self.pool)
78        .await?;
79        records_from_json_rows(rows)
80    }
81}
82
83async fn insert_children(
84    tx: &mut Transaction<'_, sqlx::Postgres>,
85    decision_id: i64,
86    entry: &AuditEntry,
87) -> Result<(), SqlxAuditError> {
88    for (position, obligation) in entry.obligations.iter().enumerate() {
89        sqlx::query(
90            "insert into gatekeep_audit_obligations (decision_id, position, obligation_id) values ($1, $2, $3)",
91        )
92        .bind(decision_id)
93        .bind(position_i32(position))
94        .bind(obligation.as_str())
95        .execute(&mut **tx)
96        .await?;
97    }
98    for (position, (fact, presence)) in entry.consulted.iter().enumerate() {
99        sqlx::query(
100            "insert into gatekeep_audit_consulted_facts (decision_id, position, fact_id, presence) values ($1, $2, $3, $4)",
101        )
102        .bind(decision_id)
103        .bind(position_i32(position))
104        .bind(fact.as_str())
105        .bind(presence_label(*presence))
106        .execute(&mut **tx)
107        .await?;
108    }
109    for (slot, subject) in &entry.subjects {
110        sqlx::query(
111            "insert into gatekeep_audit_request_subjects (decision_id, slot, subject_kind, subject_id) values ($1, $2, $3, $4)",
112        )
113        .bind(decision_id)
114        .bind(slot.as_str())
115        .bind(subject.kind())
116        .bind(subject.id())
117        .execute(&mut **tx)
118        .await?;
119    }
120    if let Some(reason) = &entry.denial_reason {
121        for (key, value) in &reason.params {
122            sqlx::query(
123                "insert into gatekeep_audit_reason_params (decision_id, key, value) values ($1, $2, $3)",
124            )
125            .bind(decision_id)
126            .bind(key.as_str())
127            .bind(serde_json::to_value(value)?)
128            .execute(&mut **tx)
129            .await?;
130        }
131    }
132    Ok(())
133}
134
135async fn insert_outbox(
136    tx: &mut Transaction<'_, sqlx::Postgres>,
137    decision_id: i64,
138    entry: &AuditEntry,
139) -> Result<(), SqlxAuditError> {
140    sqlx::query("insert into gatekeep_audit_outbox (decision_id, payload) values ($1, $2)")
141        .bind(decision_id)
142        .bind(serde_json::to_value(entry)?)
143        .execute(&mut **tx)
144        .await?;
145    Ok(())
146}
gatekeep_sqlx/audit/postgres.rs

gatekeep_sqlx/audit/
postgres.rs
