refactor unwraps

core/payment/src/error.rs

@@ -112,6 +112,7 @@ pub mod processor {
     use bigdecimal::BigDecimal;
     use std::fmt::Display;
     use tokio::time::error::Elapsed;
+    use ya_client_model::NodeId;
     use ya_core_model::driver::AccountMode;
     use ya_core_model::payment::local::{
         GenericError, ValidateAllocationError as GsbValidateAllocationError,
@@ -172,35 +173,38 @@ pub mod processor {
             Self(e.to_string())
         }
 
-        pub fn platform(order: &Order, platform: &str) -> Result<(), Self> {
+        pub fn platform(order: &Order, platform: &str) -> Result<(NodeId, NodeId), Self> {
             Err(Self(format!(
                 "Invalid platform for payment order {}: {} != {}",
                 order.id, order.payment_platform, platform
             )))
         }
 
-        pub fn payer_addr(order: &Order, payer_addr: &str) -> Result<(), Self> {
+        pub fn payer_addr(order: &Order, payer_addr: &str) -> Result<(NodeId, NodeId), Self> {
             Err(Self(format!(
                 "Invalid payer address for payment order {}: {} != {}",
                 order.id, order.payer_addr, payer_addr
             )))
         }
 
-        pub fn payee_addr(order: &Order, payee_addr: &str) -> Result<(), Self> {
+        pub fn payee_addr(order: &Order, payee_addr: &str) -> Result<(NodeId, NodeId), Self> {
             Err(Self(format!(
                 "Invalid payee address for payment order {}: {} != {}",
                 order.id, order.payee_addr, payee_addr
             )))
         }
 
-        pub fn amount(expected: &BigDecimal, actual: &BigDecimal) -> Result<(), Self> {
+        pub fn amount(
+            expected: &BigDecimal,
+            actual: &BigDecimal,
+        ) -> Result<(NodeId, NodeId), Self> {
             Err(Self(format!(
                 "Invalid payment amount: {} != {}",
                 expected, actual
             )))
         }
 
-        pub fn zero_amount(order: &Order) -> Result<(), Self> {
+        pub fn zero_amount(order: &Order) -> Result<(NodeId, NodeId), Self> {
             Err(Self(format!(
                 "Payment order can not have 0 amount. order_id={}",
                 order.id
@@ -222,6 +226,8 @@ pub mod processor {
         Sign(#[from] ya_core_model::driver::GenericError),
         #[error("Internal timeout")]
         InternalTimeout(#[from] Elapsed),
+        #[error("Critical error when notifying: {0}")]
+        Critical(String),
     }
 
     impl NotifyPaymentError {

core/payment/src/processor.rs

@@ -53,15 +53,39 @@
     payer_addr: &str,
     payee_addr: &str,
     amount: &BigDecimal,
-) -> Result<(), OrderValidationError> {
+) -> Result<(NodeId, NodeId), OrderValidationError> {
     if orders.is_empty() {
         return Err(OrderValidationError::new(
             "orders not found in the database",
         ));
     }
 
+    let mut payer_id: Option<NodeId> = None;
+    let mut payee_id: Option<NodeId> = None;
+
     let mut total_amount = BigDecimal::zero();
     for order in orders.iter() {
+        if let Some(payer_id) = payer_id {
+            if order.payer_id != payer_id {
+                return Err(OrderValidationError::new(format!(
+                    "payer_id mismatch {} vs {}",
+                    order.payer_id, payer_id
+                )));
+            }
+        } else {
+            payer_id = Some(order.payer_id);
+        }
+        if let Some(payee_id) = payee_id {
+            if order.payee_id != payee_id {
+                return Err(OrderValidationError::new(format!(
+                    "payee_id mismatch {} vs {}",
+                    order.payee_id, payee_id
+                )));
+            }
+        } else {
+            payee_id = Some(order.payee_id);
+        }
+
         if order.amount.0 == BigDecimal::zero() {
             return OrderValidationError::zero_amount(order);
         }
@@ -82,7 +106,10 @@
         return OrderValidationError::amount(&total_amount, amount);
     }
 
-    Ok(())
+    Ok((
+        payer_id.ok_or(OrderValidationError::new("payer_id is null"))?,
+        payee_id.ok_or(OrderValidationError::new("payee_id is null"))?,
+    ))
 }
 
 #[derive(Clone, Debug)]
@@ -425,16 +452,18 @@
         let payer_id: NodeId;
         let payee_id: NodeId;
         let payment_id: String;
-        let mut payment: Payment;
-
-        {
+        let mut payment = {
             let db_executor = self.db_executor.timeout_lock(DB_LOCK_TIMEOUT).await?;
 
             let orders = db_executor
                 .as_dao::<OrderDao>()
                 .get_many(msg.order_ids, driver.clone())
                 .await?;
-            validate_orders(
+
+            // FIXME: This is a hack. Payment orders realized by a single transaction are not guaranteed
+            //        to have the same payer and payee IDs. Fixing this requires a major redesign of the
+            //        data model. Payments can no longer by assigned to a single payer and payee.
+            (payer_id, payee_id) = validate_orders(
                 &orders,
                 &payment_platform,
                 &payer_addr,
@@ -461,12 +490,6 @@
                 }
             }
 
-            // FIXME: This is a hack. Payment orders realized by a single transaction are not guaranteed
-            //        to have the same payer and payee IDs. Fixing this requires a major redesign of the
-            //        data model. Payments can no longer by assigned to a single payer and payee.
-            payer_id = orders.get(0).unwrap().payer_id;
-            payee_id = orders.get(0).unwrap().payee_id;
-
             let payment_dao: PaymentDao = db_executor.as_dao();
 
             payment_id = payment_dao
@@ -483,34 +506,35 @@
                 )
                 .await?;
 
+
             let signed_payment = payment_dao
                 .get(payment_id.clone(), payer_id)
                 .await?
                 .unwrap();
             payment = signed_payment.payload;
         }
 
         // Allocation IDs are requestor's private matter and should not be sent to provider
         for agreement_payment in payment.agreement_payments.iter_mut() {
             agreement_payment.allocation_id = None;
         }
         for activity_payment in payment.activity_payments.iter_mut() {
             activity_payment.allocation_id = None;
         }
 
         let signature_canonicalized = driver_endpoint(&driver)
             .send(driver::SignPaymentCanonicalized(payment.clone()))
             .await??;
         let signature = driver_endpoint(&driver)
             .send(driver::SignPayment(payment.clone()))
             .await??;
 
         counter!("payment.amount.sent", ya_metrics::utils::cryptocurrency_to_u64(&msg.amount), "platform" => payment_platform);
         // This is unconditional because at this point the invoice *has been paid*.
         // Whether the provider was correctly notified of this fact is another matter.
         counter!("payment.invoices.requestor.paid", 1);
         let msg = SendPayment::new(payment.clone(), signature);
         let msg_with_bytes = SendSignedPayment::new(payment, signature_canonicalized);
 
         let db_executor = Arc::clone(&self.db_executor);