Add a simple replay protection

This mechanism ensures transactions are unique. They are optional and client generated.

utxo/src/ledger.rs

@@ -215,6 +215,12 @@ where
             )
             .await?;
 
+        if let Some(ref replay_protection) = transaction.replay_protection {
+            batch
+                .store_replay_protection(replay_protection, &transaction.revision.transaction_id)
+                .await?;
+        }
+
         let accounts = transaction.accounts();
 
         for account in accounts.spends {

utxo/src/storage/cache/batch.rs

@@ -66,6 +66,16 @@ where
         Ok(())
     }
 
+    async fn store_replay_protection(
+        &mut self,
+        protection: &str,
+        transaction_id: &TxId,
+    ) -> Result<(), Error> {
+        self.inner
+            .store_replay_protection(protection, transaction_id)
+            .await
+    }
+
     async fn relate_account_to_transaction(
         &mut self,
         account_type: AccountTransactionType,

utxo/src/storage/mod.rs

@@ -135,6 +135,10 @@ pub enum Error {
     /// Error with the encoding/decoding
     Encoding(String),
 
+    #[error("Transaction already exists: {0}")]
+    /// The transaction already exists. Error thrown when the replay protection is already stored
+    AlreadyExists(TxId),
+
     #[error("Not enough unspent payments (missing {0} cents)")]
     /// TODO: Convert the AmountCents to Amount for better error reporting upstream
     NotEnoughUnspentPayments(AmountCents),
@@ -160,6 +164,13 @@ pub trait Batch<'a> {
         status: ReceivedPaymentStatus,
     ) -> Result<(), Error>;
 
+    /// Stores the replay protection. It fails if the protection is already stored.
+    async fn store_replay_protection(
+        &mut self,
+        protection: &str,
+        transaction_id: &TxId,
+    ) -> Result<(), Error>;
+
     /// Create a new list of payments
     async fn create_payments(
         &mut self,
@@ -375,7 +386,55 @@ pub mod test {
             $crate::storage_unit_test!(not_spendable_new_payments_not_spendable);
             $crate::storage_unit_test!(subscribe_realtime);
             $crate::storage_unit_test!(transaction_locking);
+            $crate::storage_unit_test!(transaction_replay_protection);
+        };
+    }
+
+    pub async fn transaction_replay_protection<T>(storage: T)
+    where
+        T: Storage + Send + Sync,
+    {
+        let config = Config {
+            storage,
+            token_manager: Default::default(),
+            status: StatusManager::default(),
         };
+
+        let ledger = Ledger::new(config);
+
+        let asset: Asset = "USD/2".parse().expect("valid asset");
+        let deposit = Transaction::new_external_deposit(
+            "test reference".to_owned(),
+            "pending".into(),
+            vec![],
+            vec![(
+                "alice".parse().expect("account"),
+                asset.from_human("100.99").expect("valid amount"),
+            )],
+        )
+        .expect("valid tx")
+        .set_replay_protection("test".to_owned())
+        .expect("valid tx");
+
+        assert!(ledger.store(deposit).await.is_ok());
+
+        let deposit = Transaction::new_external_deposit(
+            "test reference".to_owned(),
+            "pending".into(),
+            vec![],
+            vec![(
+                "alice".parse().expect("account"),
+                asset.from_human("100.99").expect("valid amount"),
+            )],
+        )
+        .expect("valid tx")
+        .set_replay_protection("test".to_owned())
+        .expect("valid tx");
+
+        let result = ledger.store(deposit).await;
+
+        assert!(result.is_err());
+        assert!(result.unwrap_err().to_string().contains("already exists"));
     }
 
     pub async fn transaction_locking<T>(storage: T)

utxo/src/storage/sqlite/batch.rs

@@ -64,6 +64,27 @@ impl<'a> storage::Batch<'a> for Batch<'a> {
             .map_err(|e| Error::Storage(e.to_string()))
     }
 
+    async fn store_replay_protection(
+        &mut self,
+        protection: &str,
+        transaction_id: &TxId,
+    ) -> Result<(), Error> {
+        if let Err(e) = sqlx::query(
+            r#"INSERT INTO "transactions_replay_protection"("protection_id", "transaction_id") VALUES(?, ?) "#,
+        ).bind(protection).bind(transaction_id.to_string())
+        .execute(&mut *self.inner)
+        .await {
+            Err(if let Ok(Some(row)) = sqlx::query(r#"SELECT "transaction_id" FROM "transactions_replay_protection" WHERE "protection_id" = ? "#).bind(protection)
+                .fetch_optional(&mut *self.inner).await {
+                    Error::AlreadyExists(row.get::<String, usize>(0).parse()?)
+                } else {
+            Error::Storage(e.to_string())
+                })
+        } else {
+            Ok(())
+        }
+    }
+
     async fn spend_payments(
         &mut self,
         transaction_id: &TxId,

utxo/src/storage/sqlite/mod.rs

@@ -377,6 +377,11 @@ impl SQLite {
             PRIMARY KEY ("tag", "created_at", "transaction_id")
         );
         CREATE INDEX IF NOT EXISTS "transaction_id_and_tags" ON "transactions_by_tags" ("transaction_id");
+        CREATE TABLE IF NOT EXISTS "transactions_replay_protection" (
+            "protection_id" VARCHAR(67) PRIMARY KEY,
+            "transaction_id" VARCHAR(67),
+            "created_at" DATETIME DEFAULT CURRENT_TIMESTAMP
+        );
         CREATE TABLE IF NOT EXISTS "payments" (
             "payment_id" VARCHAR(80) NOT NULL PRIMARY KEY,
             "to" VARCHAR(64) NOT NULL,

utxo/src/transaction/base_tx.rs

@@ -18,6 +18,8 @@ use std::collections::HashMap;
 /// spent must be equal to the sum of the payments created, unless the transaction is a Deposit or
 /// Withdrawal transaction.
 pub struct BaseTx {
+    /// A unique identifier generated by the client to make sure the transaction was not created before. If provided the storage layer will make sure it is unique.
+    pub replay_protection: Option<String>,
     /// List of spend payments to create this transaction
     #[serde(skip_serializing_if = "Vec::is_empty")]
     pub spends: Vec<PaymentFrom>,
@@ -52,6 +54,7 @@ impl BaseTx {
             spends,
             creates,
             reference,
+            replay_protection: None,
             typ,
             created_at: Utc::now(),
         };

utxo/src/transaction/mod.rs

@@ -123,6 +123,24 @@ impl Transaction {
         })
     }
 
+    /// Consumes the current transaction and replaces it with a similar transaction with replay protection
+    pub fn set_replay_protection(mut self, replay_protection: String) -> Result<Self, Error> {
+        self.transaction.replay_protection = Some(replay_protection);
+
+        let new_tx_id = self.transaction.id()?;
+        self.revision.transaction_id = new_tx_id.clone();
+
+        let new_rev_id = self.revision.rev_id()?;
+
+        Ok(Self {
+            id: new_tx_id,
+            revisions: vec![new_rev_id.clone()],
+            revision_id: new_rev_id,
+            revision: self.revision,
+            transaction: self.transaction,
+        })
+    }
+
     /// Returns the filterable fields
     pub fn get_filterable_fields(&self) -> Vec<FilterableValue> {
         let mut filters = vec![