refactor: split long functions in relayer-lib

packages/relayer-lib/Cargo.toml

@@ -17,7 +17,7 @@ sp1-ics07-tendermint-prover   = { workspace = true, optional = true }
 sp1-ics07-tendermint-utils    = { workspace = true, optional = true }
 
 serde = { workspace = true, features = ["derive"] }
-prost = { workspace = true }
+prost = { workspace = true, features = ["std"] }
 
 async-trait = { workspace = true }
 anyhow      = { workspace = true, features = ["std"] }
@@ -26,6 +26,7 @@ tracing     = { workspace = true, default-features = true }
 
 tendermint     = { workspace = true, features = ["std"] }
 tendermint-rpc = { workspace = true, features = ["http-client"] }
+tendermint-light-client-verifier = { workspace = true }
 
 ibc-proto-eureka    = { workspace = true }
 ibc-core-host-types = { workspace = true }

packages/relayer-lib/src/events/eureka.rs

@@ -2,7 +2,7 @@
 
 use alloy::{hex, sol_types::SolEvent};
 use ibc_eureka_solidity_types::ics26::router::{
-    routerEvents, AckPacket, RecvPacket, SendPacket, TimeoutPacket, WriteAcknowledgement,
+    routerEvents, RecvPacket, SendPacket, WriteAcknowledgement,
 };
 use ibc_proto_eureka::ibc::core::channel::v2::{Acknowledgement, Packet};
 use prost::Message;
@@ -18,10 +18,6 @@ pub enum EurekaEvent {
     SendPacket(SendPacket),
     /// A packet was received.
     RecvPacket(RecvPacket),
-    /// A packet was acknowledged.
-    AckPacket(AckPacket),
-    /// A packet timed out.
-    TimeoutPacket(TimeoutPacket),
     /// An acknowledgement was written.
     WriteAcknowledgement(WriteAcknowledgement),
 }
@@ -30,12 +26,10 @@ impl EurekaEvent {
     /// Get the signature of the events for EVM.
     /// This is used to filter the logs.
     #[must_use]
-    pub const fn evm_signatures() -> [&'static str; 5] {
+    pub const fn evm_signatures() -> [&'static str; 3] {
         [
             SendPacket::SIGNATURE,
             RecvPacket::SIGNATURE,
-            AckPacket::SIGNATURE,
-            TimeoutPacket::SIGNATURE,
             WriteAcknowledgement::SIGNATURE,
         ]
     }
@@ -48,9 +42,11 @@ impl TryFrom<routerEvents> for EurekaEvent {
         match event {
             routerEvents::SendPacket(event) => Ok(Self::SendPacket(event)),
             routerEvents::RecvPacket(event) => Ok(Self::RecvPacket(event)),
-            routerEvents::AckPacket(event) => Ok(Self::AckPacket(event)),
-            routerEvents::TimeoutPacket(event) => Ok(Self::TimeoutPacket(event)),
             routerEvents::WriteAcknowledgement(event) => Ok(Self::WriteAcknowledgement(event)),
+            routerEvents::AckPacket(_) => Err(anyhow::anyhow!("AckPacket event is not used")),
+            routerEvents::TimeoutPacket(_) => {
+                Err(anyhow::anyhow!("TimeoutPacket event is not used"))
+            }
             routerEvents::Noop(_) => Err(anyhow::anyhow!("Noop event")),
             routerEvents::IBCAppAdded(_) => Err(anyhow::anyhow!("IBCAppAdded event")),
             routerEvents::OwnershipTransferred(_) => {
@@ -69,30 +65,28 @@ impl TryFrom<TmEvent> for EurekaEvent {
                 .attributes
                 .into_iter()
                 .find_map(|attr| {
-                    if attr.key_str().ok()? == cosmos_sdk::ATTRIBUTE_KEY_ENCODED_PACKET_HEX {
-                        let packet: Vec<u8> = hex::decode(attr.value_str().ok()?).ok()?;
-                        let packet = Packet::decode(packet.as_slice()).ok()?;
-                        Some(Self::SendPacket(SendPacket {
-                            packet: packet.try_into().ok()?,
-                        }))
-                    } else {
-                        None
+                    if attr.key_str().ok()? != cosmos_sdk::ATTRIBUTE_KEY_ENCODED_PACKET_HEX {
+                        return None;
                     }
+                    let packet: Vec<u8> = hex::decode(attr.value_str().ok()?).ok()?;
+                    let packet = Packet::decode(packet.as_slice()).ok()?;
+                    Some(Self::SendPacket(SendPacket {
+                        packet: packet.try_into().ok()?,
+                    }))
                 })
                 .ok_or_else(|| anyhow::anyhow!("No packet data found")),
             cosmos_sdk::EVENT_TYPE_RECV_PACKET => event
                 .attributes
                 .into_iter()
                 .find_map(|attr| {
-                    if attr.key_str().ok()? == cosmos_sdk::ATTRIBUTE_KEY_ENCODED_PACKET_HEX {
-                        let packet: Vec<u8> = hex::decode(attr.value_str().ok()?).ok()?;
-                        let packet = Packet::decode(packet.as_slice()).ok()?;
-                        Some(Self::RecvPacket(RecvPacket {
-                            packet: packet.try_into().ok()?,
-                        }))
-                    } else {
-                        None
+                    if attr.key_str().ok()? != cosmos_sdk::ATTRIBUTE_KEY_ENCODED_PACKET_HEX {
+                        return None;
                     }
+                    let packet: Vec<u8> = hex::decode(attr.value_str().ok()?).ok()?;
+                    let packet = Packet::decode(packet.as_slice()).ok()?;
+                    Some(Self::RecvPacket(RecvPacket {
+                        packet: packet.try_into().ok()?,
+                    }))
                 })
                 .ok_or_else(|| anyhow::anyhow!("No packet data found")),
             cosmos_sdk::EVENT_TYPE_WRITE_ACK => {

packages/relayer-lib/src/lib.rs

@@ -6,3 +6,4 @@ pub mod chain;
 pub mod events;
 pub mod listener;
 pub mod tx_builder;
+pub(crate) mod utils;

packages/relayer-lib/src/tx_builder/cosmos_to_cosmos.rs

@@ -2,24 +2,26 @@
 //! the Cosmos SDK chain from events received from another Cosmos SDK chain.
 
 use anyhow::Result;
-use futures::future;
 use ibc_proto_eureka::{
+    cosmos::tx::v1beta1::TxBody,
+    google::protobuf::Any,
     ibc::{
         core::{
-            channel::v2::{
-                Channel, MsgRecvPacket, MsgTimeout, QueryChannelRequest, QueryChannelResponse,
-            },
-            client::v1::{Height, MsgUpdateClient},
+            channel::v2::{Channel, QueryChannelRequest, QueryChannelResponse},
+            client::v1::MsgUpdateClient,
         },
         lightclients::tendermint::v1::ClientState,
     },
-    Protobuf,
 };
 use prost::Message;
 use sp1_ics07_tendermint_utils::{light_block::LightBlockExt, rpc::TendermintRpcExt};
 use tendermint_rpc::{Client, HttpClient};
 
-use crate::{chain::CosmosSdk, events::EurekaEvent};
+use crate::{
+    chain::CosmosSdk,
+    events::EurekaEvent,
+    utils::cosmos::{src_events_to_recv_and_ack_msgs, target_events_to_timeout_msgs},
+};
 
 use super::r#trait::TxBuilderService;
 
@@ -71,17 +73,12 @@ impl TxBuilder {
 
 #[async_trait::async_trait]
 impl TxBuilderService<CosmosSdk, CosmosSdk> for TxBuilder {
-    #[allow(clippy::too_many_lines)]
     async fn relay_events(
         &self,
         src_events: Vec<EurekaEvent>,
         target_events: Vec<EurekaEvent>,
         target_channel_id: String,
     ) -> Result<Vec<u8>> {
-        let now = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)?
-            .as_secs();
-
         let channel = self.channel(target_channel_id.clone()).await?;
         let client_state = ClientState::decode(
             self.target_tm_client
@@ -93,132 +90,30 @@ impl TxBuilderService<CosmosSdk, CosmosSdk> for TxBuilder {
 
         let target_light_block = self.source_tm_client.get_light_block(None).await?;
         let target_height = target_light_block.height().value().try_into()?;
-
-        let _timeout_msgs = future::try_join_all(
-            target_events
-                .into_iter()
-                .filter(|e| match e {
-                    EurekaEvent::SendPacket(se) => {
-                        now >= se.packet.timeoutTimestamp
-                            && se.packet.sourceChannel == target_channel_id
-                    }
-                    _ => false,
-                })
-                .map(|e| async {
-                    match e {
-                        EurekaEvent::SendPacket(se) => {
-                            let ibc_path = se.packet.receipt_commitment_path();
-                            self.source_tm_client
-                                .prove_path(&[b"ibc".to_vec(), ibc_path], target_height)
-                                .await
-                                .map(|(v, p)| {
-                                    if v.is_empty() {
-                                        Some(MsgTimeout {
-                                            packet: Some(se.packet.into()),
-                                            proof_unreceived: p.encode_vec(),
-                                            proof_height: Some(Height {
-                                                revision_number: client_state
-                                                    .latest_height
-                                                    .unwrap_or_default()
-                                                    .revision_number,
-                                                revision_height: target_height.into(),
-                                            }),
-                                            signer: self.signer_address.clone(),
-                                        })
-                                    } else {
-                                        None
-                                    }
-                                })
-                        }
-                        _ => unreachable!(),
-                    }
-                }),
+        let revision_number = client_state
+            .latest_height
+            .ok_or_else(|| anyhow::anyhow!("No latest height found"))?
+            .revision_number;
+
+        let timeout_msgs = target_events_to_timeout_msgs(
+            target_events,
+            &self.source_tm_client,
+            &target_channel_id,
+            revision_number,
+            target_height,
+            &self.signer_address,
         )
-        .await?
-        .into_iter()
-        .flatten()
-        .collect::<Vec<_>>();
-
-        let (src_send_events, src_ack_events): (Vec<_>, Vec<_>) = src_events
-            .into_iter()
-            .filter(|e| match e {
-                EurekaEvent::SendPacket(se) => {
-                    se.packet.timeoutTimestamp > now && se.packet.destChannel == target_channel_id
-                }
-                EurekaEvent::WriteAcknowledgement(we) => {
-                    we.packet.sourceChannel == target_channel_id
-                }
-                _ => false,
-            })
-            .partition(|e| match e {
-                EurekaEvent::SendPacket(_) => true,
-                EurekaEvent::WriteAcknowledgement(_) => false,
-                _ => unreachable!(),
-            });
-
-        let _recv_msgs = future::try_join_all(src_send_events.into_iter().map(|e| async {
-            match e {
-                EurekaEvent::SendPacket(se) => {
-                    let ibc_path = se.packet.commitment_path();
-                    self.source_tm_client
-                        .prove_path(&[b"ibc".to_vec(), ibc_path], target_height)
-                        .await
-                        .map(|(v, p)| {
-                            if v.is_empty() {
-                                Some(MsgRecvPacket {
-                                    packet: Some(se.packet.into()),
-                                    proof_height: Some(Height {
-                                        revision_number: client_state
-                                            .latest_height
-                                            .unwrap_or_default()
-                                            .revision_number,
-                                        revision_height: target_height.into(),
-                                    }),
-                                    proof_commitment: p.encode_vec(),
-                                    signer: self.signer_address.clone(),
-                                })
-                            } else {
-                                None
-                            }
-                        })
-                }
-                _ => unreachable!(),
-            }
-        }))
-        .await?
-        .into_iter()
-        .flatten()
-        .collect::<Vec<_>>();
-
-        let _ack_msgs = future::try_join_all(src_ack_events.into_iter().map(|e| async {
-            match e {
-                EurekaEvent::WriteAcknowledgement(we) => {
-                    let ibc_path = we.packet.ack_commitment_path();
-                    self.source_tm_client
-                        .prove_path(&[b"ibc".to_vec(), ibc_path], target_height)
-                        .await
-                        .map(|(v, p)| {
-                            if v.is_empty() {
-                                Some(MsgRecvPacket {
-                                    packet: Some(we.packet.into()),
-                                    proof_height: Some(Height {
-                                        revision_number: client_state
-                                            .latest_height
-                                            .unwrap_or_default()
-                                            .revision_number,
-                                        revision_height: target_height.into(),
-                                    }),
-                                    proof_commitment: p.encode_vec(),
-                                    signer: self.signer_address.clone(),
-                                })
-                            } else {
-                                None
-                            }
-                        })
-                }
-                _ => unreachable!(),
-            }
-        }));
+        .await?;
+
+        let (recv_msgs, ack_msgs) = src_events_to_recv_and_ack_msgs(
+            src_events,
+            &self.source_tm_client,
+            &target_channel_id,
+            revision_number,
+            target_height,
+            &self.signer_address,
+        )
+        .await?;
 
         let trusted_light_block = self
             .source_tm_client
@@ -230,15 +125,23 @@ impl TxBuilderService<CosmosSdk, CosmosSdk> for TxBuilder {
                     .try_into()?,
             ))
             .await?;
-
         let proposed_header = target_light_block.into_header(&trusted_light_block);
-
-        let _update_msg = MsgUpdateClient {
+        let update_msg = MsgUpdateClient {
             client_id: channel.client_id,
             client_message: Some(proposed_header.into()),
             signer: self.signer_address.clone(),
         };
 
-        todo!()
+        let all_msgs = std::iter::once(Any::from_msg(&update_msg))
+            .chain(timeout_msgs.into_iter().map(|m| Any::from_msg(&m)))
+            .chain(recv_msgs.into_iter().map(|m| Any::from_msg(&m)))
+            .chain(ack_msgs.into_iter().map(|m| Any::from_msg(&m)))
+            .collect::<Result<Vec<_>, _>>()?;
+
+        let tx_body = TxBody {
+            messages: all_msgs,
+            ..Default::default()
+        };
+        Ok(tx_body.encode_to_vec())
     }
 }