feat: process response, deserialize bcs events

Author: 0xmovses

protocol-units/bridge/chains/movement/src/event_monitoring.rs

@@ -2,11 +2,14 @@ use std::{pin::Pin, task::Poll};
 
 use crate::MovementClient;
 use crate::{event_types::MovementChainEvent, types::MovementHash, utils::MovementAddress};
+use anyhow::Result;
+use aptos_sdk::rest_client::Response;
+use aptos_types::contract_event::{ContractEvent, ContractEventV1, EventWithVersion};
 use async_stream::try_stream;
-use bridge_shared::{
-	bridge_monitoring::{BridgeContractCounterpartyEvent, BridgeContractCounterpartyMonitoring},
-	counterparty_contract::SmartContractCounterpartyEvent,
+use bridge_shared::bridge_monitoring::{
+	BridgeContractCounterpartyEvent, BridgeContractCounterpartyMonitoring,
 };
+use bridge_shared::types::CounterpartyCompletedDetails;
 use futures::{FutureExt, Stream, StreamExt};
 use tokio::sync::mpsc::UnboundedReceiver;
 
@@ -44,24 +47,43 @@ impl Stream for MovementCounterpartyMonitoring<MovementAddress, MovementHash> {
 		let this = self.get_mut();
 		let stream = try_stream! {
 			loop {
-				let struct_tag = format!("0x{}::atomic_bridge_counterpary::BridgeTransferAssetsLockedEvent", client.counterparty_address.to_hex_literal());
-				rest_client.get_account_events_bcs(client.counterparty_address, struct_tag.as_str(), field_name, start, limit);
-				let response = reqwest::get("https://httpbin.org/ip").await.map_err(|e| {
-					tracing::error!("Failed to get response: {:?}", e);
-					Error::msg("Failed to get response")
-				})?;
-
-				// Create an event from the response (replace with actual logic)
-				let event = BridgeContractCounterpartyEvent {
-					// Populate event fields based on the response
-				};
+				let struct_tag = format!(
+						"0x{}::atomic_bridge_counterpary::BridgeCounterpartyEvents",
+						client.counterparty_address.to_hex_literal()
+				);
+				let response = rest_client
+								.get_account_events_bcs(
+										client.counterparty_address,
+										struct_tag.as_str(),
+										"bridge_transfer_assets_locked",
+										Some(1),
+										None
+								).await?;
+						let events = process_response(response);
+				let bridge_transfer_details = bcs::from_bytes::<CounterpartyCompletedDetails<MovementAddress, MovementHash>>(
+						&response.event_data
+				);
 
 				// Yield the event
-				yield Ok(event);
+				yield Ok(events);
 			}
 		};
 
 		let mut stream = Box::pin(stream);
 		Pin::new(&mut stream).poll_next(cx)
 	}
 }
+
+fn process_response(
+	res: Response<Vec<EventWithVersion>>,
+) -> Result<Vec<CounterpartyCompletedDetails<MovementAddress, MovementHash>>, bcs::Error> {
+	res.into_inner()
+		.into_iter()
+		.map(|e| {
+			let event_data = e.event.event_data(); // Use the method from the trait
+			bcs::from_bytes::<CounterpartyCompletedDetails<MovementAddress, MovementHash>>(
+				event_data,
+			)
+		})
+		.collect() // Collect the results, handling potential errors
+}

protocol-units/bridge/move-modules/sources/atomic_bridge_counterparty.move

@@ -37,6 +37,7 @@ module atomic_bridge::atomic_bridge_counterparty {
     /// An event triggered upon locking assets for a bridge transfer 
     struct BridgeTransferAssetsLockedEvent has store, drop {
         bridge_transfer_id: vector<u8>,
+        initiator: vector<u8>,
         recipient: address,
         amount: u64,
         hash_lock: vector<u8>,
@@ -103,6 +104,7 @@ module atomic_bridge::atomic_bridge_counterparty {
         event::emit(
             BridgeTransferAssetsLockedEvent {
                 bridge_transfer_id,
+                initiator,
                 recipient,
                 amount,
                 hash_lock,

protocol-units/bridge/shared/Cargo.toml

@@ -13,6 +13,7 @@ rust-version.workspace = true
 
 [dependencies]
 async-trait = "0.1.80"
+bcs = { workspace = true }
 delegate = "0.12.0"
 derive_more = { workspace = true, features = ["deref", "deref_mut"] }
 futures.workspace = true
@@ -22,6 +23,7 @@ thiserror.workspace = true
 tracing.workspace = true
 rand.workspace = true
 rand_chacha = "0.2.2"
+serde = { workspace = true, features = ["derive"] }
 futures-time = "3.0.0"
 alloy = { workspace = true, features = [
   "full",

protocol-units/bridge/shared/src/types.rs

@@ -2,12 +2,13 @@ use alloy::primitives::Uint;
 use derive_more::{Deref, DerefMut};
 use hex::{self, FromHexError};
 use rand::{Rng, RngCore};
+use serde::{Deserialize, Serialize};
 use std::{fmt::Debug, hash::Hash};
 use thiserror::Error;
 
 use crate::bridge_contracts::{BridgeContractCounterpartyError, BridgeContractInitiatorError};
 
-#[derive(Deref, Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Deref, Debug, Clone, PartialEq, Eq, Hash, Deserialize, Serialize)]
 pub struct BridgeTransferId<H>(pub H);
 
 impl<H> BridgeTransferId<H> {
@@ -55,7 +56,7 @@ where
 	}
 }
 
-#[derive(Deref, Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Deref, Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub struct InitiatorAddress<A>(pub A);
 
 impl From<&str> for InitiatorAddress<Vec<u8>> {
@@ -70,7 +71,7 @@ impl From<String> for InitiatorAddress<Vec<u8>> {
 	}
 }
 
-#[derive(Deref, Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Deref, Debug, Clone, PartialEq, Eq, Hash, Deserialize, Serialize)]
 pub struct RecipientAddress<A>(pub A);
 
 impl From<&str> for RecipientAddress<Vec<u8>> {
@@ -85,7 +86,7 @@ pub struct RecipientAddressCounterparty<A>(pub A);
 #[derive(Deref, Debug, Clone, PartialEq, Eq, Hash)]
 pub struct InitiatorAddressCounterParty(pub Vec<u8>);
 
-#[derive(Deref, Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Deref, Debug, Clone, PartialEq, Eq, Hash, Deserialize, Serialize)]
 pub struct HashLock<H>(pub H);
 
 impl<H> HashLock<H> {
@@ -107,7 +108,7 @@ pub fn convert_hash_lock<H: From<O>, O>(other: HashLock<O>) -> HashLock<H> {
 	HashLock(From::from(other.0))
 }
 
-#[derive(Deref, Debug, Clone, PartialEq, Eq)]
+#[derive(Deref, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct HashLockPreImage(pub Vec<u8>);
 
 impl AsRef<[u8]> for HashLockPreImage {
@@ -126,7 +127,7 @@ impl HashLockPreImage {
 	}
 }
 
-#[derive(Deref, Debug, Clone, PartialEq, Eq)]
+#[derive(Deref, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct TimeLock(pub u64);
 
 impl From<Uint<256, 4>> for TimeLock {
@@ -137,7 +138,7 @@ impl From<Uint<256, 4>> for TimeLock {
 	}
 }
 
-#[derive(Deref, DerefMut, Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Deref, DerefMut, Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Amount(pub u64);
 
 impl From<Uint<256, 4>> for Amount {
@@ -174,7 +175,7 @@ pub struct LockDetails<A, H> {
 	pub amount: Amount,
 }
 
-#[derive(Debug, PartialEq, Eq, Clone)]
+#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
 pub struct CounterpartyCompletedDetails<A, H> {
 	pub bridge_transfer_id: BridgeTransferId<H>,
 	pub initiator_address: InitiatorAddress<Vec<u8>>,