keepsake_sqlx/repository/
relation.rs1use 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 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 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 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 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 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}