e2e test and fixing changes along the way

Cargo.toml

@@ -24,6 +24,8 @@ flume                                                   = "0.11.0"
 sp-rpc                                                  = "29.0.0"
 subxt                                                   = "0.35.0"
 tokio                                                   = "1.36.0"
+hex                                                     = "0.4.3"
+hex-literal                                             = "0.4.1"
 tower-http                                              = { version = "0.4.0", features = ["full"] }
 tower                                                   = { version = "0.4.13", features = ["full"] }
 jsonrpsee                                               = { version = "0.17", features = ["server", "client-core", "http-client", "ws-client", "macros"] }

av-layer/Cargo.toml

@@ -25,4 +25,5 @@ serde_json                                  = { workspace = true}
 subxt                                       = { workspace = true}
 tracing 									= { workspace = true}
 sp-tracing 									= { workspace = true}
-tracing-subscriber 							= { workspace = true}
+tracing-subscriber 							= { workspace = true}
+hex                                         = { workspace = true}

av-layer/src/handlers.rs

@@ -1,12 +1,12 @@
 use crate::traits::*;
+use anyhow::ensure;
 use jsonrpsee::core::{async_trait, SubscriptionResult};
 use jsonrpsee::core::{Error::Custom, RpcResult};
 use jsonrpsee::{PendingSubscriptionSink, SubscriptionMessage};
-use parity_scale_codec::alloc::sync::Once;
 use parity_scale_codec::{Decode, Encode};
 use primitives::{
-    BlockchainNetwork, ConfirmationStatus, TxConfirmationObject, TxObject, TxSimulationObject,
-    VaneCallData, VaneMultiAddress,
+    BlockchainNetwork, ConfirmationStatus, MultiId, TxConfirmationObject, TxObject,
+    TxSimulationObject, VaneCallData, VaneMultiAddress,
 };
 use serde_json::Value as JsonValue;
 use sp_core::ecdsa::{Public as ecdsaPublic, Signature as ECDSASignature};
@@ -20,26 +20,7 @@ use std::{
 };
 use subxt::utils::{AccountId32, MultiAddress, MultiSignature};
 use tokio::sync::Mutex;
-use tracing_subscriber;
-
-// Tracing setup
-static INIT: Once = Once::new();
-pub fn init_tracing() -> anyhow::Result<()> {
-    // Add test tracing (from sp_tracing::init_for_tests()) but filtering for xcm logs only
-    let vane_subscriber = tracing_subscriber::fmt()
-        .compact()
-        .with_file(true)
-        .with_line_number(true)
-        .with_target(true)
-        .finish();
-
-    tracing::dispatcher::set_global_default(vane_subscriber.into())
-        .expect("Failed to initialise tracer");
-    Ok(())
-}
 
-/// Types for easier code navigation
-pub type MultiId = VaneMultiAddress<AccountId32, ()>;
 /// A mock database storing each address to the transactions each having a key
 /// `address` ===> `multi_id`=====> `Vec<u8>`
 pub struct MockDB {
@@ -56,7 +37,7 @@ pub struct MockDB {
     // Store ready to be simulated tx `TxSimulationObject` (queue)
     pub simulation: VecDeque<Vec<u8>>,
     // Record reverted transactions per sender
-    pub reverted_transactions: BTreeMap<MultiAddress<AccountId32, ()>, Vec<u8>>,
+    pub reverted_transactions: BTreeMap<VaneMultiAddress<AccountId32, ()>, Vec<u8>>,
 
     // ============================================================================
     // METRICS
@@ -87,8 +68,8 @@ impl TransactionHandler {
         inner_db_multi_ids.push(multi_id.clone());
 
         db.transactions.insert(multi_id, data.encode());
-        db.multi_ids.insert(address, inner_db_multi_ids);
-        tracing::info!("recorded tx to the memory db")
+        db.multi_ids.insert(address.clone(), inner_db_multi_ids);
+        tracing::info!("recorded tx to the memory db for {:?}", address)
     }
 
     pub async fn set_confirmation_transaction_data(
@@ -140,6 +121,20 @@ impl TransactionHandler {
         }
     }
 
+    pub async fn get_reverted_txs(&self) -> Vec<TxConfirmationObject> {
+        let db = self.db.lock().await;
+        let reverted_txs: Vec<TxConfirmationObject> = db
+            .reverted_transactions
+            .values()
+            .into_iter()
+            .map(|tx| {
+                let decoded_tx: TxConfirmationObject = Decode::decode(&mut &tx[..]).expect("hh");
+                decoded_tx
+            })
+            .collect();
+        return reverted_txs;
+    }
+
     pub async fn propagate_tx(&self, tx_simulate: TxSimulationObject) {
         let mut db = self.db.lock().await;
         db.simulation.push_front(tx_simulate.encode());
@@ -162,6 +157,15 @@ impl TransactionHandler {
         }
     }
 
+    pub async fn record_reverted_tx(
+        &self,
+        sender: VaneMultiAddress<AccountId32, ()>,
+        reverted_tx: TxConfirmationObject,
+    ) {
+        let mut db = self.db.lock().await;
+        db.reverted_transactions
+            .insert(sender, reverted_tx.encode());
+    }
     // METRICS
 
     pub async fn record_subscriber(&self, id: JsonValue) {
@@ -209,7 +213,7 @@ impl TransactionServer for TransactionHandler {
         todo!()
     }
 
-    async fn subscribe_tx_confirmation(
+    async fn receiver_subscribe_tx_confirmation(
         &self,
         pending: PendingSubscriptionSink,
         address: VaneMultiAddress<AccountId32, ()>,
@@ -241,7 +245,7 @@ impl TransactionServer for TransactionHandler {
     }
 
     // Subscribe for sender to listen to confirmed tx from the receiver
-    async fn subscribe_tx_confirmation_sender(
+    async fn sender_subscribe_tx_confirmation(
         &self,
         pending: PendingSubscriptionSink,
         address: VaneMultiAddress<AccountId32, ()>,
@@ -288,7 +292,9 @@ impl TransactionServer for TransactionHandler {
                     .ok_or(Custom("Transaction Not Found".to_string()))?;
                 // record the confirmation
 
-                let msg = tx.clone().call;
+                // message to sign for address verification
+                let msg = tx.clone().get_tx_id();
+
                 let sig = Sr25519Signature::from_slice(&signature)
                     .ok_or(Custom("Failed to convert signature sr25519".to_string()))?;
 
@@ -297,13 +303,17 @@ impl TransactionServer for TransactionHandler {
                     .try_into()
                     .expect("Failed to covert address to bytes");
                 let public_account = sr25519Public::from_h256(H256::from(account_bytes));
+
                 if sig.verify(&msg.encode()[..], &public_account) {
                     let mut tx_confirmation_object: TxConfirmationObject = tx.into();
+
+                    // update the confirmed address
+                    tx_confirmation_object.set_confirmed_receiver(address);
+
                     // update the confirmation status
                     tx_confirmation_object.update_confirmation_status(
                         primitives::ConfirmationStatus::WaitingForSender,
                     );
-                    tx_confirmation_object.set_receiver_sig(signature);
                     // store the tx confirmation object
                     self.set_confirmation_transaction_data(multi_id.into(), tx_confirmation_object)
                         .await
@@ -328,12 +338,16 @@ impl TransactionServer for TransactionHandler {
                     .get_confirmation_transaction_data(multi_id.clone().into())
                     .await
                     .ok_or(Custom("Confirmation data unavailable".to_string()))?;
+
                 // check if the if the receiver has confirmed
                 if tx.get_confirmation_status() != ConfirmationStatus::WaitingForSender {
                     return Err(Custom("Wait for receiver to confirm".to_string()));
                 }
                 // verify the signature and the address
-                let msg = tx.clone().call;
+
+                // message to sign for address verification
+                let msg = tx.clone().get_tx_id();
+
                 let sig = Sr25519Signature::from_slice(&signature)
                     .expect("Failed to convert signature sr25519");
 
@@ -342,13 +356,28 @@ impl TransactionServer for TransactionHandler {
                     .try_into()
                     .expect("Failed to covert address to bytes");
                 let public_account = sr25519Public::from_h256(H256::from(account_bytes));
+
                 if sig.verify(&msg.encode()[..], &public_account) {
-                    tx.update_confirmation_status(ConfirmationStatus::Ready);
-                    let tx_simulation_object: TxSimulationObject = tx.into();
-                    // store to the ready to be simulated tx storage
-                    self.propagate_tx(tx_simulation_object).await;
+                    // confirm the resulting multi_id
+                    let multi_id = tx.calculate_confirmed_multi_id(address.clone());
+
+                    if multi_id == tx.get_multi_id() {
+                        tx.set_confirmed_sender(address);
+
+                        tx.update_confirmation_status(ConfirmationStatus::Ready);
+                        let tx_simulation_object: TxSimulationObject = tx.into();
+                        // store to the ready to be simulated tx storage
+                        self.propagate_tx(tx_simulation_object).await;
+                        tracing::info!("sender confirmed");
+                    } else {
+                        // record to reverted tx, as this tx is automatically reverted due to mismatch in address confirmation
+                        tx.update_confirmation_status(ConfirmationStatus::RejectedMismatchAddress);
+
+                        tracing::info!("tx reverted due to mismatched confirmed address");
+                        self.record_reverted_tx(address, tx).await;
+                        Err(Custom("Address Mismatched , Tx reverted".to_string()))?;
+                    }
                 };
-                tracing::info!("sender confirmed");
                 Ok(())
             }
             _ => Err(Custom("Blockchain network not supported".to_string())),
@@ -364,6 +393,14 @@ impl TransactionServer for TransactionHandler {
         todo!()
     }
 
+    async fn subscribe_revert_tx(&self, pending: PendingSubscriptionSink) -> SubscriptionResult {
+        let sink = pending.accept().await?;
+        let reverted_txs: Vec<TxConfirmationObject> = self.get_reverted_txs().await;
+        let json_reverted_txs = SubscriptionMessage::from_json(&reverted_txs)?;
+        sink.send(json_reverted_txs).await?;
+        Ok(())
+    }
+
     async fn receive_confirmed_tx(&self, pending: PendingSubscriptionSink) -> SubscriptionResult {
         let sink = pending.accept().await?;
         // fetch the confirmed and ready to be simulated txn

av-layer/src/main.rs

@@ -2,7 +2,6 @@ use anyhow::Ok;
 use clap::Parser;
 use jsonrpsee::server::ServerBuilder;
 use std::collections::BTreeMap;
-use std::collections::HashMap;
 use std::collections::VecDeque;
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -12,11 +11,11 @@ mod traits;
 
 use handlers::MockDB;
 use handlers::TransactionHandler;
+use parity_scale_codec::alloc::sync::Once;
 use tokio::sync::Mutex;
+use tracing_subscriber;
 use traits::TransactionServer;
 
-use crate::handlers::init_tracing;
-
 /// Address Verification layer cli server arguments
 #[derive(Parser, Debug)]
 #[command(version, about, long_about = None)]
@@ -34,9 +33,26 @@ pub struct ServerProfile {
     port: u16,
 }
 
+// Tracing setup
+static INIT: Once = Once::new();
+pub fn init_tracing() -> anyhow::Result<()> {
+    // Add test tracing (from sp_tracing::init_for_tests()) but filtering for xcm logs only
+    let vane_subscriber = tracing_subscriber::fmt()
+        .compact()
+        .with_file(true)
+        .with_line_number(true)
+        .with_target(true)
+        .finish();
+
+    tracing::dispatcher::set_global_default(vane_subscriber.into())
+        .expect("Failed to initialise tracer");
+    Ok(())
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     init_tracing()?;
+    tracing::info!("initiliasing av layer 🔥⚡️");
     //let args = AvLayerServerCli::parse();
     // Initialise the database
     let mock_db_transactions = BTreeMap::new();
@@ -54,7 +70,6 @@ async fn main() -> anyhow::Result<()> {
             subscribed: Vec::new(),
         })),
     };
-    println!("Starting server");
 
     // Initialize the server
     run_rpc_server(rpc_handler, "127.0.0.1:8000".to_owned()).await?;

av-layer/src/traits.rs

@@ -26,15 +26,15 @@ pub trait Transaction {
 
     /// Subscription to start listening to any upcoming confirmation request
     /// returns `Vec<TxConfirmationObject>` in encoded format
-    #[subscription(name = "subscribeTxConfirmation", item=Vec<Vec<u8>>)]
-    async fn subscribe_tx_confirmation(
+    #[subscription(name = "receiverSubscribeTxConfirmation", unsubscribe= "receiverUnsubscribeTxConfirmation", item=Vec<Vec<u8>>)]
+    async fn receiver_subscribe_tx_confirmation(
         &self,
         address: VaneMultiAddress<AccountId32, ()>,
     ) -> SubscriptionResult;
 
     /// Subscriptiom for sender to listen to incoming confirmed transactions from the receiver
-    #[subscription(name = "subscribeTxConfirmationSender", item=Vec<TxConfirmationObject>)]
-    async fn subscribe_tx_confirmation_sender(
+    #[subscription(name = "senderSubscribeTxConfirmation",unsubscribe= "senderUnsubscribeTxConfirmation", item=Vec<TxConfirmationObject>)]
+    async fn sender_subscribe_tx_confirmation(
         &self,
         address: VaneMultiAddress<AccountId32, ()>,
     ) -> SubscriptionResult;
@@ -69,6 +69,10 @@ pub trait Transaction {
         network: BlockchainNetwork,
     ) -> RpcResult<()>;
 
+    /// Subscribe to reverted transactions
+    #[subscription(name = "subscribeRevertTx",unsubscribe= "unsubscribeRevertTx", item=Vec<TxConfirmationObject>)]
+    async fn subscribe_revert_tx(&self) -> SubscriptionResult;
+
     /// Handling confirmation of transaction from receiver and sender
     /// A websocket connection
 

integration-test/Cargo.toml

@@ -23,3 +23,5 @@ subxt-signer                        = { workspace = true}
 tracing 							= { workspace = true}
 sp-tracing 							= { workspace = true}
 tracing-subscriber 					= { workspace = true}
+clap                                = { workspace = true, features = ["derive"]}
+hex                                 = { workspace = true}

integration-test/src/main.rs

@@ -28,20 +28,40 @@ use subxt::tx::TxPayload;
 use subxt::utils::{AccountId32, MultiAddress};
 use subxt::{Metadata, OnlineClient, PolkadotConfig};
 use subxt_signer::sr25519::{dev, Keypair};
+use jsonrpsee::rpc_params;
+use clap::Parser;
+
 #[subxt::subxt(runtime_metadata_path = "polkadot.scale")]
 pub mod polkadot {}
 
 // Polkadot testing
 // Solana testing
 // Ethereum testing
 
+#[derive(Parser, Debug)]
+#[command(version, about, long_about = None)]
+struct VaneTestCli {
+    /// Name of the server
+    #[arg(short, long)]
+    name: String
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    // generate accounts
-    // let alicePair = sr25519Pair::from_string("//Alice", None).expect("Failed to generate key pair");
-    // let bobPair = sr25519Pair::from_string("//Bob", None).expect("Failed to generate key pair");
-    let alice = dev::alice().public_key();
-    //let bob:VaneMultiAddress<u128,u32> = VaneMultiAddress::Address32(dev::bob().public_key().into());
+
+    let args = VaneTestCli::parse();
+
+    if args.name == "polkadot".to_string() {
+        let alice = dev::alice();
+        let bob = dev::bob();
+        let vane_account = ();
+
+        let polkadot_test = PolkadotTest::connect().await;
+
+
+    }
+
+
     // construct a transfer tx
     //let transfer_call = polkadot::tx().balances().transfer_keep_alive(bob, 10_000);
     // send to vane av-layer
@@ -66,7 +86,7 @@ pub struct PolkadotTest {
 impl PolkadotTest {
     pub async fn connect() -> PolkadotTest {
         let client = WsClientBuilder::default()
-            .build("127.0.0.1:8000")
+            .build("ws://127.0.0.1:8000")
             .await
             .expect("Failed to initilise Ws");
 
@@ -87,11 +107,12 @@ impl PolkadotTest {
 
         let vane_call_data = VaneCallData::new(BlockchainNetwork::Polkadot, amount);
         // use the client to submit the transaction to av layer
+        //let params = rpc_params!(sender_multi,receiver_multi,vane_call_data);
         if self.client.is_connected() {
             self.client
                 .request(
-                    "submitTransaction",
-                    vec![sender_multi,receiver_multi],
+                    "submit_transaction",
+                    (vane_call_data, sender_multi, receiver_multi),
                 )
                 .await?;
         }