cdk_sql_common/mint/
completed_operations.rs

1//! Completed operations database implementation
2
3use std::str::FromStr;
4
5use async_trait::async_trait;
6use cdk_common::database::mint::{CompletedOperationsDatabase, CompletedOperationsTransaction};
7use cdk_common::database::Error;
8use cdk_common::util::unix_time;
9use cdk_common::{mint, Amount, PaymentMethod};
10
11use super::{SQLMintDatabase, SQLTransaction};
12use crate::pool::DatabasePool;
13use crate::stmt::{query, Column};
14use crate::{column_as_nullable_string, column_as_number, column_as_string, unpack_into};
15
16fn sql_row_to_completed_operation(row: Vec<Column>) -> Result<mint::Operation, Error> {
17    unpack_into!(
18        let (
19            operation_id,
20            operation_kind,
21            completed_at,
22            total_issued,
23            total_redeemed,
24            fee_collected,
25            payment_method
26        ) = row
27    );
28
29    let operation_id_str = column_as_string!(&operation_id);
30    let operation_id = uuid::Uuid::parse_str(&operation_id_str)
31        .map_err(|e| Error::Internal(format!("Invalid operation_id UUID: {e}")))?;
32
33    let operation_kind_str = column_as_string!(&operation_kind);
34    let operation_kind = mint::OperationKind::from_str(&operation_kind_str)
35        .map_err(|e| Error::Internal(format!("Invalid operation kind: {e}")))?;
36
37    let completed_at: u64 = column_as_number!(completed_at);
38    let total_issued_u64: u64 = column_as_number!(total_issued);
39    let total_redeemed_u64: u64 = column_as_number!(total_redeemed);
40    let fee_collected_u64: u64 = column_as_number!(fee_collected);
41
42    let total_issued = Amount::from(total_issued_u64);
43    let total_redeemed = Amount::from(total_redeemed_u64);
44    let fee_collected = Amount::from(fee_collected_u64);
45
46    let payment_method = column_as_nullable_string!(payment_method)
47        .map(|s| PaymentMethod::from_str(&s))
48        .transpose()
49        .map_err(|e| Error::Internal(format!("Invalid payment method: {e}")))?;
50
51    Ok(mint::Operation::new(
52        operation_id,
53        operation_kind,
54        total_issued,
55        total_redeemed,
56        fee_collected,
57        Some(completed_at),
58        payment_method,
59    ))
60}
61
62#[async_trait]
63impl<RM> CompletedOperationsTransaction for SQLTransaction<RM>
64where
65    RM: DatabasePool + 'static,
66{
67    type Err = Error;
68
69    async fn add_completed_operation(
70        &mut self,
71        operation: &mint::Operation,
72        fee_by_keyset: &std::collections::HashMap<cdk_common::nuts::Id, cdk_common::Amount>,
73    ) -> Result<(), Self::Err> {
74        query(
75            r#"
76            INSERT INTO completed_operations
77            (operation_id, operation_kind, completed_at, total_issued, total_redeemed, fee_collected, payment_amount, payment_fee, payment_method)
78            VALUES
79            (:operation_id, :operation_kind, :completed_at, :total_issued, :total_redeemed, :fee_collected, :payment_amount, :payment_fee, :payment_method)
80            "#,
81        )?
82        .bind("operation_id", operation.id().to_string())
83        .bind("operation_kind", operation.kind().to_string())
84        .bind("completed_at", operation.completed_at().unwrap_or(unix_time()) as i64)
85        .bind("total_issued", operation.total_issued().to_u64() as i64)
86        .bind("total_redeemed", operation.total_redeemed().to_u64() as i64)
87        .bind("fee_collected", operation.fee_collected().to_u64() as i64)
88        .bind("payment_amount", operation.payment_amount().map(|a| a.to_u64() as i64))
89        .bind("payment_fee", operation.payment_fee().map(|a| a.to_u64() as i64))
90        .bind("payment_method", operation.payment_method().map(|m| m.to_string()))
91        .execute(&self.inner)
92        .await?;
93
94        // Acquire fee row locks in keyset ID order so concurrent fee updates
95        // cannot deadlock by visiting shared keysets in opposite orders.
96        let mut fees = fee_by_keyset.iter().collect::<Vec<_>>();
97        fees.sort_unstable_by_key(|(keyset_id, _)| *keyset_id);
98
99        for (keyset_id, fee) in fees {
100            if fee.to_u64() > 0 {
101                query(
102                    r#"
103                    INSERT INTO keyset_amounts (keyset_id, total_issued, total_redeemed, fee_collected)
104                    VALUES (:keyset_id, 0, 0, :fee)
105                    ON CONFLICT (keyset_id)
106                    DO UPDATE SET fee_collected = keyset_amounts.fee_collected + EXCLUDED.fee_collected
107                    "#,
108                )?
109                .bind("keyset_id", keyset_id.to_string())
110                .bind("fee", fee.to_u64() as i64)
111                .execute(&self.inner)
112                .await?;
113            }
114        }
115
116        Ok(())
117    }
118}
119
120#[async_trait]
121impl<RM> CompletedOperationsDatabase for SQLMintDatabase<RM>
122where
123    RM: DatabasePool + 'static,
124{
125    type Err = Error;
126
127    async fn get_completed_operation(
128        &self,
129        operation_id: &uuid::Uuid,
130    ) -> Result<Option<mint::Operation>, Self::Err> {
131        let conn = self
132            .pool
133            .get()
134            .await
135            .map_err(|e| Error::Database(Box::new(e)))?;
136        Ok(query(
137            r#"
138            SELECT
139                operation_id,
140                operation_kind,
141                completed_at,
142                total_issued,
143                total_redeemed,
144                fee_collected,
145                payment_method
146            FROM
147                completed_operations
148            WHERE
149                operation_id = :operation_id
150            "#,
151        )?
152        .bind("operation_id", operation_id.to_string())
153        .fetch_one(&*conn)
154        .await?
155        .map(sql_row_to_completed_operation)
156        .transpose()?)
157    }
158
159    async fn get_completed_operations_by_kind(
160        &self,
161        operation_kind: mint::OperationKind,
162    ) -> Result<Vec<mint::Operation>, Self::Err> {
163        let conn = self
164            .pool
165            .get()
166            .await
167            .map_err(|e| Error::Database(Box::new(e)))?;
168        Ok(query(
169            r#"
170            SELECT
171                operation_id,
172                operation_kind,
173                completed_at,
174                total_issued,
175                total_redeemed,
176                fee_collected,
177                payment_method
178            FROM
179                completed_operations
180            WHERE
181                operation_kind = :operation_kind
182            ORDER BY completed_at DESC
183            "#,
184        )?
185        .bind("operation_kind", operation_kind.to_string())
186        .fetch_all(&*conn)
187        .await?
188        .into_iter()
189        .map(sql_row_to_completed_operation)
190        .collect::<Result<Vec<_>, _>>()?)
191    }
192
193    async fn get_completed_operations(&self) -> Result<Vec<mint::Operation>, Self::Err> {
194        let conn = self
195            .pool
196            .get()
197            .await
198            .map_err(|e| Error::Database(Box::new(e)))?;
199        Ok(query(
200            r#"
201            SELECT
202                operation_id,
203                operation_kind,
204                completed_at,
205                total_issued,
206                total_redeemed,
207                fee_collected,
208                payment_method
209            FROM
210                completed_operations
211            ORDER BY completed_at DESC
212            "#,
213        )?
214        .fetch_all(&*conn)
215        .await?
216        .into_iter()
217        .map(sql_row_to_completed_operation)
218        .collect::<Result<Vec<_>, _>>()?)
219    }
220}
cdk_sql_common/mint/completed_operations.rs

cdk_sql_common/mint/
completed_operations.rs
