keepsake_sqlx/repository/
expiry.rs

1use chrono::{DateTime, Utc};
2use keepsake::ExpiryCause;
3use uuid::Uuid;
4
5#[cfg(feature = "fulfillment-counters")]
6mod fulfillment;
7
8use super::{
9    KeepsakeRepository, RelationCache, RepositoryResult, TimedExpiryCandidate,
10    audit::record_audit_event_tx, support::expiry_event, validate_limit,
11};
12
13impl<C> KeepsakeRepository<C>
14where
15    C: RelationCache,
16{
17    /// Lists due timed expiry candidates in stable batch order.
18    pub async fn due_timed_expiry(
19        &self,
20        now: DateTime<Utc>,
21        limit: i64,
22    ) -> RepositoryResult<Vec<TimedExpiryCandidate>> {
23        let limit = validate_limit(limit)?;
24        let rows = sqlx::query_as::<_, TimedExpiryCandidate>(
25            r"
26            select k.id as keepsake_id, k.relation_id, k.subject_kind, k.subject_id, k.expires_at as due_at
27            from keepsakes k
28            join keepsake_relation_definitions r on r.id = k.relation_id
29            where k.state = 'applied'
30              and r.enabled
31              and k.expires_at is not null
32              and k.expires_at <= $1
33            order by k.expires_at, k.relation_id, k.subject_kind, k.subject_id, k.id
34            limit $2
35            ",
36        )
37        .bind(now)
38        .bind(limit)
39        .fetch_all(&self.pool)
40        .await?;
41        Ok(rows)
42    }
43    /// Expires a stable batch of due timed keepsakes.
44    pub async fn expire_due_timed(&self, now: DateTime<Utc>, limit: i64) -> RepositoryResult<u64> {
45        let limit = validate_limit(limit)?;
46        let mut tx = self.pool.begin().await?;
47        let candidates = due_timed_expiry_tx(&mut tx, now, limit).await?;
48        let ids = candidates
49            .iter()
50            .map(|row| row.keepsake_id)
51            .collect::<Vec<Uuid>>();
52        if ids.is_empty() {
53            tx.commit().await?;
54            return Ok(0);
55        }
56
57        let result = sqlx::query(
58            r"
59            update keepsakes
60            set state = 'expired', updated_at = $2
61            where id = any($1)
62              and state = 'applied'
63              and exists (
64                select 1
65                from keepsake_relation_definitions r
66                where r.id = keepsakes.relation_id and r.enabled
67              )
68            ",
69        )
70        .bind(&ids)
71        .bind(now)
72        .execute(&mut *tx)
73        .await?;
74        for candidate in candidates {
75            record_audit_event_tx(
76                &mut tx,
77                &expiry_event(
78                    now,
79                    ExpiryCause::Timed,
80                    candidate.keepsake_id,
81                    candidate.relation_id,
82                    candidate.subject_kind,
83                    candidate.subject_id,
84                )?,
85            )
86            .await?;
87        }
88        tx.commit().await?;
89        Ok(result.rows_affected())
90    }
91}
92
93async fn due_timed_expiry_tx(
94    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
95    now: DateTime<Utc>,
96    limit: i64,
97) -> RepositoryResult<Vec<TimedExpiryCandidate>> {
98    let rows = sqlx::query_as::<_, TimedExpiryCandidate>(
99        r"
100        select k.id as keepsake_id, k.relation_id, k.subject_kind, k.subject_id, k.expires_at as due_at
101        from keepsakes k
102        join keepsake_relation_definitions r on r.id = k.relation_id
103        where k.state = 'applied'
104          and r.enabled
105          and k.expires_at is not null
106          and k.expires_at <= $1
107        order by k.expires_at, k.relation_id, k.subject_kind, k.subject_id, k.id
108        limit $2
109        for update of k skip locked
110        for share of r
111        ",
112    )
113    .bind(now)
114    .bind(limit)
115    .fetch_all(&mut **tx)
116    .await?;
117    Ok(rows)
118}
keepsake_sqlx/repository/expiry.rs

keepsake_sqlx/repository/
expiry.rs
