keepsake_sqlx/repository/
expiry.rs

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

keepsake_sqlx/repository/
expiry.rs
