Payment service: fixed computing payment amounts

1. It was previously falsely assumed that total_amount_accepted for an activity or agreement is equal to the total amount of scheduled payments. That is not true in case of accepting non-binding debit notes (i.e. without due date) which increases total_amount_accepted but doesn't trigger payment. This was fixed by introducing total_amount_scheduled field to both activity and agreement model. This field is updated when a payment is scheduled and it is used to compute the payment amount. 2. Allocation's remaining amount was being updated when a payment was confirmed not when it was scheduled, thus allowing double-spending from an allocation if a new payment was scheduled before the previous one got confirmed. Now spent & remaining amounts are updated when scheduling payment to prevent double-spending. 3. Paying zero-amount invoices was causing unnecessary transaction costs. A quick fix has been applied not to schedule payments for zero amount. Although, a proper solution that disallows issuing such invoices in the first place is required. Signed-off-by: Adam Wierzbicki <[email protected]>
golemfactory · Jan 28, 2021 · 4d4fa94 · 4d4fa94
1 parent e794a09
commit 4d4fa94
Show file tree

Hide file tree

Showing 13 changed files with 186 additions and 25 deletions.
diff --git a/core/model/src/payment.rs b/core/model/src/payment.rs
@@ -59,8 +59,15 @@ pub mod local {
     }
 
     impl SchedulePayment {
-        pub fn from_invoice(invoice: Invoice, allocation_id: String, amount: BigDecimal) -> Self {
-            Self {
+        pub fn from_invoice(
+            invoice: Invoice,
+            allocation_id: String,
+            amount: BigDecimal,
+        ) -> Option<Self> {
+            if amount <= BigDecimal::zero() {
+                return None;
+            }
+            Some(Self {
                 title: PaymentTitle::Invoice(InvoicePayment {
                     invoice_id: invoice.invoice_id,
                     agreement_id: invoice.agreement_id,
@@ -73,14 +80,17 @@ pub mod local {
                 allocation_id,
                 amount,
                 due_date: invoice.payment_due_date,
-            }
+            })
         }
 
         pub fn from_debit_note(
             debit_note: DebitNote,
             allocation_id: String,
             amount: BigDecimal,
         ) -> Option<Self> {
+            if amount <= BigDecimal::zero() {
+                return None;
+            }
             debit_note.payment_due_date.map(|due_date| Self {
                 title: PaymentTitle::DebitNote(DebitNotePayment {
                     debit_note_id: debit_note.debit_note_id,

diff --git a/core/payment/migrations/2021-01-28-102818_scheduled_amount/down.sql b/core/payment/migrations/2021-01-28-102818_scheduled_amount/down.sql
@@ -0,0 +1,46 @@
+-- HACK: removing column 'total_amount_scheduled'
+
+PRAGMA foreign_keys=off;
+
+CREATE TABLE pay_agreement_tmp(
+    id VARCHAR(50) NOT NULL,
+    owner_id VARCHAR(50) NOT NULL,
+    role CHAR(1) NOT NULL CHECK (role in ('R', 'P')),
+    peer_id VARCHAR(50) NOT NULL,
+    payee_addr VARCHAR(50) NOT NULL,
+    payer_addr VARCHAR(50) NOT NULL,
+    payment_platform VARCHAR(50) NOT NULL,
+    total_amount_due VARCHAR(32) NOT NULL,
+    total_amount_accepted VARCHAR(32) NOT NULL,
+    total_amount_paid VARCHAR(32) NOT NULL,
+    app_session_id VARCHAR(50) NULL,
+    PRIMARY KEY (owner_id, id),
+    UNIQUE (id, role)
+);
+
+CREATE TABLE pay_activity_tmp(
+    id VARCHAR(50) NOT NULL,
+    owner_id VARCHAR(50) NOT NULL,
+    role CHAR(1) NOT NULL CHECK (role in ('R', 'P')),
+    agreement_id VARCHAR(50) NOT NULL,
+    total_amount_due VARCHAR(32) NOT NULL,
+    total_amount_accepted VARCHAR(32) NOT NULL,
+    total_amount_paid VARCHAR(32) NOT NULL,
+    PRIMARY KEY(owner_id, id),
+    UNIQUE (id, role),
+    FOREIGN KEY(owner_id, agreement_id) REFERENCES pay_agreement (owner_id, id)
+);
+
+INSERT INTO pay_agreement_tmp(id, owner_id, role, peer_id, payee_addr, payer_addr, payment_platform, total_amount_due, total_amount_accepted, total_amount_paid, app_session_id)
+SELECT id, owner_id, role, peer_id, payee_addr, payer_addr, payment_platform, total_amount_due, total_amount_accepted, total_amount_paid, app_session_id FROM pay_agreement;
+
+INSERT INTO pay_activity_tmp(id, owner_id, role, agreement_id, total_amount_due, total_amount_accepted, total_amount_paid)
+SELECT id, owner_id, role, agreement_id, total_amount_due, total_amount_accepted, total_amount_paid FROM pay_activity;
+
+DROP TABLE pay_agreement;
+DROP TABLE pay_activity;
+
+ALTER TABLE pay_agreement_tmp RENAME TO pay_agreement;
+ALTER TABLE pay_activity_tmp RENAME TO pay_activity;
+
+PRAGMA foreign_keys=on;
diff --git a/core/payment/migrations/2021-01-28-102818_scheduled_amount/up.sql b/core/payment/migrations/2021-01-28-102818_scheduled_amount/up.sql
@@ -0,0 +1,48 @@
+-- HACK: Adding not-null column 'total_amount_scheduled' without default value
+
+PRAGMA foreign_keys=off;
+
+CREATE TABLE pay_agreement_tmp(
+    id VARCHAR(50) NOT NULL,
+    owner_id VARCHAR(50) NOT NULL,
+    role CHAR(1) NOT NULL CHECK (role in ('R', 'P')),
+    peer_id VARCHAR(50) NOT NULL,
+    payee_addr VARCHAR(50) NOT NULL,
+    payer_addr VARCHAR(50) NOT NULL,
+    payment_platform VARCHAR(50) NOT NULL,
+    total_amount_due VARCHAR(32) NOT NULL,
+    total_amount_accepted VARCHAR(32) NOT NULL,
+    total_amount_scheduled VARCHAR(32) NOT NULL,
+    total_amount_paid VARCHAR(32) NOT NULL,
+    app_session_id VARCHAR(50) NULL,
+    PRIMARY KEY (owner_id, id),
+    UNIQUE (id, role)
+);
+
+CREATE TABLE pay_activity_tmp(
+    id VARCHAR(50) NOT NULL,
+    owner_id VARCHAR(50) NOT NULL,
+    role CHAR(1) NOT NULL CHECK (role in ('R', 'P')),
+    agreement_id VARCHAR(50) NOT NULL,
+    total_amount_due VARCHAR(32) NOT NULL,
+    total_amount_accepted VARCHAR(32) NOT NULL,
+    total_amount_scheduled VARCHAR(32) NOT NULL,
+    total_amount_paid VARCHAR(32) NOT NULL,
+    PRIMARY KEY(owner_id, id),
+    UNIQUE (id, role),
+    FOREIGN KEY(owner_id, agreement_id) REFERENCES pay_agreement (owner_id, id)
+);
+
+INSERT INTO pay_agreement_tmp(id, owner_id, role, peer_id, payee_addr, payer_addr, payment_platform, total_amount_due, total_amount_accepted, total_amount_scheduled, total_amount_paid, app_session_id)
+SELECT id, owner_id, role, peer_id, payee_addr, payer_addr, payment_platform, total_amount_due, total_amount_accepted, total_amount_accepted AS total_amount_scheduled, total_amount_paid, app_session_id FROM pay_agreement;
+
+INSERT INTO pay_activity_tmp(id, owner_id, role, agreement_id, total_amount_due, total_amount_accepted, total_amount_scheduled, total_amount_paid)
+SELECT id, owner_id, role, agreement_id, total_amount_due, total_amount_accepted, total_amount_accepted AS total_amount_scheduled, total_amount_paid FROM pay_activity;
+
+DROP TABLE pay_agreement;
+DROP TABLE pay_activity;
+
+ALTER TABLE pay_agreement_tmp RENAME TO pay_agreement;
+ALTER TABLE pay_activity_tmp RENAME TO pay_activity;
+
+PRAGMA foreign_keys=on;
diff --git a/core/payment/src/api/debit_notes.rs b/core/payment/src/api/debit_notes.rs
@@ -269,7 +269,7 @@ async fn accept_debit_note(
         Ok(None) => return response::server_error(&format!("Activity {} not found", activity_id)),
         Err(e) => return response::server_error(&e),
     };
-    let amount_to_pay = &debit_note.total_amount_due - &activity.total_amount_accepted.0;
+    let amount_to_pay = &debit_note.total_amount_due - &activity.total_amount_scheduled.0;
 
     let allocation = match db
         .as_dao::<AllocationDao>()

diff --git a/core/payment/src/api/invoices.rs b/core/payment/src/api/invoices.rs
@@ -314,7 +314,7 @@ async fn accept_invoice(
         }
         Err(e) => return response::server_error(&e),
     };
-    let amount_to_pay = &invoice.amount - &agreement.total_amount_accepted.0;
+    let amount_to_pay = &invoice.amount - &agreement.total_amount_scheduled.0;
 
     let allocation = match db
         .as_dao::<AllocationDao>()
@@ -327,7 +327,6 @@ async fn accept_invoice(
         }
         Err(e) => return response::server_error(&e),
     };
-    // FIXME: remaining amount should be 'locked' until payment is done to avoid double spending
     if amount_to_pay > allocation.remaining_amount {
         let msg = format!(
             "Not enough funds. Allocated: {} Needed: {}",
@@ -347,7 +346,9 @@ async fn accept_invoice(
                 .service(PUBLIC_SERVICE)
                 .call(accept_msg)
                 .await??;
-            bus::service(LOCAL_SERVICE).send(schedule_msg).await??;
+            if let Some(msg) = schedule_msg {
+                bus::service(LOCAL_SERVICE).send(msg).await??;
+            }
             dao.accept(invoice_id, node_id).await?;
 
             counter!("payment.invoices.requestor.accepted", 1);

diff --git a/core/payment/src/dao/activity.rs b/core/payment/src/dao/activity.rs
@@ -56,6 +56,24 @@ pub fn set_amount_accepted(
     agreement::increase_amount_accepted(&agreement_id, owner_id, &amount_delta, conn)
 }
 
+pub fn increase_amount_scheduled(
+    activity_id: &String,
+    owner_id: &NodeId,
+    amount: &BigDecimal,
+    conn: &ConnType,
+) -> DbResult<()> {
+    assert!(amount > &BigDecimal::zero().into()); // TODO: Remove when payment service is production-ready.
+    let activity: WriteObj = dsl::pay_activity
+        .find((activity_id, owner_id))
+        .first(conn)?;
+    let total_amount_scheduled: BigDecimalField =
+        (&activity.total_amount_scheduled.0 + amount).into();
+    diesel::update(&activity)
+        .set(dsl::total_amount_scheduled.eq(total_amount_scheduled))
+        .execute(conn)?;
+    agreement::increase_amount_scheduled(&activity.agreement_id, owner_id, amount, conn)
+}
+
 pub fn increase_amount_paid(
     activity_id: &String,
     owner_id: &NodeId,
@@ -140,6 +158,7 @@ impl<'a> ActivityDao<'a> {
                     dsl::agreement_id,
                     dsl::total_amount_due,
                     dsl::total_amount_accepted,
+                    dsl::total_amount_scheduled,
                     dsl::total_amount_paid,
                     agreement_dsl::peer_id,
                     agreement_dsl::payee_addr,

diff --git a/core/payment/src/dao/agreement.rs b/core/payment/src/dao/agreement.rs
@@ -88,6 +88,24 @@ pub fn increase_amount_accepted(
     Ok(())
 }
 
+pub fn increase_amount_scheduled(
+    agreement_id: &String,
+    owner_id: &NodeId,
+    amount: &BigDecimal,
+    conn: &ConnType,
+) -> DbResult<()> {
+    assert!(amount > &BigDecimal::zero().into()); // TODO: Remove when payment service is production-ready.
+    let agreement: ReadObj = dsl::pay_agreement
+        .find((agreement_id, owner_id))
+        .first(conn)?;
+    let total_amount_scheduled: BigDecimalField =
+        (&agreement.total_amount_scheduled.0 + amount).into();
+    diesel::update(&agreement)
+        .set(dsl::total_amount_scheduled.eq(total_amount_scheduled))
+        .execute(conn)?;
+    Ok(())
+}
+
 pub fn set_amount_accepted(
     agreement_id: &String,
     owner_id: &NodeId,

diff --git a/core/payment/src/dao/order.rs b/core/payment/src/dao/order.rs
@@ -1,3 +1,4 @@
+use crate::dao::{activity, agreement, allocation};
 use crate::error::DbResult;
 use crate::models::order::{ReadObj, WriteObj};
 use crate::schema::pay_debit_note::dsl as debit_note_dsl;
@@ -7,7 +8,9 @@ use diesel::{
     self, BoolExpressionMethods, ExpressionMethods, JoinOnDsl, NullableExpressionMethods, QueryDsl,
     RunQueryDsl,
 };
-use ya_core_model::payment::local::SchedulePayment;
+use ya_core_model::payment::local::{
+    DebitNotePayment, InvoicePayment, PaymentTitle, SchedulePayment,
+};
 use ya_persistence::executor::{do_with_transaction, readonly_transaction, AsDao, PoolType};
 
 pub struct OrderDao<'c> {
@@ -22,8 +25,27 @@ impl<'c> AsDao<'c> for OrderDao<'c> {
 
 impl<'c> OrderDao<'c> {
     pub async fn create(&self, msg: SchedulePayment, id: String, driver: String) -> DbResult<()> {
-        let order = WriteObj::new(msg, id, driver);
         do_with_transaction(self.pool, move |conn| {
+            match &msg.title {
+                PaymentTitle::DebitNote(DebitNotePayment { activity_id, .. }) => {
+                    activity::increase_amount_scheduled(
+                        activity_id,
+                        &msg.payer_id,
+                        &msg.amount,
+                        conn,
+                    )?
+                }
+                PaymentTitle::Invoice(InvoicePayment { agreement_id, .. }) => {
+                    agreement::increase_amount_scheduled(
+                        agreement_id,
+                        &msg.payer_id,
+                        &msg.amount,
+                        conn,
+                    )?
+                }
+            };
+            let order = WriteObj::new(msg, id, driver);
+            allocation::spend_from_allocation(&order.allocation_id, &order.amount, conn)?;
             diesel::insert_into(dsl::pay_order)
                 .values(order)
                 .execute(conn)?;

diff --git a/core/payment/src/dao/payment.rs b/core/payment/src/dao/payment.rs
@@ -1,4 +1,4 @@
-use crate::dao::{activity, agreement, allocation};
+use crate::dao::{activity, agreement};
 use crate::error::DbResult;
 use crate::models::payment::{
     ActivityPayment as DbActivityPayment, AgreementPayment as DbAgreementPayment, ReadObj, WriteObj,
@@ -37,13 +37,6 @@ fn insert_activity_payments(
 
         activity::increase_amount_paid(&activity_payment.activity_id, &owner_id, &amount, conn)?;
 
-        // Update spent & remaining amount for allocation (if applicable)
-        if let Some(allocation_id) = &allocation_id {
-            log::trace!("Updating spent & remaining amount for allocation...");
-            allocation::spend_from_allocation(allocation_id, &amount, conn)?;
-            log::trace!("Allocation updated.");
-        }
-
         diesel::insert_into(activity_pay_dsl::pay_activity_payment)
             .values(DbActivityPayment {
                 payment_id: payment_id.clone(),
@@ -72,13 +65,6 @@ fn insert_agreement_payments(
 
         agreement::increase_amount_paid(&agreement_payment.agreement_id, &owner_id, &amount, conn)?;
 
-        // Update spent & remaining amount for allocation (if applicable)
-        if let Some(allocation_id) = &allocation_id {
-            log::trace!("Updating spent & remaining amount for allocation...");
-            allocation::spend_from_allocation(allocation_id, &amount, conn)?;
-            log::trace!("Allocation updated.");
-        }
-
         diesel::insert_into(agreement_pay_dsl::pay_agreement_payment)
             .values(DbAgreementPayment {
                 payment_id: payment_id.clone(),

diff --git a/core/payment/src/models/activity.rs b/core/payment/src/models/activity.rs
@@ -2,15 +2,17 @@ use crate::schema::pay_activity;
 use ya_client_model::NodeId;
 use ya_persistence::types::{BigDecimalField, Role};
 
-#[derive(Debug, Insertable)]
+#[derive(Debug, Insertable, Queryable, Identifiable)]
 #[table_name = "pay_activity"]
+#[primary_key(id, owner_id)]
 pub struct WriteObj {
     pub id: String,
     pub owner_id: NodeId,
     pub role: Role,
     pub agreement_id: String,
     pub total_amount_due: BigDecimalField,
     pub total_amount_accepted: BigDecimalField,
+    pub total_amount_scheduled: BigDecimalField,
     pub total_amount_paid: BigDecimalField,
 }
 
@@ -23,6 +25,7 @@ impl WriteObj {
             agreement_id,
             total_amount_due: Default::default(),
             total_amount_accepted: Default::default(),
+            total_amount_scheduled: Default::default(),
             total_amount_paid: Default::default(),
         }
     }
@@ -38,6 +41,7 @@ pub struct ReadObj {
     pub agreement_id: String,
     pub total_amount_due: BigDecimalField,
     pub total_amount_accepted: BigDecimalField,
+    pub total_amount_scheduled: BigDecimalField,
     pub total_amount_paid: BigDecimalField,
 
     pub peer_id: NodeId,    // From Agreement

diff --git a/core/payment/src/models/agreement.rs b/core/payment/src/models/agreement.rs
@@ -19,6 +19,7 @@ pub struct WriteObj {
     pub payment_platform: String,
     pub total_amount_due: BigDecimalField,
     pub total_amount_accepted: BigDecimalField,
+    pub total_amount_scheduled: BigDecimalField,
     pub total_amount_paid: BigDecimalField,
     pub app_session_id: Option<String>,
 }
@@ -61,6 +62,7 @@ impl WriteObj {
             payment_platform,
             total_amount_due: Default::default(),
             total_amount_accepted: Default::default(),
+            total_amount_scheduled: Default::default(),
             total_amount_paid: Default::default(),
             app_session_id: agreement.app_session_id,
         }

diff --git a/core/payment/src/processor.rs b/core/payment/src/processor.rs
@@ -322,6 +322,9 @@ impl PaymentProcessor {
         let payer_addr = msg.sender;
         let payee_addr = msg.recipient;
 
+        if msg.order_ids.len() == 0 {
+            return Err(OrderValidationError::new("order_ids is empty").into());
+        }
         let orders = self
             .db_executor
             .as_dao::<OrderDao>()

diff --git a/core/payment/src/schema.rs b/core/payment/src/schema.rs
@@ -6,6 +6,7 @@ table! {
         agreement_id -> Text,
         total_amount_due -> Text,
         total_amount_accepted -> Text,
+        total_amount_scheduled -> Text,
         total_amount_paid -> Text,
     }
 }
@@ -31,6 +32,7 @@ table! {
         payment_platform -> Text,
         total_amount_due -> Text,
         total_amount_accepted -> Text,
+        total_amount_scheduled -> Text,
         total_amount_paid -> Text,
         app_session_id -> Nullable<Text>,
     }