newton-chain-watcher 0.5.2

//! HTTP client for pushing chain events to the gateway's `/watcher` endpoint.
//!
//! Uses reqwest directly to POST JSON payloads to the gateway's `/watcher` path.
//! Optionally signs each request with HMAC-SHA256 over `timestamp\nbody` so the
//! gateway can reject forgeries (Octane #10).

use crate::event::ChainEvent;
use hmac::{Mac, SimpleHmac};
use newton_metric::record_chain_watcher_gateway_relay_duration;
use sha2::Sha256;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tracing::{error, info, warn};

const DEFAULT_MAX_RETRIES: u32 = 3;
const INITIAL_BACKOFF_MS: u64 = 100;
const MAX_BACKOFF_MS: u64 = 5000;

/// HTTP header carrying the unix-seconds timestamp the body was signed at.
pub const WATCHER_TS_HEADER: &str = "x-newton-watcher-timestamp";
/// HTTP header carrying the hex-encoded HMAC-SHA256 signature.
pub const WATCHER_SIG_HEADER: &str = "x-newton-watcher-signature";

/// client for pushing events to gateway's `/watcher` endpoint
#[derive(Debug, Clone)]
pub struct GatewayClient {
    client: reqwest::Client,
    watcher_url: String,
    max_retries: u32,
    psk: Option<Vec<u8>>,
}

impl GatewayClient {
    /// create a new gateway client without HMAC signing (back-compat for tests).
    pub fn new(watcher_url: String, max_retries: Option<u32>) -> Self {
        Self::with_psk(watcher_url, max_retries, None)
    }

    /// create a new gateway client, optionally signing each request with `psk`.
    ///
    /// `psk` is a hex-encoded shared secret (typically 32 bytes). When `None`,
    /// requests are sent unsigned and the gateway must be configured the same way.
    pub fn with_psk(watcher_url: String, max_retries: Option<u32>, psk: Option<&str>) -> Self {
        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(10))
            .pool_max_idle_per_host(10)
            .tcp_keepalive(Duration::from_secs(60))
            .tcp_nodelay(true)
            .build()
            .expect("reqwest client should build with default TLS");

        let psk = psk.and_then(|s| match hex::decode(s.trim_start_matches("0x")) {
            Ok(bytes) if !bytes.is_empty() => Some(bytes),
            Ok(_) => {
                warn!("watcher PSK decoded to empty bytes; ignoring");
                None
            }
            Err(e) => {
                warn!(error = %e, "watcher PSK is not valid hex; ignoring");
                None
            }
        });

        Self {
            client,
            watcher_url,
            max_retries: max_retries.unwrap_or(DEFAULT_MAX_RETRIES),
            psk,
        }
    }

    /// Compute `HMAC-SHA256(psk, ts || "\n" || body)` and return its hex form.
    fn sign(psk: &[u8], ts: &str, body: &[u8]) -> String {
        let mut mac = <SimpleHmac<Sha256> as Mac>::new_from_slice(psk).expect("HMAC accepts any key length");
        mac.update(ts.as_bytes());
        mac.update(b"\n");
        mac.update(body);
        hex::encode(mac.finalize().into_bytes())
    }

    /// push a chain event to the gateway's `/watcher` endpoint
    ///
    /// Retries with exponential backoff on transient errors.
    /// Returns Ok when gateway acknowledges receipt.
    pub async fn submit_event(&self, event: &ChainEvent) -> eyre::Result<()> {
        // Serialize once so signature and body bytes match exactly. `reqwest`'s
        // `.json()` would re-serialize per attempt and we'd risk producing a
        // sig over different bytes (e.g. ordering drift between retries).
        let body = serde_json::to_vec(event)?;
        let start = Instant::now();
        let mut attempt = 0u32;
        let mut delay = Duration::from_millis(INITIAL_BACKOFF_MS);

        loop {
            attempt += 1;

            let mut req = self
                .client
                .post(&self.watcher_url)
                .header(reqwest::header::CONTENT_TYPE, "application/json")
                .body(body.clone());
            if let Some(psk) = &self.psk {
                let ts = SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .map(|d| d.as_secs())
                    .unwrap_or(0)
                    .to_string();
                let sig = Self::sign(psk, &ts, &body);
                req = req.header(WATCHER_TS_HEADER, ts).header(WATCHER_SIG_HEADER, sig);
            }
            match req.send().await {
                Ok(response) => {
                    if response.status().is_success() {
                        record_chain_watcher_gateway_relay_duration(event.chain_id, start.elapsed().as_secs_f64());
                        info!(
                            chain_id = event.chain_id,
                            block_number = event.block_number,
                            attempt,
                            "event submitted to gateway"
                        );
                        return Ok(());
                    }

                    let status = response.status();
                    let body = response.text().await.unwrap_or_default();

                    // 4xx errors are not retryable (bad request, etc.)
                    if status.is_client_error() {
                        return Err(eyre::eyre!("gateway rejected event: status={}, body={}", status, body));
                    }

                    // 5xx errors are retryable
                    if attempt >= self.max_retries {
                        return Err(eyre::eyre!(
                            "gateway returned {} after {} attempts: {}",
                            status,
                            attempt,
                            body
                        ));
                    }
                    warn!(
                        chain_id = event.chain_id,
                        status = %status,
                        attempt,
                        "gateway returned server error, retrying"
                    );
                }
                Err(e) => {
                    if attempt >= self.max_retries {
                        return Err(eyre::eyre!("failed to reach gateway after {} attempts: {}", attempt, e));
                    }

                    // Network errors are retryable
                    warn!(
                        chain_id = event.chain_id,
                        attempt,
                        error = %e,
                        "failed to reach gateway, retrying"
                    );
                }
            }

            tokio::time::sleep(delay).await;
            delay = delay.saturating_mul(2).min(Duration::from_millis(MAX_BACKOFF_MS));
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event::{ChainEvent, ChainEventType};
    use alloy::primitives::{Address, FixedBytes, B256};
    use wiremock::{
        matchers::{method, path},
        Mock, MockServer, ResponseTemplate,
    };

    fn test_event() -> ChainEvent {
        ChainEvent {
            chain_id: 31337,
            event_type: ChainEventType::OperatorAdded {
                operator: Address::ZERO,
                operator_set_avs: Address::ZERO,
                operator_set_id: 0,
            },
            block_number: 100,
            tx_hash: B256::ZERO,
        }
    }

    #[tokio::test]
    async fn submit_event_succeeds_on_200() {
        let server = MockServer::start().await;

        Mock::given(method("POST"))
            .and(path("/watcher"))
            .respond_with(ResponseTemplate::new(200).set_body_string("ok"))
            .expect(1)
            .mount(&server)
            .await;

        let client = GatewayClient::new(format!("{}/watcher", server.uri()), Some(3));
        let result = client.submit_event(&test_event()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn submit_event_fails_immediately_on_4xx() {
        let server = MockServer::start().await;

        Mock::given(method("POST"))
            .and(path("/watcher"))
            .respond_with(ResponseTemplate::new(400).set_body_string("bad request"))
            // 4xx should NOT retry — expect exactly 1 call
            .expect(1)
            .mount(&server)
            .await;

        let client = GatewayClient::new(format!("{}/watcher", server.uri()), Some(3));
        let result = client.submit_event(&test_event()).await;
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("rejected"), "error should mention rejection: {err}");
    }

    #[tokio::test]
    async fn submit_event_retries_on_5xx_then_fails() {
        let server = MockServer::start().await;

        Mock::given(method("POST"))
            .and(path("/watcher"))
            .respond_with(ResponseTemplate::new(500).set_body_string("internal error"))
            // max_retries=2, so expect exactly 2 attempts
            .expect(2)
            .mount(&server)
            .await;

        let client = GatewayClient::new(format!("{}/watcher", server.uri()), Some(2));
        let result = client.submit_event(&test_event()).await;
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("after 2 attempts"),
            "error should mention attempt count: {err}"
        );
    }

    #[tokio::test]
    async fn submit_event_retries_on_network_error_then_fails() {
        // Point at a port that refuses connections
        let client = GatewayClient::new("http://127.0.0.1:1/watcher".to_string(), Some(2));
        let result = client.submit_event(&test_event()).await;
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("after 2 attempts"),
            "error should mention attempt count: {err}"
        );
    }

    #[tokio::test]
    async fn default_max_retries_is_3() {
        let client = GatewayClient::new("http://localhost:1234/watcher".to_string(), None);
        assert_eq!(client.max_retries, DEFAULT_MAX_RETRIES);
        assert_eq!(client.max_retries, 3);
    }

    #[tokio::test]
    async fn submit_event_succeeds_after_retry() {
        let server = MockServer::start().await;

        // First call: 500, second call: 200
        Mock::given(method("POST"))
            .and(path("/watcher"))
            .respond_with(ResponseTemplate::new(500).set_body_string("error"))
            .up_to_n_times(1)
            .expect(1)
            .mount(&server)
            .await;

        Mock::given(method("POST"))
            .and(path("/watcher"))
            .respond_with(ResponseTemplate::new(200).set_body_string("ok"))
            .expect(1)
            .mount(&server)
            .await;

        let client = GatewayClient::new(format!("{}/watcher", server.uri()), Some(3));
        let result = client.submit_event(&test_event()).await;
        assert!(result.is_ok());
    }
}