libeufin

db.rs (5547B)

---

 /*
   This file is part of TALER
   Copyright (C) 2026 Taler Systems SA
 
   TALER is free software; you can redistribute it and/or modify it under the
   terms of the GNU Affero General Public License as published by the Free Software
   Foundation; either version 3, or (at your option) any later version.
 
   TALER is distributed in the hope that it will be useful, but WITHOUT ANY
   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
   A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more details.
 
   You should have received a copy of the GNU Affero General Public License along with
   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
 */
 
 use sqlx::PgPool;
 use taler_api::config::DbCfg;
 use tokio::sync::watch::Sender;
 
 pub mod exchange;
 pub mod initiated;
 pub mod list;
 pub mod payment;
 pub mod transfer;
 
 const SCHEMA: &str = "libeufin_nexus";
 pub const UNSETTLED: &str = "status NOT IN ('success', 'permanent_failure', 'late_failure')";
 pub const PENDING: &str = "status IN ('unsubmitted', 'pending')";
 
 pub async fn pool(cfg: &DbCfg) -> anyhow::Result<PgPool> {
     let pool = taler_common::db::pool(cfg.cfg.clone(), SCHEMA).await?;
     Ok(pool)
 }
 
 pub async fn dbinit(cfg: &DbCfg, reset: bool) -> anyhow::Result<PgPool> {
     let pool = taler_common::db::pool(cfg.cfg.clone(), SCHEMA).await?;
     let mut db = pool.acquire().await?;
     taler_common::db::dbinit(&mut db, cfg.sql_dir.as_ref(), "libeufin-nexus", reset).await?;
     Ok(pool)
 }
 
 pub async fn notification_listener(
     pool: PgPool,
     in_channel: Sender<i64>,
     taler_in_channel: Sender<i64>,
     taler_out_channel: Sender<i64>,
 ) -> sqlx::Result<()> {
     taler_api::notification::notification_listener!(&pool,
         "nexus_revenue_tx" => (row_id: i64) {
             in_channel.send_replace(row_id);
         },
         "nexus_incoming_tx" => (row_id: i64) {
             taler_in_channel.send_replace(row_id);
         },
         "nexus_outgoing_tx" => (row_id: i64) {
             taler_out_channel.send_replace(row_id);
         }
     )
 }
 
 #[cfg(test)]
 pub mod test {
     use libeufin_ebics::db::test::ebics_routine;
     use sqlx::{PgPool, Row, postgres::PgRow};
     use taler_api::db::TypeHelper;
     use taler_common::db::IncomingType;
     use taler_macros::db_test;
     use taler_test_utils::routine::Status;
 
     pub async fn check_count(db: &PgPool, nb_tx: usize, nb_bounce: usize) {
         assert_eq!(
             sqlx::query_as::<_, (i64, i64)>(
                 "SELECT (SELECT count(*) FROM incoming_transactions) + (SELECT count(*) FROM outgoing_transactions), (SELECT count(*) FROM bounced_transactions)",
             )
             .fetch_one(db)
             .await
             .unwrap(),
             (nb_tx as i64, nb_bounce as i64)
         );
     }
 
     pub async fn check_in_count(
         db: &PgPool,
         nb_incoming: usize,
         nb_bounce: usize,
         nb_talerable: usize,
     ) {
         assert_eq!(
             sqlx::query_as::<_, (i64, i64, i64)>(
                 "
             SELECT (SELECT count(*) FROM incoming_transactions),
                (SELECT count(*) FROM bounced_transactions),
                (SELECT count(*) FROM talerable_incoming_transactions)
                ",
             )
             .fetch_one(db)
             .await
             .unwrap(),
             (nb_incoming as i64, nb_bounce as i64, nb_talerable as i64)
         );
     }
 
     pub async fn check_out_count(db: &PgPool, nb_outgoing: u64, nb_talerable: u64) {
         assert_eq!(
             sqlx::query_as::<_, (i64, i64)>(
                 "
                 SELECT (SELECT count(*) FROM outgoing_transactions),
                 (SELECT count(*) FROM talerable_outgoing_transactions)
                ",
             )
             .fetch_one(db)
             .await
             .unwrap(),
             (nb_outgoing as i64, nb_talerable as i64)
         );
     }
 
     pub async fn check_in(db: &PgPool) -> Vec<Status> {
         sqlx::query(
             "
             SELECT pending_recurrent_incoming_transactions.authorization_pub IS NOT NULL, initiated_outgoing_transaction_id IS NOT NULL, debit_payto IS NULL OR subject IS NULL, type::text, metadata
             FROM incoming_transactions
                 LEFT JOIN talerable_incoming_transactions USING (incoming_transaction_id)
                 LEFT JOIN pending_recurrent_incoming_transactions USING (incoming_transaction_id)
                 LEFT JOIN bounced_transactions USING (incoming_transaction_id)
             ORDER BY incoming_transaction_id
         ",
         )
         .try_map(|r: PgRow| {
             Ok(
                 if r.try_get_flag(0)? {
                     Status::Pending
                 } else if r.try_get_flag(1)? {
                     Status::Bounced
                 } else if r.try_get_flag(2)? {
                     Status::Incomplete
                 } else {
                     match r.try_get_opt_parse(3)? {
                         None => Status::Simple,
                         Some(IncomingType::reserve) => Status::Reserve(r.try_get(4)?),
                         Some(IncomingType::kyc) => Status::Kyc(r.try_get(4)?),
                         Some(e) => unreachable!("{e:?}")
                     }
                 }
             )
         })
         .fetch_all(db)
         .await
         .unwrap()
     }
 
     pub async fn check_in_state(db: &PgPool, state: &[Status]) {
         assert_eq!(state, check_in(db).await);
     }
 
     #[db_test]
     pub async fn ebics(db: PgPool) {
         ebics_routine(&db).await;
     }
 }