keepsake_sqlx/repository/
mutation.rs

1use keepsake::{
2    ApplyKeepsake, AuditDecision, AuditEvent, AuditEventType, Keepsake, RelationDefinition,
3    RelationId, RevokeKeepsake,
4};
5use sqlx::{Postgres, Transaction};
6use uuid::Uuid;
7
8use super::audit::{audit_context_from_command, record_audit_event_tx};
9use super::{
10    AppliedKeepsake, AppliedKeepsakeRow, AppliedKeepsakeWriteRow, KeepsakeRepository,
11    RelationCache, RelationRow, RepositoryError, RepositoryResult,
12};
13
14impl<C> KeepsakeRepository<C>
15where
16    C: RelationCache,
17{
18    /// Applies a command idempotently and records its audit event atomically.
19    ///
20    /// If an active keepsake already exists for the subject and relation, the existing
21    /// row is returned with `duplicate_prevented` set to true, even if the relation
22    /// has since been disabled. Disabled relations reject new non-duplicate applies.
23    pub async fn apply(&self, command: &ApplyKeepsake) -> RepositoryResult<AppliedKeepsake> {
24        command.subject.validate()?;
25        command.context.validate()?;
26
27        let mut tx = self.pool.begin().await?;
28        let relation = relation_for_share_tx(&mut tx, command.relation_id).await?;
29        let metadata = serde_json::to_value(&command.metadata)?;
30
31        let applied = sqlx::query_as::<_, AppliedKeepsakeWriteRow>(
32            r"
33            insert into keepsakes
34                (id, subject_kind, subject_id, relation_id, state, expiry_policy, applied_at, expires_at, metadata, created_at, updated_at)
35            select
36                $1,
37                $2,
38                $3,
39                r.id,
40                'applied',
41                r.expiry_policy,
42                $4,
43                case
44                    when r.expiry_policy->>'type' = 'at'
45                    then (r.expiry_policy->>'timestamp')::timestamptz
46                    else null
47                end,
48                $5,
49                $4,
50                $4
51            from keepsake_relation_definitions r
52            where r.id = $6
53            on conflict (subject_kind, subject_id, relation_id) where state = 'applied'
54            do update set updated_at = keepsakes.updated_at
55            returning id, subject_kind, subject_id, relation_id, state, expiry_policy, applied_at,
56                expires_at, fulfilled_at, revoked_at, metadata, (xmax <> 0) as duplicate_prevented
57            ",
58        )
59        .bind(command.id)
60        .bind(&command.subject.kind)
61        .bind(&command.subject.id)
62        .bind(command.at)
63        .bind(metadata)
64        .bind(command.relation_id)
65        .fetch_one(&mut *tx)
66        .await?;
67
68        if !relation.enabled && !applied.duplicate_prevented {
69            return Err(RepositoryError::RelationDisabled {
70                relation_id: command.relation_id,
71            });
72        }
73
74        let (keepsake, duplicate_prevented) = applied.try_into_parts()?;
75        let event = AuditEvent {
76            event_type: if duplicate_prevented {
77                AuditEventType::DuplicateApply
78            } else {
79                AuditEventType::Apply
80            },
81            at: command.at,
82            actor: command.context.actor.clone(),
83            keepsake_id: keepsake.id(),
84            subject: keepsake.subject().clone(),
85            relation_id: command.relation_id,
86            decision: AuditDecision::Applied {
87                duplicate_prevented,
88            },
89            context: audit_context_from_command(&command.context),
90        };
91        record_audit_event_tx(&mut tx, &event).await?;
92        tx.commit().await?;
93        Ok(AppliedKeepsake {
94            keepsake,
95            duplicate_prevented,
96        })
97    }
98
99    /// Revokes an active keepsake from a command and records its audit event atomically.
100    pub async fn revoke(&self, command: &RevokeKeepsake) -> RepositoryResult<bool> {
101        command.context.validate()?;
102
103        let mut tx = self.pool.begin().await?;
104        let revoked = revoke_tx(&mut tx, command.keepsake_id, command.at).await?;
105        if let Some(keepsake) = &revoked {
106            let event = AuditEvent {
107                event_type: AuditEventType::Revoke,
108                at: command.at,
109                actor: command.context.actor.clone(),
110                keepsake_id: keepsake.id(),
111                subject: keepsake.subject().clone(),
112                relation_id: keepsake.relation_id(),
113                decision: AuditDecision::Revoked,
114                context: audit_context_from_command(&command.context),
115            };
116            record_audit_event_tx(&mut tx, &event).await?;
117        }
118        tx.commit().await?;
119        Ok(revoked.is_some())
120    }
121}
122
123async fn revoke_tx(
124    tx: &mut Transaction<'_, Postgres>,
125    keepsake_id: Uuid,
126    at: chrono::DateTime<chrono::Utc>,
127) -> RepositoryResult<Option<Keepsake>> {
128    let row = sqlx::query_as::<_, AppliedKeepsakeRow>(
129        r"
130        update keepsakes
131        set state = 'revoked', revoked_at = $2, updated_at = $2
132        where id = $1 and state = 'applied'
133        returning id, subject_kind, subject_id, relation_id, state, expiry_policy, applied_at,
134            expires_at, fulfilled_at, revoked_at, metadata
135        ",
136    )
137    .bind(keepsake_id)
138    .bind(at)
139    .fetch_optional(&mut **tx)
140    .await?;
141
142    row.map(AppliedKeepsakeRow::try_into_keepsake).transpose()
143}
144
145async fn relation_for_share_tx(
146    tx: &mut Transaction<'_, Postgres>,
147    relation_id: RelationId,
148) -> RepositoryResult<RelationDefinition> {
149    let row = sqlx::query_as::<_, RelationRow>(
150        r"
151        select id, kind, key, enabled, expiry_policy
152        from keepsake_relation_definitions
153        where id = $1
154        for share
155        ",
156    )
157    .bind(relation_id)
158    .fetch_one(&mut **tx)
159    .await?;
160    row.try_into_relation()
161}
keepsake_sqlx/repository/mutation.rs

keepsake_sqlx/repository/
mutation.rs
