update: refactor process_events

Author: 0xmovses

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

@@ -1,15 +1,19 @@
 use std::{pin::Pin, task::Poll};
 
 use crate::MovementClient;
-use crate::{event_types::MovementChainEvent, types::MovementHash, utils::MovementAddress};
+use crate::{
+	event_types::{CounterpartyEventKind, MovementChainEvent},
+	types::MovementHash,
+	utils::MovementAddress,
+};
 use anyhow::Result;
 use aptos_sdk::rest_client::Response;
-use aptos_types::contract_event::{ContractEvent, ContractEventV1, EventWithVersion};
+use aptos_types::contract_event::EventWithVersion;
 use async_stream::try_stream;
 use bridge_shared::bridge_monitoring::{
 	BridgeContractCounterpartyEvent, BridgeContractCounterpartyMonitoring,
 };
-use bridge_shared::types::CounterpartyCompletedDetails;
+use bridge_shared::types::{CounterpartyCompletedDetails, LockDetails};
 use futures::{FutureExt, Stream, StreamExt};
 use tokio::sync::mpsc::UnboundedReceiver;
 
@@ -51,20 +55,16 @@ impl Stream for MovementCounterpartyMonitoring<MovementAddress, MovementHash> {
 						"0x{}::atomic_bridge_counterpary::BridgeCounterpartyEvents",
 						client.counterparty_address.to_hex_literal()
 				);
-				let response = rest_client
+				let locked_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
-				);
+				let events = process_response(locked_response)?;
 
-				// Yield the event
 				yield Ok(events);
 			}
 		};
@@ -76,14 +76,25 @@ impl Stream for MovementCounterpartyMonitoring<MovementAddress, MovementHash> {
 
 fn process_response(
 	res: Response<Vec<EventWithVersion>>,
-) -> Result<Vec<CounterpartyCompletedDetails<MovementAddress, MovementHash>>, bcs::Error> {
+	kind: CounterpartyEventKind,
+) -> Result<Vec<BridgeContractCounterpartyEvent<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,
-			)
+			let data = e.event.event_data(); // Use the method from the trait
+			match kind {
+				CounterpartyEventKind::Locked => {
+					let locked_details =
+						bcs::from_bytes::<LockDetails<MovementAddress, [u8; 32]>>(data)?;
+					Ok(BridgeContractCounterpartyEvent::Locked(locked_details))
+				}
+				CounterpartyEventKind::Completed => {
+					let completed_details = bcs::from_bytes::<
+						CounterpartyCompletedDetails<MovementAddress, [u8; 32]>,
+					>(data)?;
+					Ok(BridgeContractCounterpartyEvent::Completed(completed_details))
+				}
+			}
 		})
-		.collect() // Collect the results, handling potential errors
+		.collect()
 }

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

@@ -6,3 +6,8 @@ pub enum MovementChainEvent<A, H> {
 	CounterpartyContractEvent(SCCResult<A, H>),
 	Noop,
 }
+
+pub enum CounterpartyEventKind {
+	Locked,
+	Completed,
+}

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

@@ -165,7 +165,7 @@ impl<A, H> Default for BridgeTransferDetails<A, H> {
 	}
 }
 
-#[derive(Debug, PartialEq, Eq, Clone)]
+#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
 pub struct LockDetails<A, H> {
 	pub bridge_transfer_id: BridgeTransferId<H>,
 	pub initiator_address: InitiatorAddress<Vec<u8>>,