eth-engine: Parse the response from the connector and save the leftovers if required

Author: gakonst

crates/interledger-settlement-engines/src/engines/ethereum_ledger/eth_engine.rs

@@ -35,6 +35,7 @@ use url::Url;
 use uuid::Uuid;
 
 use crate::stores::redis_ethereum_ledger::*;
+use crate::stores::LeftoversStore;
 use crate::{ApiResponse, CreateAccount, SettlementEngine, SettlementEngineApi};
 use interledger_settlement::{Convert, ConvertDetails, Quantity};
 
@@ -100,7 +101,7 @@ pub struct EthereumLedgerSettlementEngineBuilder<'a, S, Si, A> {
 
 impl<'a, S, Si, A> EthereumLedgerSettlementEngineBuilder<'a, S, Si, A>
 where
-    S: EthereumStore<Account = A> + Clone + Send + Sync + 'static,
+    S: EthereumStore<Account = A> + LeftoversStore + Clone + Send + Sync + 'static,
     Si: EthereumLedgerTxSigner + Clone + Send + Sync + 'static,
     A: EthereumAccount + Send + Sync + 'static,
 {
@@ -222,7 +223,7 @@ where
 
 impl<S, Si, A> EthereumLedgerSettlementEngine<S, Si, A>
 where
-    S: EthereumStore<Account = A> + Clone + Send + Sync + 'static,
+    S: EthereumStore<Account = A> + LeftoversStore + Clone + Send + Sync + 'static,
     Si: EthereumLedgerTxSigner + Clone + Send + Sync + 'static,
     A: EthereumAccount + Send + Sync + 'static,
 {
@@ -489,8 +490,13 @@ where
             .push("settlements");
         let client = Client::new();
         debug!("Making POST to {:?} {:?} about {:?}", url, amount, tx_hash);
+        let store = self.store.clone();
         let action = move || {
+            // need to make 2 clones, one to own the store in the function
+            // and one for the retry closure..
+            let store = store.clone();
             let account_id = account_id.clone();
+            let account_id_clone2 = account_id.clone();
             client
                 .post(url.clone())
                 .header("Idempotency-Key", tx_hash.to_string())
@@ -504,11 +510,64 @@ where
                 })
                 .and_then(move |response| {
                     trace!("Accounting system responded with {:?}", response);
-                    if response.status().is_success() {
-                        Ok(())
-                    } else {
-                        Err(())
+                    if !response.status().is_success() {
+                        return Either::A(err(()));
                     }
+                    Either::B(
+                        response
+                            .into_body()
+                            .concat2()
+                            .map_err(|err| {
+                                let err = format!("Couldn't retrieve body {:?}", err);
+                                error!("{}", err);
+                            })
+                            .and_then(move |body| {
+                                // Get the amount stored by the connector and
+                                // check for any precision loss / overflow
+                                serde_json::from_slice::<Quantity>(&body).map_err(|err| {
+                                    let err = format!(
+                                        "Couldn't parse body {:?} into Quantity {:?}",
+                                        body, err
+                                    );
+                                    error!("{}", err);
+                                })
+                            })
+                            .and_then(move |quantity: Quantity| {
+                                // Convert both to BigUInts so we can do arithmetic
+                                join_all(vec![
+                                    result(BigUint::from_str(&quantity.amount)),
+                                    result(BigUint::from_str(&amount.to_string())),
+                                ])
+                                .map_err(|err| {
+                                    let error_msg =
+                                        format!("Error converting to BigUints {:?}", err);
+                                    error!("{:?}", error_msg);
+                                })
+                                .and_then(
+                                    move |amounts: Vec<BigUint>| {
+                                        let connector_amount = amounts[0].clone();
+                                        let engine_amount = amounts[1].clone();
+                                        // Scale the amount settled by the
+                                        // connector back up to our scale
+                                        let scaled_connector_amount = connector_amount
+                                            .normalize_scale(ConvertDetails {
+                                                from: quantity.scale,
+                                                to: engine_scale,
+                                            });
+
+                                        let diff: BigUint = engine_amount - scaled_connector_amount;
+                                        if diff > BigUint::from(0u32) {
+                                            // connector settled less than we
+                                            // instructed it to, so we must save
+                                            // the difference for the leftovers
+                                            Either::A(store.save_leftovers(account_id_clone2, diff))
+                                        } else {
+                                            Either::B(ok(()))
+                                        }
+                                    },
+                                )
+                            }),
+                    )
                 })
         };
         Retry::spawn(
@@ -611,7 +670,7 @@ where
 
 impl<S, Si, A> SettlementEngine for EthereumLedgerSettlementEngine<S, Si, A>
 where
-    S: EthereumStore<Account = A> + Clone + Send + Sync + 'static,
+    S: EthereumStore<Account = A> + LeftoversStore + Clone + Send + Sync + 'static,
     Si: EthereumLedgerTxSigner + Clone + Send + Sync + 'static,
     A: EthereumAccount + Send + Sync + 'static,
 {