Start working on stacks verify message handler.

ampd/src/config.rs

@@ -132,6 +132,12 @@ mod tests {
             type = 'StellarVerifierSetVerifier'
             cosmwasm_contract = '{}'
             http_url = 'http://localhost:8000'
+
+
+            [[handlers]]
+            type = 'StacksMsgVerifier'
+            cosmwasm_contract = '{}'
+            http_url = 'http://localhost:8000'
             ",
             TMAddress::random(PREFIX),
             TMAddress::random(PREFIX),
@@ -143,10 +149,11 @@ mod tests {
             TMAddress::random(PREFIX),
             TMAddress::random(PREFIX),
             TMAddress::random(PREFIX),
+            TMAddress::random(PREFIX),
         );
 
         let cfg: Config = toml::from_str(config_str.as_str()).unwrap();
-        assert_eq!(cfg.handlers.len(), 10);
+        assert_eq!(cfg.handlers.len(), 11);
     }
 
     #[test]
@@ -350,6 +357,12 @@ mod tests {
                     ),
                     http_url: Url::from_str("http://127.0.0.1").unwrap(),
                 },
+                HandlerConfig::StacksMsgVerifier {
+                    cosmwasm_contract: TMAddress::from(
+                        AccountId::new("axelar", &[0u8; 32]).unwrap(),
+                    ),
+                    http_url: Url::from_str("http://127.0.0.1").unwrap(),
+                },
             ],
             ..Config::default()
         }

ampd/src/handlers/config.rs

@@ -63,6 +63,10 @@ pub enum Config {
         cosmwasm_contract: TMAddress,
         http_url: Url,
     },
+    StacksMsgVerifier {
+        cosmwasm_contract: TMAddress,
+        http_url: Url,
+    },
 }
 
 fn validate_evm_verifier_set_verifier_configs<'de, D>(configs: &[Config]) -> Result<(), D::Error>
@@ -159,6 +163,7 @@ where
         Config::StellarVerifierSetVerifier,
         "Stellar verifier set verifier"
     )?;
+    ensure_unique_config!(&configs, Config::StacksMsgVerifier, "Stacks message verifier")?;
 
     Ok(configs)
 }
@@ -305,5 +310,22 @@ mod tests {
                 Err(e) if e.to_string().contains("only one Stellar verifier set verifier config is allowed")
             )
         );
+
+        let configs = vec![
+            Config::StacksMsgVerifier {
+                cosmwasm_contract: TMAddress::random(PREFIX),
+                http_url: "http://localhost:8080/".parse().unwrap(),
+            },
+            Config::StacksMsgVerifier {
+                cosmwasm_contract: TMAddress::random(PREFIX),
+                http_url: "http://localhost:8080/".parse().unwrap(),
+            },
+        ];
+
+        assert!(
+            matches!(deserialize_handler_configs(to_value(configs).unwrap()),
+                Err(e) if e.to_string().contains("only one Stacks message verifier config is allowed")
+            )
+        );
     }
 }

ampd/src/handlers/mod.rs

@@ -5,6 +5,7 @@ pub mod evm_verify_verifier_set;
 pub mod multisig;
 pub mod mvx_verify_msg;
 pub mod mvx_verify_verifier_set;
+pub mod stacks_verify_msg;
 pub(crate) mod stellar_verify_msg;
 pub(crate) mod stellar_verify_verifier_set;
 pub mod sui_verify_msg;

ampd/src/handlers/stacks_verify_msg.rs

@@ -0,0 +1,135 @@
+use std::collections::HashSet;
+use std::convert::TryInto;
+
+use async_trait::async_trait;
+use axelar_wasm_std::voting::{PollId, Vote};
+use cosmrs::cosmwasm::MsgExecuteContract;
+use cosmrs::tx::Msg;
+use cosmrs::Any;
+use error_stack::ResultExt;
+use events::Error::EventTypeMismatch;
+use events::Event;
+use events_derive::try_from;
+use serde::Deserialize;
+use tokio::sync::watch::Receiver;
+use tracing::info;
+use voting_verifier::msg::ExecuteMsg;
+
+use crate::event_processor::EventHandler;
+use crate::handlers::errors::Error;
+use crate::stacks::http_client::Client;
+use crate::stacks::verifier::verify_message;
+use crate::types::{Hash, TMAddress};
+
+type Result<T> = error_stack::Result<T, Error>;
+
+#[derive(Deserialize, Debug)]
+pub struct Message {
+    pub tx_id: Hash,
+    pub event_index: u32,
+    pub destination_address: String,
+    pub destination_chain: router_api::ChainName,
+    pub source_address: String, // TODO
+    pub payload_hash: Hash,
+}
+
+#[derive(Deserialize, Debug)]
+#[try_from("wasm-messages_poll_started")]
+struct PollStartedEvent {
+    poll_id: PollId,
+    source_gateway_address: String, // TODO
+    messages: Vec<Message>,
+    participants: Vec<TMAddress>,
+    expires_at: u64,
+}
+
+pub struct Handler {
+    verifier: TMAddress,
+    voting_verifier_contract: TMAddress,
+    http_client: Client,
+    latest_block_height: Receiver<u64>,
+}
+
+impl Handler {
+    pub fn new(
+        verifier: TMAddress,
+        voting_verifier_contract: TMAddress,
+        http_client: Client,
+        latest_block_height: Receiver<u64>,
+    ) -> Self {
+        Self {
+            verifier,
+            voting_verifier_contract,
+            http_client,
+            latest_block_height,
+        }
+    }
+
+    fn vote_msg(&self, poll_id: PollId, votes: Vec<Vote>) -> MsgExecuteContract {
+        MsgExecuteContract {
+            sender: self.verifier.as_ref().clone(),
+            contract: self.voting_verifier_contract.as_ref().clone(),
+            msg: serde_json::to_vec(&ExecuteMsg::Vote { poll_id, votes })
+                .expect("vote msg should serialize"),
+            funds: vec![],
+        }
+    }
+}
+
+#[async_trait]
+impl EventHandler for Handler {
+    type Err = Error;
+
+    async fn handle(&self, event: &Event) -> Result<Vec<Any>> {
+        if !event.is_from_contract(self.voting_verifier_contract.as_ref()) {
+            return Ok(vec![]);
+        }
+
+        let PollStartedEvent {
+            poll_id,
+            source_gateway_address,
+            messages,
+            participants,
+            expires_at,
+            ..
+        } = match event.try_into() as error_stack::Result<_, _> {
+            Err(report) if matches!(report.current_context(), EventTypeMismatch(_)) => {
+                return Ok(vec![]);
+            }
+            event => event.change_context(Error::DeserializeEvent)?,
+        };
+
+        if !participants.contains(&self.verifier) {
+            return Ok(vec![]);
+        }
+
+        let latest_block_height = *self.latest_block_height.borrow();
+        if latest_block_height >= expires_at {
+            info!(poll_id = poll_id.to_string(), "skipping expired poll");
+
+            return Ok(vec![]);
+        }
+
+        let tx_hashes: HashSet<_> = messages.iter().map(|message| message.tx_id).collect();
+        let transactions_info = self
+            .http_client
+            .get_transactions(tx_hashes)
+            .await;
+
+        let votes: Vec<Vote> = messages
+            .iter()
+            .map(|msg| {
+                transactions_info
+                    .get(&msg.tx_id)
+                    .map_or(Vote::NotFound, |transaction| {
+                        verify_message(&source_gateway_address, transaction, msg)
+                    })
+            })
+            .collect();
+
+        Ok(vec![self
+            .vote_msg(poll_id, votes)
+            .into_any()
+            .expect("vote msg should serialize")])
+    }
+}

ampd/src/lib.rs

@@ -40,6 +40,7 @@ mod health_check;
 mod json_rpc;
 mod mvx;
 mod queue;
+mod stacks;
 mod stellar;
 mod sui;
 mod tm_client;
@@ -51,6 +52,7 @@ pub use grpc::{client, proto};
 
 use crate::asyncutil::future::RetryPolicy;
 use crate::broadcaster::confirm_tx::TxConfirmer;
+use crate::stacks::http_client::Client;
 
 const PREFIX: &str = "axelar";
 const DEFAULT_RPC_TIMEOUT: Duration = Duration::from_secs(3);
@@ -388,6 +390,19 @@ where
                     ),
                     event_processor_config.clone(),
                 ),
+                handlers::config::Config::StacksMsgVerifier {
+                    cosmwasm_contract,
+                    http_url,
+                } => self.create_handler_task(
+                    "stacks-msg-verifier",
+                    handlers::stacks_verify_msg::Handler::new(
+                        verifier.clone(),
+                        cosmwasm_contract,
+                        Client::new_http(http_url.to_string().trim_end_matches('/').into()),
+                        self.block_height_monitor.latest_block_height(),
+                    ),
+                    event_processor_config.clone(),
+                ),
             };
             self.event_processor = self.event_processor.add_task(task);
         }