feat(discard): add new cell_emitter feature with cell filter feature on and he…

crates/relayer/Cargo.toml

@@ -38,6 +38,7 @@ axon-tools = { git = "https://github.com/axonweb3/axon.git", package = "axon-too
     "impl-serde",
     "proof",
 ] }
+emitter-core = { git = "https://github.com/axonweb3/emitter", rev = "315e7e8" }
 
 subtle-encoding = "0.5"
 humantime-serde = "1.1.1"

crates/relayer/src/chain/axon.rs

@@ -27,7 +27,6 @@ use crate::{
     event::{monitor::TxMonitorCmd, IbcEventWithHeight},
     ibc_contract::OwnableIBCHandlerEvents,
     keyring::{KeyRing, Secp256k1KeyPair},
-    light_client::{axon::LightClient as AxonLightClient, LightClient},
     misbehaviour::MisbehaviourEvidence,
 };
 use eth_light_client_in_ckb_prover::Receipts;
@@ -44,9 +43,7 @@ use ibc_proto::{
 use ibc_relayer_types::{
     applications::ics31_icq::response::CrossChainQueryResponse,
     clients::ics07_axon::{
-        client_state::AxonClientState,
-        consensus_state::AxonConsensusState,
-        header::{AxonHeader, AXON_HEADER_TYPE_URL},
+        client_state::AxonClientState, consensus_state::AxonConsensusState, header::AxonHeader,
         light_block::AxonLightBlock,
     },
     core::{
@@ -82,7 +79,7 @@ use ibc_relayer_types::{
 };
 use tendermint_rpc::endpoint::broadcast::tx_sync::Response;
 
-use self::{contract::OwnableIBCHandler, monitor::AxonEventMonitor};
+use self::{contract::OwnableIBCHandler, emitter::CkbSyncManager, monitor::AxonEventMonitor};
 
 type ContractProvider = SignerMiddleware<Provider<Ws>, Wallet<SigningKey>>;
 type IBCContract = OwnableIBCHandler<ContractProvider>;
@@ -112,6 +109,7 @@ use super::{
 use tokio::runtime::Runtime as TokioRuntime;
 
 pub mod contract;
+mod emitter;
 mod monitor;
 mod msg;
 mod rpc;
@@ -150,10 +148,9 @@ pub struct IBCInfoCache {
 pub struct AxonChain {
     rt: Arc<TokioRuntime>,
     config: AxonChainConfig,
-    light_client: AxonLightClient,
     tx_monitor_cmd: Option<TxMonitorCmd>,
     rpc_client: rpc::AxonRpcClient,
-    client: Provider<Ws>,
+    client: Arc<Provider<Ws>>,
     keybase: KeyRing<Secp256k1KeyPair>,
     chain_id: u64,
     ibc_cache: Arc<RwLock<IBCInfoCache>>,
@@ -168,7 +165,10 @@ impl AxonChain {
 
     fn contract_provider(&self) -> Result<Arc<ContractProvider>, Error> {
         let wallet = self.get_wallet(&self.config.key_name)?;
-        Ok(Arc::new(SignerMiddleware::new(self.client.clone(), wallet)))
+        Ok(Arc::new(SignerMiddleware::new(
+            self.client.as_ref().clone(),
+            wallet,
+        )))
     }
 
     fn contract(&self) -> Result<IBCContract, Error> {
@@ -208,30 +208,20 @@ impl ChainEndpoint for AxonChain {
 
         let url = config.websocket_addr.clone();
         let rpc_client = rpc::AxonRpcClient::new(&config.rpc_addr);
-        let client = rt
-            .block_on(Provider::<Ws>::connect(url.to_string()))
-            .map_err(|_| Error::web_socket(url.into()))?;
+        let client = Arc::new(
+            rt.block_on(Provider::<Ws>::connect(url.to_string()))
+                .map_err(|_| Error::web_socket(url.into()))?,
+        );
         let chain_id = rt
             .block_on(client.get_chainid())
             .map_err(|e| Error::other_error(e.to_string()))?
             .as_u64();
-        let light_client = AxonLightClient::from_config(&config, rt.clone())?;
         let ibc_cache = Arc::new(RwLock::new(IBCInfoCache::default()));
 
-        // TODO: since Ckb endpoint uses Axon metadata cell as its light client, Axon
-        //       endpoint has no need to monitor the update of its metadata
-        //
-        // let metadata = rt.block_on(rpc_client.get_current_metadata())?;
-        // let epoch_len = metadata.version.end - metadata.version.start + 1;
-        // light_client.bootstrap(client.clone(), rpc_client.clone(), epoch_len)?;
-
-        // FIXME remove the light client or fully implement it
-
         Ok(Self {
             rt,
             config,
             keybase,
-            light_client,
             tx_monitor_cmd: None,
             chain_id,
             rpc_client,
@@ -329,22 +319,20 @@ impl ChainEndpoint for AxonChain {
     // TODO the light client is unimplemented
     fn verify_header(
         &mut self,
-        trusted: Height,
-        target: Height,
-        client_state: &AnyClientState,
+        _trusted: Height,
+        _target: Height,
+        _client_state: &AnyClientState,
     ) -> Result<Self::LightBlock, Error> {
-        self.light_client
-            .verify(trusted, target, client_state)
-            .map(|v| v.target)
+        Ok(AxonLightBlock {})
     }
 
     // TODO the light client is unimplemented
     fn check_misbehaviour(
         &mut self,
-        update: &UpdateClient,
-        client_state: &AnyClientState,
+        _update: &UpdateClient,
+        _client_state: &AnyClientState,
     ) -> Result<Option<MisbehaviourEvidence>, Error> {
-        self.light_client.check_misbehaviour(update, client_state)
+        Ok(None)
     }
 
     fn query_balance(&self, key_name: Option<&str>, denom: Option<&str>) -> Result<Balance, Error> {
@@ -1195,15 +1183,28 @@ fn extract_path_and_base_from_full_denom(parts: Vec<&str>) -> (String, String) {
 impl AxonChain {
     fn init_event_monitor(&mut self) -> Result<TxMonitorCmd, Error> {
         crate::time!("axon_init_event_monitor");
-        // let header_receiver = self.light_client.subscribe();
+
         let ibc_cache = self.ibc_cache.clone();
+        let mut cell_process_manager = CkbSyncManager::new(
+            self.rt.clone(),
+            &self.config.emitter_ckb_url.to_string(),
+            self.chain_id,
+            self.contract_provider()?,
+            self.config.emitter_cell_start_block_number,
+        );
+
+        // start sync ckb header
+        cell_process_manager
+            .spawn_header_processor(self.config.emitter_header_start_block_number)
+            .map_err(Error::event_monitor)?;
+
         let (mut event_monitor, monitor_tx) = AxonEventMonitor::new(
             self.config.id.clone(),
             self.config.websocket_addr.clone(),
             self.config.contract_address,
-            // header_receiver,
             self.rt.clone(),
             ibc_cache.clone(),
+            cell_process_manager,
         )
         .map_err(Error::event_monitor)?;
 
@@ -1385,24 +1386,6 @@ impl AxonChain {
             create_client::TYPE_URL => {
                 convert!(self, msg, MsgCreateClient, create_client)
             }
-            // TODO: this update_client uses Hermes internal message to handle the Axon-specific function,
-            // so maybe there is possibility to create a new one to do so
-            update_client::TYPE_URL => {
-                let msg = update_client::MsgUpdateClient::from_any(msg)
-                    .map_err(|e| Error::protobuf_decode(update_client::TYPE_URL.to_string(), e))?;
-                let bytes = msg.header.value.as_slice();
-                let type_url = msg.header.type_url;
-                let to = match type_url.as_str() {
-                    AXON_HEADER_TYPE_URL => self.config.ckb_light_client_contract_address,
-                    "CELL_TYPE_URL" => self.config.image_cell_contract_address,
-                    type_url => {
-                        return Err(Error::other_error(format!("unknown type_url {type_url}")))
-                    }
-                };
-                let tx = TransactionRequest::new().to(to).data(bytes.to_vec());
-                self.rt
-                    .block_on(async { Ok(self.client.send_transaction(tx, None).await?.await?) })
-            }
             // connection
             conn_open_init::TYPE_URL => {
                 convert!(self, msg, MsgConnectionOpenInit, connection_open_init)
@@ -1559,7 +1542,7 @@ impl AxonChain {
         let mut ibc_cache = self.ibc_cache.write().unwrap();
         cache_ics_tx_hash_with_event(&mut ibc_cache, event.clone(), tx_hash);
         tracing::info!(
-            "{} transaciton {} committed to {}",
+            "🎉 {} transaciton {} committed to {}",
             event.event_type().as_str(),
             hex::encode(tx_hash),
             self.id()