keepsake_sqlx/repository/
relation.rs

1use chrono::{DateTime, Utc};
2use keepsake::{RelationDefinition, RelationId, RelationKey, RelationSpec};
3use uuid::Uuid;
4
5use super::{KeepsakeRepository, RelationCache, RelationRow, RepositoryError, RepositoryResult};
6
7impl<C> KeepsakeRepository<C>
8where
9    C: RelationCache,
10{
11    /// Inserts or updates a relation definition by its natural relation key.
12    ///
13    /// If a relation already exists for the same kind/name, its stable id is preserved and
14    /// the returned definition contains the existing id.
15    pub async fn upsert_relation(
16        &self,
17        relation: &RelationDefinition,
18        at: DateTime<Utc>,
19    ) -> RepositoryResult<RelationDefinition> {
20        let expiry_policy = serde_json::to_value(&relation.expiry)?;
21        let row = sqlx::query_as::<_, RelationRow>(
22            r"
23            insert into keepsake_relation_definitions
24                (id, kind, key, enabled, expiry_policy, created_at, updated_at)
25            values ($1, $2, $3, $4, $5, $6, $6)
26            on conflict (kind, key) do update set
27                enabled = excluded.enabled,
28                expiry_policy = excluded.expiry_policy,
29                updated_at = $6
30            returning id, kind, key, enabled, expiry_policy
31            ",
32        )
33        .bind(relation.id)
34        .bind(relation.key.kind())
35        .bind(relation.key.name())
36        .bind(relation.enabled)
37        .bind(expiry_policy)
38        .bind(at)
39        .fetch_one(&self.pool)
40        .await?;
41        let relation = row.try_into_relation()?;
42        self.relation_cache.remove_by_id(relation.id).await;
43        Ok(relation)
44    }
45
46    /// Inserts or updates a typed relation spec by its natural relation key.
47    pub async fn upsert_relation_spec<Spec>(
48        &self,
49        at: DateTime<Utc>,
50    ) -> RepositoryResult<RelationDefinition>
51    where
52        Spec: RelationSpec,
53    {
54        let relation = RelationDefinition::from_spec::<Spec>(at)?;
55        let expiry_policy = serde_json::to_value(&relation.expiry)?;
56        let mut tx = self.pool.begin().await?;
57        let row = sqlx::query_as::<_, RelationRow>(
58            r"
59            insert into keepsake_relation_definitions
60                (id, kind, key, enabled, expiry_policy, created_at, updated_at)
61            values ($1, $2, $3, $4, $5, $6, $6)
62            on conflict (kind, key) do update set
63                enabled = excluded.enabled,
64                expiry_policy = excluded.expiry_policy,
65                updated_at = $6
66            where keepsake_relation_definitions.id = excluded.id
67            returning id, kind, key, enabled, expiry_policy
68            ",
69        )
70        .bind(relation.id)
71        .bind(relation.key.kind())
72        .bind(relation.key.name())
73        .bind(relation.enabled)
74        .bind(expiry_policy)
75        .bind(at)
76        .fetch_optional(&mut *tx)
77        .await?;
78
79        let Some(row) = row else {
80            let stored_relation_id = sqlx::query_scalar::<_, Uuid>(
81                r"
82                select id
83                from keepsake_relation_definitions
84                where kind = $1 and key = $2
85                ",
86            )
87            .bind(relation.key.kind())
88            .bind(relation.key.name())
89            .fetch_one(&mut *tx)
90            .await?;
91            return Err(RepositoryError::RelationSpecIdMismatch {
92                kind: relation.key.kind().to_owned(),
93                name: relation.key.name().to_owned(),
94                expected_relation_id: relation.id,
95                stored_relation_id,
96            });
97        };
98
99        tx.commit().await?;
100        let relation = row.try_into_relation()?;
101        self.relation_cache.remove_by_id(relation.id).await;
102        Ok(relation)
103    }
104
105    /// Looks up a relation definition by stable id.
106    pub async fn relation_by_id(
107        &self,
108        relation_id: RelationId,
109    ) -> RepositoryResult<Option<RelationDefinition>> {
110        if let Some(relation) = self.relation_cache.get_by_id(relation_id).await {
111            return Ok(Some(relation));
112        }
113
114        let relation = self.fetch_relation_by_id(relation_id).await?;
115        if let Some(relation) = &relation {
116            self.relation_cache.store(relation).await;
117        }
118        Ok(relation)
119    }
120
121    /// Looks up a relation definition by its natural relation key.
122    pub async fn relation_by_key(
123        &self,
124        key: &RelationKey,
125    ) -> RepositoryResult<Option<RelationDefinition>> {
126        if let Some(relation) = self.relation_cache.get_by_key(key).await {
127            return Ok(Some(relation));
128        }
129
130        let relation = self.fetch_relation_by_key(key).await?;
131        if let Some(relation) = &relation {
132            self.relation_cache.store(relation).await;
133        }
134        Ok(relation)
135    }
136
137    /// Enables or disables a relation.
138    pub async fn set_relation_enabled(
139        &self,
140        relation_id: RelationId,
141        enabled: bool,
142        at: DateTime<Utc>,
143    ) -> RepositoryResult<bool> {
144        let result = sqlx::query(
145            r"
146            update keepsake_relation_definitions
147            set enabled = $2, updated_at = $3
148            where id = $1
149            ",
150        )
151        .bind(relation_id)
152        .bind(enabled)
153        .bind(at)
154        .execute(&self.pool)
155        .await?;
156        let changed = result.rows_affected() == 1;
157        if changed {
158            self.relation_cache.remove_by_id(relation_id).await;
159        }
160        Ok(changed)
161    }
162
163    async fn fetch_relation_by_id(
164        &self,
165        relation_id: RelationId,
166    ) -> RepositoryResult<Option<RelationDefinition>> {
167        let row = sqlx::query_as::<_, RelationRow>(
168            r"
169            select id, kind, key, enabled, expiry_policy
170            from keepsake_relation_definitions
171            where id = $1
172            ",
173        )
174        .bind(relation_id)
175        .fetch_optional(&self.pool)
176        .await?;
177
178        row.map(RelationRow::try_into_relation).transpose()
179    }
180
181    async fn fetch_relation_by_key(
182        &self,
183        key: &RelationKey,
184    ) -> RepositoryResult<Option<RelationDefinition>> {
185        let row = sqlx::query_as::<_, RelationRow>(
186            r"
187            select id, kind, key, enabled, expiry_policy
188            from keepsake_relation_definitions
189            where kind = $1 and key = $2
190            ",
191        )
192        .bind(key.kind())
193        .bind(key.name())
194        .fetch_optional(&self.pool)
195        .await?;
196
197        row.map(RelationRow::try_into_relation).transpose()
198    }
199}
keepsake_sqlx/repository/relation.rs

keepsake_sqlx/repository/
relation.rs
