stateset_db/sqlite/
payments.rs

1//! SQLite implementation of payment repository
2
3use super::{
4    build_in_clause, map_db_error, params_refs, parse_datetime_opt_row, parse_datetime_row,
5    parse_decimal_row, parse_enum_row, parse_uuid_opt_row, parse_uuid_row, uuid_params,
6    with_immediate_transaction,
7};
8use r2d2::Pool;
9use r2d2_sqlite::SqliteConnectionManager;
10use rusqlite::{Row, params};
11use stateset_core::{
12    BatchResult, CommerceError, CreatePayment, CreatePaymentMethod, CreateRefund, CustomerId,
13    InvoiceId, OrderId, Payment, PaymentFilter, PaymentId, PaymentMethod, PaymentRepository,
14    PaymentTransactionStatus, Refund, RefundStatus, Result, UpdatePayment, generate_payment_number,
15    generate_refund_number, validate_batch_size,
16};
17use uuid::Uuid;
18
19#[derive(Debug)]
20pub struct SqlitePaymentRepository {
21    pool: Pool<SqliteConnectionManager>,
22}
23
24impl SqlitePaymentRepository {
25    #[must_use]
26    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
27        Self { pool }
28    }
29
30    fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
31        self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
32    }
33
34    fn row_to_payment(row: &Row<'_>) -> rusqlite::Result<Payment> {
35        Ok(Payment {
36            id: PaymentId::from(parse_uuid_row(&row.get::<_, String>("id")?, "payment", "id")?),
37            payment_number: row.get("payment_number")?,
38            order_id: parse_uuid_opt_row(
39                row.get::<_, Option<String>>("order_id")?,
40                "payment",
41                "order_id",
42            )?
43            .map(OrderId::from),
44            invoice_id: parse_uuid_opt_row(
45                row.get::<_, Option<String>>("invoice_id")?,
46                "payment",
47                "invoice_id",
48            )?,
49            customer_id: parse_uuid_opt_row(
50                row.get::<_, Option<String>>("customer_id")?,
51                "payment",
52                "customer_id",
53            )?
54            .map(CustomerId::from),
55            status: parse_enum_row(&row.get::<_, String>("status")?, "payment", "status")?,
56            payment_method: parse_enum_row(
57                &row.get::<_, String>("payment_method")?,
58                "payment",
59                "payment_method",
60            )?,
61            amount: parse_decimal_row(&row.get::<_, String>("amount")?, "payment", "amount")?,
62            currency: row.get("currency")?,
63            amount_refunded: parse_decimal_row(
64                &row.get::<_, String>("amount_refunded")?,
65                "payment",
66                "amount_refunded",
67            )?,
68            external_id: row.get("external_id")?,
69            idempotency_key: row.get("idempotency_key")?,
70            processor: row.get("processor")?,
71            card_brand: match row.get::<_, Option<String>>("card_brand")? {
72                Some(value) => Some(parse_enum_row(&value, "payment", "card_brand")?),
73                None => None,
74            },
75            card_last4: row.get("card_last4")?,
76            card_exp_month: row.get("card_exp_month")?,
77            card_exp_year: row.get("card_exp_year")?,
78            // Blockchain/Stablecoin fields
79            blockchain_network: match row
80                .get::<_, Option<String>>("blockchain_network")
81                .ok()
82                .flatten()
83            {
84                Some(value) => Some(parse_enum_row(&value, "payment", "blockchain_network")?),
85                None => None,
86            },
87            stablecoin_type: match row.get::<_, Option<String>>("stablecoin_type").ok().flatten() {
88                Some(value) => Some(parse_enum_row(&value, "payment", "stablecoin_type")?),
89                None => None,
90            },
91            from_wallet_address: row.get("from_wallet_address").ok().flatten(),
92            to_wallet_address: row.get("to_wallet_address").ok().flatten(),
93            tx_hash: row.get("tx_hash").ok().flatten(),
94            block_number: row.get("block_number").ok().flatten(),
95            confirmations: row.get("confirmations").ok().flatten(),
96            token_address: row.get("token_address").ok().flatten(),
97            ves_intent_id: row.get("ves_intent_id").ok().flatten(),
98            billing_email: row.get("billing_email")?,
99            billing_name: row.get("billing_name")?,
100            billing_address: row.get("billing_address")?,
101            description: row.get("description")?,
102            failure_reason: row.get("failure_reason")?,
103            failure_code: row.get("failure_code")?,
104            metadata: row.get("metadata")?,
105            paid_at: parse_datetime_opt_row(
106                row.get::<_, Option<String>>("paid_at")?,
107                "payment",
108                "paid_at",
109            )?,
110            version: row.get::<_, Option<i32>>("version")?.unwrap_or(1),
111            created_at: parse_datetime_row(
112                &row.get::<_, String>("created_at")?,
113                "payment",
114                "created_at",
115            )?,
116            updated_at: parse_datetime_row(
117                &row.get::<_, String>("updated_at")?,
118                "payment",
119                "updated_at",
120            )?,
121        })
122    }
123
124    fn row_to_refund(row: &Row<'_>) -> rusqlite::Result<Refund> {
125        Ok(Refund {
126            id: parse_uuid_row(&row.get::<_, String>("id")?, "refund", "id")?,
127            refund_number: row.get("refund_number")?,
128            payment_id: PaymentId::from(parse_uuid_row(
129                &row.get::<_, String>("payment_id")?,
130                "refund",
131                "payment_id",
132            )?),
133            status: parse_enum_row(&row.get::<_, String>("status")?, "refund", "status")?,
134            amount: parse_decimal_row(&row.get::<_, String>("amount")?, "refund", "amount")?,
135            currency: row.get("currency")?,
136            reason: row.get("reason")?,
137            external_id: row.get("external_id")?,
138            idempotency_key: row.get("idempotency_key")?,
139            failure_reason: row.get("failure_reason")?,
140            notes: row.get("notes")?,
141            refunded_at: parse_datetime_opt_row(
142                row.get::<_, Option<String>>("refunded_at")?,
143                "refund",
144                "refunded_at",
145            )?,
146            created_at: parse_datetime_row(
147                &row.get::<_, String>("created_at")?,
148                "refund",
149                "created_at",
150            )?,
151            updated_at: parse_datetime_row(
152                &row.get::<_, String>("updated_at")?,
153                "refund",
154                "updated_at",
155            )?,
156        })
157    }
158
159    fn row_to_payment_method(row: &Row<'_>) -> rusqlite::Result<PaymentMethod> {
160        Ok(PaymentMethod {
161            id: parse_uuid_row(&row.get::<_, String>("id")?, "payment_method", "id")?,
162            customer_id: CustomerId::from(parse_uuid_row(
163                &row.get::<_, String>("customer_id")?,
164                "payment_method",
165                "customer_id",
166            )?),
167            method_type: parse_enum_row(
168                &row.get::<_, String>("method_type")?,
169                "payment_method",
170                "method_type",
171            )?,
172            is_default: row.get::<_, i32>("is_default")? != 0,
173            card_brand: match row.get::<_, Option<String>>("card_brand")? {
174                Some(value) => Some(parse_enum_row(&value, "payment_method", "card_brand")?),
175                None => None,
176            },
177            card_last4: row.get("card_last4")?,
178            card_exp_month: row.get("card_exp_month")?,
179            card_exp_year: row.get("card_exp_year")?,
180            cardholder_name: row.get("cardholder_name")?,
181            bank_name: row.get("bank_name")?,
182            account_last4: row.get("account_last4")?,
183            // Blockchain/Wallet fields
184            wallet_address: row.get("wallet_address").ok().flatten(),
185            blockchain_network: match row
186                .get::<_, Option<String>>("blockchain_network")
187                .ok()
188                .flatten()
189            {
190                Some(value) => {
191                    Some(parse_enum_row(&value, "payment_method", "blockchain_network")?)
192                }
193                None => None,
194            },
195            stablecoin_type: match row.get::<_, Option<String>>("stablecoin_type").ok().flatten() {
196                Some(value) => Some(parse_enum_row(&value, "payment_method", "stablecoin_type")?),
197                None => None,
198            },
199            external_id: row.get("external_id")?,
200            billing_address: row.get("billing_address")?,
201            created_at: parse_datetime_row(
202                &row.get::<_, String>("created_at")?,
203                "payment_method",
204                "created_at",
205            )?,
206            updated_at: parse_datetime_row(
207                &row.get::<_, String>("updated_at")?,
208                "payment_method",
209                "updated_at",
210            )?,
211        })
212    }
213
214    fn get_by_idempotency_key(&self, key: &str) -> Result<Option<Payment>> {
215        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
216        let mut stmt = conn
217            .prepare("SELECT * FROM payments WHERE idempotency_key = ?")
218            .map_err(map_db_error)?;
219        let result = stmt.query_row([key], Self::row_to_payment);
220        match result {
221            Ok(payment) => Ok(Some(payment)),
222            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
223            Err(e) => Err(map_db_error(e)),
224        }
225    }
226
227    fn get_refund_by_idempotency_key(&self, key: &str) -> Result<Option<Refund>> {
228        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
229        let mut stmt = conn
230            .prepare("SELECT * FROM refunds WHERE idempotency_key = ?")
231            .map_err(map_db_error)?;
232        let result = stmt.query_row([key], Self::row_to_refund);
233        match result {
234            Ok(refund) => Ok(Some(refund)),
235            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
236            Err(e) => Err(map_db_error(e)),
237        }
238    }
239}
240
241impl PaymentRepository for SqlitePaymentRepository {
242    fn create(&self, input: CreatePayment) -> Result<Payment> {
243        if let Some(key) = input.idempotency_key.as_deref() {
244            if let Some(existing) = self.get_by_idempotency_key(key)? {
245                return Ok(existing);
246            }
247        }
248
249        let id = Uuid::new_v4();
250        let now = chrono::Utc::now();
251        let payment_number = generate_payment_number();
252
253        {
254            let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
255            conn.execute(
256                "INSERT INTO payments (id, payment_number, order_id, invoice_id, customer_id, status,
257                 payment_method, amount, currency, amount_refunded, external_id, idempotency_key, processor,
258                 card_brand, card_last4, card_exp_month, card_exp_year, billing_email, billing_name,
259                 billing_address, description, metadata, created_at, updated_at)
260                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
261                params![
262                    id.to_string(),
263                    payment_number,
264                    input.order_id.map(|id| id.to_string()),
265                    input.invoice_id.map(|id| id.to_string()),
266                    input.customer_id.map(|id| id.to_string()),
267                    PaymentTransactionStatus::Pending.to_string(),
268                    input.payment_method.to_string(),
269                    input.amount.to_string(),
270                    input.currency.unwrap_or_default(),
271                    "0",
272                    input.external_id,
273                    input.idempotency_key,
274                    input.processor,
275                    input.card_brand.map(|b| b.to_string()),
276                    input.card_last4,
277                    input.card_exp_month,
278                    input.card_exp_year,
279                    input.billing_email,
280                    input.billing_name,
281                    input.billing_address,
282                    input.description,
283                    input.metadata,
284                    now.to_rfc3339(),
285                    now.to_rfc3339(),
286                ],
287            ).map_err(map_db_error)?;
288        }
289
290        self.get(PaymentId::from(id))?.ok_or(CommerceError::NotFound)
291    }
292
293    fn get(&self, id: PaymentId) -> Result<Option<Payment>> {
294        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
295        let mut stmt = conn.prepare("SELECT * FROM payments WHERE id = ?").map_err(map_db_error)?;
296        let result = stmt.query_row([id.to_string()], Self::row_to_payment);
297        match result {
298            Ok(payment) => Ok(Some(payment)),
299            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
300            Err(e) => Err(map_db_error(e)),
301        }
302    }
303
304    fn get_by_number(&self, payment_number: &str) -> Result<Option<Payment>> {
305        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
306        let mut stmt = conn
307            .prepare("SELECT * FROM payments WHERE payment_number = ?")
308            .map_err(map_db_error)?;
309        let result = stmt.query_row([payment_number], Self::row_to_payment);
310        match result {
311            Ok(payment) => Ok(Some(payment)),
312            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
313            Err(e) => Err(map_db_error(e)),
314        }
315    }
316
317    fn get_by_external_id(&self, external_id: &str) -> Result<Option<Payment>> {
318        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
319        let mut stmt =
320            conn.prepare("SELECT * FROM payments WHERE external_id = ?").map_err(map_db_error)?;
321        let result = stmt.query_row([external_id], Self::row_to_payment);
322        match result {
323            Ok(payment) => Ok(Some(payment)),
324            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
325            Err(e) => Err(map_db_error(e)),
326        }
327    }
328
329    fn update(&self, id: PaymentId, input: UpdatePayment) -> Result<Payment> {
330        let payment = self.get(id)?.ok_or(CommerceError::NotFound)?;
331        let now = chrono::Utc::now();
332
333        {
334            let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
335            conn.execute(
336                "UPDATE payments SET status = ?, external_id = ?, failure_reason = ?,
337                 failure_code = ?, metadata = ?, updated_at = ? WHERE id = ?",
338                params![
339                    input.status.unwrap_or(payment.status).to_string(),
340                    input.external_id.or(payment.external_id),
341                    input.failure_reason.or(payment.failure_reason),
342                    input.failure_code.or(payment.failure_code),
343                    input.metadata.or(payment.metadata),
344                    now.to_rfc3339(),
345                    id.to_string(),
346                ],
347            )
348            .map_err(map_db_error)?;
349        }
350
351        self.get(id)?.ok_or(CommerceError::NotFound)
352    }
353
354    fn list(&self, filter: PaymentFilter) -> Result<Vec<Payment>> {
355        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
356
357        let mut sql = "SELECT * FROM payments WHERE 1=1".to_string();
358        let mut params_vec: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
359
360        if let Some(order_id) = &filter.order_id {
361            sql.push_str(" AND order_id = ?");
362            params_vec.push(Box::new(order_id.to_string()));
363        }
364        if let Some(invoice_id) = &filter.invoice_id {
365            sql.push_str(" AND invoice_id = ?");
366            params_vec.push(Box::new(invoice_id.to_string()));
367        }
368        if let Some(customer_id) = &filter.customer_id {
369            sql.push_str(" AND customer_id = ?");
370            params_vec.push(Box::new(customer_id.to_string()));
371        }
372        if let Some(status) = &filter.status {
373            sql.push_str(" AND status = ?");
374            params_vec.push(Box::new(status.to_string()));
375        }
376
377        sql.push_str(" ORDER BY created_at DESC");
378
379        crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
380
381        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
382        let params_refs: Vec<&dyn rusqlite::ToSql> =
383            params_vec.iter().map(std::convert::AsRef::as_ref).collect();
384        let rows =
385            stmt.query_map(params_refs.as_slice(), Self::row_to_payment).map_err(map_db_error)?;
386
387        let mut payments = Vec::new();
388        for row in rows {
389            payments.push(row.map_err(map_db_error)?);
390        }
391        Ok(payments)
392    }
393
394    fn for_order(&self, order_id: OrderId) -> Result<Vec<Payment>> {
395        self.list(PaymentFilter { order_id: Some(order_id), ..Default::default() })
396    }
397
398    fn for_invoice(&self, invoice_id: InvoiceId) -> Result<Vec<Payment>> {
399        self.list(PaymentFilter { invoice_id: Some(invoice_id.into()), ..Default::default() })
400    }
401
402    fn mark_processing(&self, id: PaymentId) -> Result<Payment> {
403        self.update(
404            id,
405            UpdatePayment {
406                status: Some(PaymentTransactionStatus::Processing),
407                ..Default::default()
408            },
409        )
410    }
411
412    fn mark_completed(&self, id: PaymentId) -> Result<Payment> {
413        let now = chrono::Utc::now();
414
415        {
416            let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
417            conn.execute(
418                "UPDATE payments SET status = ?, paid_at = ?, updated_at = ? WHERE id = ?",
419                params![
420                    PaymentTransactionStatus::Completed.to_string(),
421                    now.to_rfc3339(),
422                    now.to_rfc3339(),
423                    id.to_string()
424                ],
425            )
426            .map_err(map_db_error)?;
427        }
428
429        self.get(id)?.ok_or(CommerceError::NotFound)
430    }
431
432    fn mark_failed(&self, id: PaymentId, reason: &str, code: Option<&str>) -> Result<Payment> {
433        let now = chrono::Utc::now();
434
435        {
436            let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
437            conn.execute(
438                "UPDATE payments SET status = ?, failure_reason = ?, failure_code = ?, updated_at = ? WHERE id = ?",
439                params![PaymentTransactionStatus::Failed.to_string(), reason, code, now.to_rfc3339(), id.to_string()],
440            ).map_err(map_db_error)?;
441        }
442
443        self.get(id)?.ok_or(CommerceError::NotFound)
444    }
445
446    fn cancel(&self, id: PaymentId) -> Result<Payment> {
447        self.update(
448            id,
449            UpdatePayment {
450                status: Some(PaymentTransactionStatus::Cancelled),
451                ..Default::default()
452            },
453        )
454    }
455
456    fn create_refund(&self, input: CreateRefund) -> Result<Refund> {
457        if let Some(key) = input.idempotency_key.as_deref() {
458            if let Some(existing) = self.get_refund_by_idempotency_key(key)? {
459                return Ok(existing);
460            }
461        }
462
463        let id = Uuid::new_v4();
464        let now = chrono::Utc::now();
465        let refund_number = generate_refund_number();
466        let payment_id = input.payment_id;
467
468        // The read of the payment, the over-refund validation, and the refund
469        // INSERT all run inside ONE `IMMEDIATE` transaction. IMMEDIATE acquires
470        // the database write lock up front, so a concurrent `create_refund` for
471        // the same payment is serialized rather than racing: each caller sees
472        // the other's freshly-inserted in-flight refund and cannot both pass the
473        // remaining-balance check. (Previously the read+validate happened on a
474        // separate, lock-free connection from the INSERT, so two callers could
475        // each validate against the same stale balance and together over-refund
476        // the payment once both were completed.)
477        //
478        // Domain failures (`NotFound`, `validate_refund`'s `ValidationError`)
479        // are smuggled out of the closure as `ToSqlConversionFailure(CommerceError)`
480        // and unwrapped back to their original variants by `map_db_error`.
481        with_immediate_transaction(&self.pool, |tx| {
482            let mut payment = tx
483                .query_row(
484                    "SELECT * FROM payments WHERE id = ?",
485                    [payment_id.to_string()],
486                    Self::row_to_payment,
487                )
488                .map_err(|e| match e {
489                    rusqlite::Error::QueryReturnedNoRows => {
490                        rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::NotFound))
491                    }
492                    other => other,
493                })?;
494
495            // Reserve against in-flight (non-terminal) refunds as well as the
496            // already-committed `amount_refunded`. A `Pending`/`Processing`
497            // refund has not yet folded its amount into `amount_refunded`, but
498            // it WILL once completed, so it must count against the remaining
499            // refundable balance to prevent concurrent over-refund. `Failed` /
500            // `Cancelled` refunds release their reservation and are excluded.
501            //
502            // `amount` is a TEXT column, so the amounts are read as text and
503            // summed with `rust_decimal::Decimal` in Rust. Doing `SUM(amount)`
504            // in SQL would coerce the TEXT values to IEEE-754 floats (the same
505            // money-precision defect avoided on the `complete_refund` write
506            // path).
507            let mut in_flight = rust_decimal::Decimal::ZERO;
508            {
509                let mut stmt = tx.prepare(
510                    "SELECT amount FROM refunds \
511                     WHERE payment_id = ? AND status IN ('pending', 'processing')",
512                )?;
513                let rows = stmt.query_map([payment_id.to_string()], |row| {
514                    let amount: String = row.get(0)?;
515                    parse_decimal_row(&amount, "refund", "amount")
516                })?;
517                for row in rows {
518                    in_flight += row?;
519                }
520            }
521
522            // Fold the in-flight reservation into the payment's refunded total so
523            // the unmodified `validate_refund` guard sees the true remaining
524            // balance. `validate_refund` still owns all of the rules (refundable
525            // status, positive amount, not exceeding remaining).
526            payment.amount_refunded += in_flight;
527            let refund_amount = payment
528                .validate_refund(input.amount)
529                .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
530
531            tx.execute(
532                "INSERT INTO refunds (id, refund_number, payment_id, status, amount, currency, reason, external_id, idempotency_key, notes, created_at, updated_at)
533                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
534                params![
535                    id.to_string(),
536                    refund_number,
537                    payment_id.to_string(),
538                    RefundStatus::Pending.to_string(),
539                    refund_amount.to_string(),
540                    payment.currency,
541                    input.reason,
542                    input.external_id,
543                    input.idempotency_key,
544                    input.notes,
545                    now.to_rfc3339(),
546                    now.to_rfc3339(),
547                ],
548            )?;
549
550            Ok(())
551        })?;
552
553        self.get_refund(id)?.ok_or(CommerceError::NotFound)
554    }
555
556    fn get_refund(&self, id: Uuid) -> Result<Option<Refund>> {
557        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
558        let mut stmt = conn.prepare("SELECT * FROM refunds WHERE id = ?").map_err(map_db_error)?;
559        let result = stmt.query_row([id.to_string()], Self::row_to_refund);
560        match result {
561            Ok(refund) => Ok(Some(refund)),
562            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
563            Err(e) => Err(map_db_error(e)),
564        }
565    }
566
567    fn get_refunds(&self, payment_id: PaymentId) -> Result<Vec<Refund>> {
568        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
569        let mut stmt = conn
570            .prepare("SELECT * FROM refunds WHERE payment_id = ? ORDER BY created_at DESC")
571            .map_err(map_db_error)?;
572        let rows =
573            stmt.query_map([payment_id.to_string()], Self::row_to_refund).map_err(map_db_error)?;
574
575        let mut refunds = Vec::new();
576        for row in rows {
577            refunds.push(row.map_err(map_db_error)?);
578        }
579        Ok(refunds)
580    }
581
582    fn complete_refund(&self, id: Uuid) -> Result<Refund> {
583        let refund = self.get_refund(id)?.ok_or(CommerceError::NotFound)?;
584        let now = chrono::Utc::now();
585
586        with_immediate_transaction(&self.pool, |tx| {
587            // Read the CURRENT status inside the write transaction so concurrent
588            // completions serialize on the IMMEDIATE lock and only one of them
589            // folds the refund into the payment.
590            let current_status: RefundStatus = parse_enum_row(
591                &tx.query_row(
592                    "SELECT status FROM refunds WHERE id = ?",
593                    params![id.to_string()],
594                    |row| row.get::<_, String>(0),
595                )?,
596                "refund",
597                "status",
598            )?;
599
600            // Idempotent: completing an already-completed refund is a no-op (a
601            // duplicated payment-processor webhook or a retry must NOT re-add the
602            // amount to the payment's `amount_refunded`).
603            if current_status == RefundStatus::Completed {
604                return Ok(());
605            }
606            // A failed/cancelled refund is terminal and cannot be completed.
607            if current_status.is_terminal() {
608                return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
609                    CommerceError::ValidationError(format!(
610                        "Cannot complete a {current_status} refund"
611                    )),
612                )));
613            }
614
615            tx.execute(
616                "UPDATE refunds SET status = ?, refunded_at = ?, updated_at = ? WHERE id = ?",
617                params![
618                    RefundStatus::Completed.to_string(),
619                    now.to_rfc3339(),
620                    now.to_rfc3339(),
621                    id.to_string()
622                ],
623            )?;
624
625            // Update payment amount_refunded.
626            //
627            // `amount`/`amount_refunded` are TEXT columns (migration 006), so
628            // doing the addition or comparison in SQL would coerce the values
629            // to IEEE-754 floats (e.g. '0.10' + '0.20' = 0.30000000000000004,
630            // and the `>= amount` status comparison would be wrong). Instead we
631            // read the current values, compute the new balance and status with
632            // `rust_decimal::Decimal` in Rust, and write the precomputed TEXT
633            // values back as bound parameters.
634            let (current_refunded, payment_amount): (String, String) = tx.query_row(
635                "SELECT amount_refunded, amount FROM payments WHERE id = ?",
636                params![refund.payment_id.to_string()],
637                |row| Ok((row.get(0)?, row.get(1)?)),
638            )?;
639            let current_refunded =
640                parse_decimal_row(&current_refunded, "payment", "amount_refunded")?;
641            let payment_amount = parse_decimal_row(&payment_amount, "payment", "amount")?;
642
643            let new_refunded = current_refunded + refund.amount;
644            let new_status = if new_refunded >= payment_amount {
645                PaymentTransactionStatus::Refunded
646            } else {
647                PaymentTransactionStatus::PartiallyRefunded
648            };
649
650            tx.execute(
651                "UPDATE payments SET amount_refunded = ?, status = ?, updated_at = ? WHERE id = ?",
652                params![
653                    new_refunded.to_string(),
654                    new_status.to_string(),
655                    now.to_rfc3339(),
656                    refund.payment_id.to_string()
657                ],
658            )?;
659
660            Ok(())
661        })?;
662
663        self.get_refund(id)?.ok_or(CommerceError::NotFound)
664    }
665
666    fn fail_refund(&self, id: Uuid, reason: &str) -> Result<Refund> {
667        let now = chrono::Utc::now();
668
669        {
670            let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
671            conn.execute(
672                "UPDATE refunds SET status = ?, failure_reason = ?, updated_at = ? WHERE id = ?",
673                params![RefundStatus::Failed.to_string(), reason, now.to_rfc3339(), id.to_string()],
674            )
675            .map_err(map_db_error)?;
676        }
677
678        self.get_refund(id)?.ok_or(CommerceError::NotFound)
679    }
680
681    fn create_payment_method(&self, input: CreatePaymentMethod) -> Result<PaymentMethod> {
682        let id = Uuid::new_v4();
683        let now = chrono::Utc::now();
684
685        with_immediate_transaction(&self.pool, |tx| {
686            // If setting as default, clear existing default
687            if input.is_default.unwrap_or(false) {
688                tx.execute(
689                    "UPDATE payment_methods SET is_default = 0 WHERE customer_id = ?",
690                    [input.customer_id.to_string()],
691                )?;
692            }
693
694            tx.execute(
695                "INSERT INTO payment_methods (id, customer_id, method_type, is_default, card_brand,
696                 card_last4, card_exp_month, card_exp_year, cardholder_name, bank_name, account_last4,
697                 external_id, billing_address, created_at, updated_at)
698                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
699                params![
700                    id.to_string(),
701                    input.customer_id.to_string(),
702                    input.method_type.to_string(),
703                    i32::from(input.is_default.unwrap_or(false)),
704                    input.card_brand.map(|b| b.to_string()),
705                    input.card_last4,
706                    input.card_exp_month,
707                    input.card_exp_year,
708                    input.cardholder_name,
709                    input.bank_name,
710                    input.account_last4,
711                    input.external_id,
712                    input.billing_address,
713                    now.to_rfc3339(),
714                    now.to_rfc3339(),
715                ],
716            )?;
717
718            Ok(())
719        })?;
720
721        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
722        let mut stmt =
723            conn.prepare("SELECT * FROM payment_methods WHERE id = ?").map_err(map_db_error)?;
724        stmt.query_row([id.to_string()], Self::row_to_payment_method).map_err(map_db_error)
725    }
726
727    fn get_payment_methods(&self, customer_id: CustomerId) -> Result<Vec<PaymentMethod>> {
728        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
729        let mut stmt = conn.prepare("SELECT * FROM payment_methods WHERE customer_id = ? ORDER BY is_default DESC, created_at DESC").map_err(map_db_error)?;
730        let rows = stmt
731            .query_map([customer_id.to_string()], Self::row_to_payment_method)
732            .map_err(map_db_error)?;
733
734        let mut methods = Vec::new();
735        for row in rows {
736            methods.push(row.map_err(map_db_error)?);
737        }
738        Ok(methods)
739    }
740
741    fn delete_payment_method(&self, id: Uuid) -> Result<()> {
742        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
743        conn.execute("DELETE FROM payment_methods WHERE id = ?", [id.to_string()])
744            .map_err(map_db_error)?;
745        Ok(())
746    }
747
748    fn set_default_payment_method(&self, customer_id: CustomerId, method_id: Uuid) -> Result<()> {
749        with_immediate_transaction(&self.pool, |tx| {
750            tx.execute(
751                "UPDATE payment_methods SET is_default = 0 WHERE customer_id = ?",
752                [customer_id.to_string()],
753            )?;
754
755            tx.execute(
756                "UPDATE payment_methods SET is_default = 1 WHERE id = ? AND customer_id = ?",
757                params![method_id.to_string(), customer_id.to_string()],
758            )?;
759
760            Ok(())
761        })?;
762
763        Ok(())
764    }
765
766    fn count(&self, filter: PaymentFilter) -> Result<u64> {
767        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
768
769        let mut sql = "SELECT COUNT(*) FROM payments WHERE 1=1".to_string();
770        let mut params_vec: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
771
772        if let Some(order_id) = &filter.order_id {
773            sql.push_str(" AND order_id = ?");
774            params_vec.push(Box::new(order_id.to_string()));
775        }
776        if let Some(invoice_id) = &filter.invoice_id {
777            sql.push_str(" AND invoice_id = ?");
778            params_vec.push(Box::new(invoice_id.to_string()));
779        }
780        if let Some(customer_id) = &filter.customer_id {
781            sql.push_str(" AND customer_id = ?");
782            params_vec.push(Box::new(customer_id.to_string()));
783        }
784        if let Some(status) = &filter.status {
785            sql.push_str(" AND status = ?");
786            params_vec.push(Box::new(status.to_string()));
787        }
788
789        let params_refs: Vec<&dyn rusqlite::ToSql> =
790            params_vec.iter().map(std::convert::AsRef::as_ref).collect();
791        let count: i64 =
792            conn.query_row(&sql, params_refs.as_slice(), |row| row.get(0)).map_err(map_db_error)?;
793        Ok(count as u64)
794    }
795
796    // === Batch Operations ===
797
798    fn create_batch(&self, inputs: Vec<CreatePayment>) -> Result<BatchResult<Payment>> {
799        validate_batch_size(&inputs)?;
800        let mut result = BatchResult::with_capacity(inputs.len());
801
802        for (index, input) in inputs.into_iter().enumerate() {
803            match self.create(input) {
804                Ok(payment) => result.record_success(payment),
805                Err(e) => result.record_failure(index, None, &e),
806            }
807        }
808
809        Ok(result)
810    }
811
812    fn create_batch_atomic(&self, inputs: Vec<CreatePayment>) -> Result<Vec<Payment>> {
813        validate_batch_size(&inputs)?;
814        if inputs.is_empty() {
815            return Ok(vec![]);
816        }
817
818        let mut conn = self.conn()?;
819        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
820        let mut results = Vec::with_capacity(inputs.len());
821
822        for input in inputs {
823            let id = Uuid::new_v4();
824            let now = chrono::Utc::now();
825            let payment_number = generate_payment_number();
826
827            tx.execute(
828                "INSERT INTO payments (id, payment_number, order_id, invoice_id, customer_id, status,
829                 payment_method, amount, currency, amount_refunded, external_id, idempotency_key, processor,
830                 card_brand, card_last4, card_exp_month, card_exp_year, billing_email, billing_name,
831                 billing_address, description, metadata, created_at, updated_at)
832                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
833                params![
834                    id.to_string(),
835                    payment_number.clone(),
836                    input.order_id.map(|id| id.to_string()),
837                    input.invoice_id.map(|id| id.to_string()),
838                    input.customer_id.map(|id| id.to_string()),
839                    PaymentTransactionStatus::Pending.to_string(),
840                    input.payment_method.to_string(),
841                    input.amount.to_string(),
842                    input.currency.unwrap_or_default(),
843                    "0",
844                    input.external_id.clone(),
845                    input.idempotency_key.clone(),
846                    input.processor.clone(),
847                    input.card_brand.map(|b| b.to_string()),
848                    input.card_last4.clone(),
849                    input.card_exp_month,
850                    input.card_exp_year,
851                    input.billing_email.clone(),
852                    input.billing_name.clone(),
853                    input.billing_address.clone(),
854                    input.description.clone(),
855                    input.metadata.clone(),
856                    now.to_rfc3339(),
857                    now.to_rfc3339(),
858                ],
859            ).map_err(map_db_error)?;
860
861            results.push(Payment {
862                id: PaymentId::from(id),
863                payment_number,
864                order_id: input.order_id,
865                invoice_id: input.invoice_id,
866                customer_id: input.customer_id,
867                status: PaymentTransactionStatus::Pending,
868                payment_method: input.payment_method,
869                amount: input.amount,
870                currency: input.currency.unwrap_or_default(),
871                amount_refunded: rust_decimal::Decimal::ZERO,
872                external_id: input.external_id,
873                idempotency_key: input.idempotency_key,
874                processor: input.processor,
875                card_brand: input.card_brand,
876                card_last4: input.card_last4,
877                card_exp_month: input.card_exp_month,
878                card_exp_year: input.card_exp_year,
879                // Blockchain/Stablecoin fields
880                blockchain_network: input.blockchain_network,
881                stablecoin_type: input.stablecoin_type,
882                from_wallet_address: input.from_wallet_address,
883                to_wallet_address: input.to_wallet_address,
884                tx_hash: None,
885                block_number: None,
886                confirmations: None,
887                token_address: input.token_address,
888                ves_intent_id: None,
889                billing_email: input.billing_email,
890                billing_name: input.billing_name,
891                billing_address: input.billing_address,
892                description: input.description,
893                failure_reason: None,
894                failure_code: None,
895                metadata: input.metadata,
896                paid_at: None,
897                version: 1,
898                created_at: now,
899                updated_at: now,
900            });
901        }
902
903        tx.commit().map_err(map_db_error)?;
904        Ok(results)
905    }
906
907    fn update_batch(
908        &self,
909        updates: Vec<(PaymentId, UpdatePayment)>,
910    ) -> Result<BatchResult<Payment>> {
911        validate_batch_size(&updates)?;
912        let mut result = BatchResult::with_capacity(updates.len());
913
914        for (index, (id, input)) in updates.into_iter().enumerate() {
915            match self.update(id, input) {
916                Ok(payment) => result.record_success(payment),
917                Err(e) => result.record_failure(index, Some(id.to_string()), &e),
918            }
919        }
920
921        Ok(result)
922    }
923
924    fn update_batch_atomic(
925        &self,
926        updates: Vec<(PaymentId, UpdatePayment)>,
927    ) -> Result<Vec<Payment>> {
928        validate_batch_size(&updates)?;
929        if updates.is_empty() {
930            return Ok(vec![]);
931        }
932
933        let mut conn = self.conn()?;
934        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
935        let mut results = Vec::with_capacity(updates.len());
936
937        for (id, input) in updates {
938            let now = chrono::Utc::now();
939
940            // Get existing payment to merge with updates
941            let payment: Payment = tx
942                .query_row(
943                    "SELECT * FROM payments WHERE id = ?",
944                    [id.to_string()],
945                    Self::row_to_payment,
946                )
947                .map_err(|e| match e {
948                    rusqlite::Error::QueryReturnedNoRows => CommerceError::NotFound,
949                    e => map_db_error(e),
950                })?;
951
952            tx.execute(
953                "UPDATE payments SET status = ?, external_id = ?, failure_reason = ?,
954                 failure_code = ?, metadata = ?, updated_at = ? WHERE id = ?",
955                params![
956                    input.status.unwrap_or(payment.status).to_string(),
957                    input.external_id.or(payment.external_id),
958                    input.failure_reason.or(payment.failure_reason),
959                    input.failure_code.or(payment.failure_code),
960                    input.metadata.or(payment.metadata),
961                    now.to_rfc3339(),
962                    id.to_string(),
963                ],
964            )
965            .map_err(map_db_error)?;
966
967            // Fetch the updated payment
968            let updated_payment = tx
969                .query_row(
970                    "SELECT * FROM payments WHERE id = ?",
971                    [id.to_string()],
972                    Self::row_to_payment,
973                )
974                .map_err(map_db_error)?;
975
976            results.push(updated_payment);
977        }
978
979        tx.commit().map_err(map_db_error)?;
980        Ok(results)
981    }
982
983    fn delete_batch(&self, ids: Vec<PaymentId>) -> Result<BatchResult<Uuid>> {
984        validate_batch_size(&ids)?;
985        let mut result = BatchResult::with_capacity(ids.len());
986
987        for (index, id) in ids.into_iter().enumerate() {
988            let raw_id: Uuid = id.into();
989            let conn = self.conn()?;
990            match conn.execute("DELETE FROM payments WHERE id = ?", [id.to_string()]) {
991                Ok(rows) if rows > 0 => result.record_success(raw_id),
992                Ok(_) => {
993                    result.record_failure(index, Some(id.to_string()), &CommerceError::NotFound);
994                }
995                Err(e) => result.record_failure(index, Some(id.to_string()), &map_db_error(e)),
996            }
997        }
998
999        Ok(result)
1000    }
1001
1002    fn delete_batch_atomic(&self, ids: Vec<PaymentId>) -> Result<()> {
1003        validate_batch_size(&ids)?;
1004        if ids.is_empty() {
1005            return Ok(());
1006        }
1007
1008        let mut conn = self.conn()?;
1009        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
1010
1011        let raw_ids: Vec<Uuid> = ids.iter().map(|id| (*id).into()).collect();
1012        let placeholders = build_in_clause(ids.len());
1013        let params = uuid_params(&raw_ids);
1014        let params_refs = params_refs(&params);
1015
1016        // Delete refunds associated with these payments first
1017        let sql = format!("DELETE FROM refunds WHERE payment_id IN ({placeholders})");
1018        tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
1019
1020        // Delete payments
1021        let sql = format!("DELETE FROM payments WHERE id IN ({placeholders})");
1022        tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
1023
1024        tx.commit().map_err(map_db_error)?;
1025        Ok(())
1026    }
1027
1028    fn get_batch(&self, ids: Vec<PaymentId>) -> Result<Vec<Payment>> {
1029        validate_batch_size(&ids)?;
1030        if ids.is_empty() {
1031            return Ok(vec![]);
1032        }
1033
1034        let conn = self.conn()?;
1035        let raw_ids: Vec<Uuid> = ids.iter().map(|id| (*id).into()).collect();
1036        let placeholders = build_in_clause(ids.len());
1037        let sql = format!("SELECT * FROM payments WHERE id IN ({placeholders})");
1038
1039        let params = uuid_params(&raw_ids);
1040        let params_refs = params_refs(&params);
1041
1042        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
1043        let payments = stmt
1044            .query_map(params_refs.as_slice(), Self::row_to_payment)
1045            .map_err(map_db_error)?
1046            .collect::<rusqlite::Result<Vec<_>>>()
1047            .map_err(map_db_error)?;
1048
1049        Ok(payments)
1050    }
1051}
stateset_db/sqlite/payments.rs

stateset_db/sqlite/
payments.rs
