1use std::str::FromStr;
4
5use async_trait::async_trait;
6use cdk_common::database::mint::{Acquired, SagaDatabase, SagaTransaction};
7use cdk_common::database::Error;
8use cdk_common::mint;
9use cdk_common::util::unix_time;
10use serde_json;
11
12use super::{SQLMintDatabase, SQLTransaction};
13use crate::pool::DatabasePool;
14use crate::stmt::{query, Column};
15use crate::{column_as_number, column_as_string, unpack_into};
16
17fn sql_row_to_saga(row: Vec<Column>) -> Result<mint::Saga, Error> {
18 unpack_into!(
19 let (
20 operation_id,
21 operation_kind,
22 state,
23 quote_id,
24 finalization_data,
25 created_at,
26 updated_at
27 ) = row
28 );
29
30 let operation_id_str = column_as_string!(&operation_id);
31 let operation_id = uuid::Uuid::parse_str(&operation_id_str)
32 .map_err(|e| Error::Internal(format!("Invalid operation_id UUID: {e}")))?;
33
34 let operation_kind_str = column_as_string!(&operation_kind);
35 let operation_kind = mint::OperationKind::from_str(&operation_kind_str)
36 .map_err(|e| Error::Internal(format!("Invalid operation kind: {e}")))?;
37
38 let state_str = column_as_string!(&state);
39 let state = mint::SagaStateEnum::new(operation_kind, &state_str)
40 .map_err(|e| Error::Internal(format!("Invalid saga state: {e}")))?;
41
42 let quote_id = match "e_id {
43 Column::Text(s) => {
44 if s.is_empty() {
45 None
46 } else {
47 Some(s.clone())
48 }
49 }
50 Column::Null => None,
51 _ => None,
52 };
53
54 let finalization_data =
55 match &finalization_data {
56 Column::Text(s) => Some(serde_json::from_str(s).map_err(|e| {
57 Error::Internal(format!("Invalid melt finalization data JSON: {e}"))
58 })?),
59 Column::Null => None,
60 _ => None,
61 };
62
63 let created_at: u64 = column_as_number!(created_at);
64 let updated_at: u64 = column_as_number!(updated_at);
65
66 Ok(mint::Saga {
67 operation_id,
68 operation_kind,
69 state,
70 quote_id,
71 finalization_data,
72 created_at,
73 updated_at,
74 })
75}
76
77#[async_trait]
78impl<RM> SagaTransaction for SQLTransaction<RM>
79where
80 RM: DatabasePool + 'static,
81{
82 type Err = Error;
83
84 async fn get_saga(
85 &mut self,
86 operation_id: &uuid::Uuid,
87 ) -> Result<Option<mint::Saga>, Self::Err> {
88 Ok(query(
89 r#"
90 SELECT
91 operation_id,
92 operation_kind,
93 state,
94 quote_id,
95 finalization_data,
96 created_at,
97 updated_at
98 FROM
99 saga_state
100 WHERE
101 operation_id = :operation_id
102 FOR UPDATE
103 "#,
104 )?
105 .bind("operation_id", operation_id.to_string())
106 .fetch_one(&self.inner)
107 .await?
108 .map(sql_row_to_saga)
109 .transpose()?)
110 }
111
112 async fn get_saga_for_update(
113 &mut self,
114 operation_id: &uuid::Uuid,
115 ) -> Result<Option<Acquired<mint::Saga>>, Self::Err> {
116 Ok(self.get_saga(operation_id).await?.map(Into::into))
117 }
118
119 async fn add_saga(&mut self, saga: &mint::Saga) -> Result<(), Self::Err> {
120 let current_time = unix_time();
121
122 query(
123 r#"
124 INSERT INTO saga_state
125 (operation_id, operation_kind, state, quote_id, finalization_data, created_at, updated_at)
126 VALUES
127 (:operation_id, :operation_kind, :state, :quote_id, :finalization_data, :created_at, :updated_at)
128 "#,
129 )?
130 .bind("operation_id", saga.operation_id.to_string())
131 .bind("operation_kind", saga.operation_kind.to_string())
132 .bind("state", saga.state.state())
133 .bind("quote_id", saga.quote_id.as_deref())
134 .bind(
135 "finalization_data",
136 saga.finalization_data
137 .as_ref()
138 .map(serde_json::to_string)
139 .transpose()
140 .map_err(|e| Error::Internal(format!("Failed to serialize melt finalization data: {e}")))?,
141 )
142 .bind("created_at", saga.created_at as i64)
143 .bind("updated_at", current_time as i64)
144 .execute(&self.inner)
145 .await?;
146
147 Ok(())
148 }
149
150 async fn update_acquired_saga(
151 &mut self,
152 saga: &mut Acquired<mint::Saga>,
153 new_state: mint::SagaStateEnum,
154 ) -> Result<(), Self::Err> {
155 let current_time = unix_time();
156
157 let affected = query(
158 r#"
159 UPDATE saga_state
160 SET state = :state, updated_at = :updated_at
161 WHERE operation_id = :operation_id
162 "#,
163 )?
164 .bind("state", new_state.state())
165 .bind("updated_at", current_time as i64)
166 .bind("operation_id", saga.operation_id.to_string())
167 .execute(&self.inner)
168 .await?;
169
170 if affected != 1 {
171 return Err(Error::Internal(format!(
172 "Saga {} not found for state update to {}",
173 saga.operation_id,
174 new_state.state()
175 )));
176 }
177
178 saga.state = new_state;
179 saga.updated_at = current_time;
180
181 Ok(())
182 }
183
184 async fn update_acquired_saga_with_finalization_data(
185 &mut self,
186 saga: &mut Acquired<mint::Saga>,
187 new_state: mint::SagaStateEnum,
188 finalization_data: Option<&mint::MeltFinalizationData>,
189 ) -> Result<(), Self::Err> {
190 let current_time = unix_time();
191
192 let affected = query(
193 r#"
194 UPDATE saga_state
195 SET state = :state, finalization_data = :finalization_data, updated_at = :updated_at
196 WHERE operation_id = :operation_id
197 "#,
198 )?
199 .bind("state", new_state.state())
200 .bind(
201 "finalization_data",
202 finalization_data
203 .map(serde_json::to_string)
204 .transpose()
205 .map_err(|e| {
206 Error::Internal(format!("Failed to serialize melt finalization data: {e}"))
207 })?,
208 )
209 .bind("updated_at", current_time as i64)
210 .bind("operation_id", saga.operation_id.to_string())
211 .execute(&self.inner)
212 .await?;
213
214 if affected != 1 {
215 return Err(Error::Internal(format!(
216 "Saga {} not found for finalization update to {}",
217 saga.operation_id,
218 new_state.state()
219 )));
220 }
221
222 saga.state = new_state;
223 saga.finalization_data = finalization_data.cloned();
224 saga.updated_at = current_time;
225
226 Ok(())
227 }
228
229 async fn delete_saga(&mut self, operation_id: &uuid::Uuid) -> Result<(), Self::Err> {
230 query(
231 r#"
232 DELETE FROM saga_state
233 WHERE operation_id = :operation_id
234 "#,
235 )?
236 .bind("operation_id", operation_id.to_string())
237 .execute(&self.inner)
238 .await?;
239
240 Ok(())
241 }
242}
243
244#[async_trait]
245impl<RM> SagaDatabase for SQLMintDatabase<RM>
246where
247 RM: DatabasePool + 'static,
248{
249 type Err = Error;
250
251 async fn get_melt_saga_by_quote_id(
252 &self,
253 quote_id: &cdk_common::QuoteId,
254 ) -> Result<Option<mint::Saga>, Self::Err> {
255 let conn = self
256 .pool
257 .get()
258 .await
259 .map_err(|e| Error::Database(Box::new(e)))?;
260 Ok(query(
261 r#"
262 SELECT
263 operation_id,
264 operation_kind,
265 state,
266 quote_id,
267 finalization_data,
268 created_at,
269 updated_at
270 FROM
271 saga_state
272 WHERE
273 quote_id = :quote_id
274 AND operation_kind = :operation_kind
275 "#,
276 )?
277 .bind("quote_id", quote_id.to_string())
278 .bind("operation_kind", mint::OperationKind::Melt.to_string())
279 .fetch_one(&*conn)
280 .await?
281 .map(sql_row_to_saga)
282 .transpose()?)
283 }
284
285 async fn get_incomplete_sagas(
286 &self,
287 operation_kind: mint::OperationKind,
288 ) -> Result<Vec<mint::Saga>, Self::Err> {
289 let conn = self
290 .pool
291 .get()
292 .await
293 .map_err(|e| Error::Database(Box::new(e)))?;
294 Ok(query(
295 r#"
296 SELECT
297 operation_id,
298 operation_kind,
299 state,
300 quote_id,
301 finalization_data,
302 created_at,
303 updated_at
304 FROM
305 saga_state
306 WHERE
307 operation_kind = :operation_kind
308 ORDER BY created_at ASC
309 "#,
310 )?
311 .bind("operation_kind", operation_kind.to_string())
312 .fetch_all(&*conn)
313 .await?
314 .into_iter()
315 .map(sql_row_to_saga)
316 .collect::<Result<Vec<_>, _>>()?)
317 }
318}