keepsake_sqlx/repository/
audit.rs

1use std::collections::BTreeMap;
2
3use keepsake::{AuditEvent, RelationId};
4use sqlx::{Postgres, Transaction};
5use uuid::Uuid;
6
7use super::{
8    AuditCursor, AuditEventRecord, AuditEventRow, KeepsakeRepository, RelationCache,
9    RepositoryResult, validate_limit,
10};
11
12impl<C> KeepsakeRepository<C>
13where
14    C: RelationCache,
15{
16    /// Appends an explicit audit event without mutating lifecycle state.
17    ///
18    /// Prefer `apply` and `revoke` for lifecycle mutations so state and audit
19    /// rows commit together. This helper is for application-owned audit events
20    /// that do not have a built-in repository command.
21    pub async fn append_audit_event(&self, event: &AuditEvent) -> RepositoryResult<i64> {
22        event.subject.validate()?;
23        event.actor.validate()?;
24
25        let mut tx = self.pool.begin().await?;
26        let audit_event_id = record_audit_event_tx(&mut tx, event).await?;
27        tx.commit().await?;
28        Ok(audit_event_id)
29    }
30
31    /// Reads audit events for a keepsake in stable `(occurred_at, id)` order.
32    pub async fn audit_events_for_keepsake(
33        &self,
34        keepsake_id: Uuid,
35        after: Option<&AuditCursor>,
36        limit: i64,
37    ) -> RepositoryResult<Vec<AuditEventRecord>> {
38        let limit = validate_limit(limit)?;
39        let rows = sqlx::query_as::<_, AuditEventRow>(
40            r"
41            select id, keepsake_id, relation_id, subject_kind, subject_id, actor_kind, actor_id,
42                event_type, decision, occurred_at
43            from keepsake_audit_events
44            where keepsake_id = $1
45              and ($2::timestamptz is null or (occurred_at, id) > ($2, $3))
46            order by occurred_at, id
47            limit $4
48            ",
49        )
50        .bind(keepsake_id)
51        .bind(after.map(|cursor| cursor.occurred_at))
52        .bind(after.map(|cursor| cursor.id))
53        .bind(limit)
54        .fetch_all(&self.pool)
55        .await?;
56        self.hydrate_audit_records(rows).await
57    }
58
59    /// Reads audit events for a relation in stable `(occurred_at, id)` order.
60    pub async fn audit_events_for_relation(
61        &self,
62        relation_id: RelationId,
63        after: Option<&AuditCursor>,
64        limit: i64,
65    ) -> RepositoryResult<Vec<AuditEventRecord>> {
66        let limit = validate_limit(limit)?;
67        let rows = sqlx::query_as::<_, AuditEventRow>(
68            r"
69            select id, keepsake_id, relation_id, subject_kind, subject_id, actor_kind, actor_id,
70                event_type, decision, occurred_at
71            from keepsake_audit_events
72            where relation_id = $1
73              and ($2::timestamptz is null or (occurred_at, id) > ($2, $3))
74            order by occurred_at, id
75            limit $4
76            ",
77        )
78        .bind(relation_id)
79        .bind(after.map(|cursor| cursor.occurred_at))
80        .bind(after.map(|cursor| cursor.id))
81        .bind(limit)
82        .fetch_all(&self.pool)
83        .await?;
84        self.hydrate_audit_records(rows).await
85    }
86
87    async fn hydrate_audit_records(
88        &self,
89        rows: Vec<AuditEventRow>,
90    ) -> RepositoryResult<Vec<AuditEventRecord>> {
91        if rows.is_empty() {
92            return Ok(Vec::new());
93        }
94        let ids = rows.iter().map(|row| row.id).collect::<Vec<i64>>();
95        let attribute_rows = sqlx::query_as::<_, (i64, String, String)>(
96            r"
97            select audit_event_id, key, value
98            from keepsake_audit_context_attributes
99            where audit_event_id = any($1)
100            ",
101        )
102        .bind(&ids)
103        .fetch_all(&self.pool)
104        .await?;
105        let mut attributes = BTreeMap::<i64, BTreeMap<String, String>>::new();
106        for (event_id, key, value) in attribute_rows {
107            attributes.entry(event_id).or_default().insert(key, value);
108        }
109        rows.into_iter()
110            .map(|row| {
111                let id = row.id;
112                row.into_record(attributes.remove(&id).unwrap_or_default())
113            })
114            .collect()
115    }
116}
117
118pub(super) async fn record_audit_event_tx(
119    tx: &mut Transaction<'_, Postgres>,
120    event: &AuditEvent,
121) -> RepositoryResult<i64> {
122    let decision = serde_json::to_value(&event.decision)?;
123    let audit_event_id = sqlx::query_scalar::<_, i64>(
124        r"
125        insert into keepsake_audit_events
126            (keepsake_id, relation_id, subject_kind, subject_id, actor_kind, actor_id,
127             event_type, decision, occurred_at)
128        values ($1, $2, $3, $4, $5, $6, $7, $8, $9)
129        returning id
130        ",
131    )
132    .bind(event.keepsake_id)
133    .bind(event.relation_id)
134    .bind(&event.subject.kind)
135    .bind(&event.subject.id)
136    .bind(&event.actor.kind)
137    .bind(&event.actor.id)
138    .bind(event.event_type.as_str())
139    .bind(decision)
140    .bind(event.at)
141    .fetch_one(&mut **tx)
142    .await?;
143
144    if event.context.attributes.is_empty() {
145        return Ok(audit_event_id);
146    }
147
148    let keys = event.context.attributes.keys().cloned().collect::<Vec<_>>();
149    let values = event
150        .context
151        .attributes
152        .values()
153        .cloned()
154        .collect::<Vec<_>>();
155    sqlx::query(
156        r"
157        insert into keepsake_audit_context_attributes (audit_event_id, key, value)
158        select $1, attribute.key, attribute.value
159        from unnest($2::text[], $3::text[]) as attribute(key, value)
160        ",
161    )
162    .bind(audit_event_id)
163    .bind(&keys)
164    .bind(&values)
165    .execute(&mut **tx)
166    .await?;
167
168    Ok(audit_event_id)
169}
keepsake_sqlx/repository/audit.rs

keepsake_sqlx/repository/
audit.rs
