newton-chain-watcher 0.5.2

//! Smart filter for distinguishing gateway-originated vs direct on-chain tasks
//!
//! Uses a Redis SET to track task IDs that the gateway has already broadcast.
//! Tasks found in the SET are skipped; tasks NOT in the SET are relayed to the
//! gateway as direct on-chain tasks.

use alloy::primitives::FixedBytes;
use newton_metric::{inc_chain_watcher_redis_lookup, record_chain_watcher_filter_lookup_duration};
use redis::{aio::ConnectionManager, AsyncCommands, Client};
use std::time::Instant;
use tracing::{debug, warn};

/// Redis key prefix for the seen-tasks SET per chain
const SEEN_TASKS_KEY_PREFIX: &str = "newton:seen-tasks";

/// TTL for the seen-tasks SET (1 hour)
const SEEN_TASKS_TTL_SECS: u64 = 3600;

/// smart filter that checks Redis SET for task IDs already seen by the gateway
pub struct TaskFilter {
    chain_id: u64,
    conn: ConnectionManager,
}

impl std::fmt::Debug for TaskFilter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TaskFilter")
            .field("chain_id", &self.chain_id)
            .field("redis", &"connected")
            .finish()
    }
}

impl TaskFilter {
    /// create a new task filter backed by Redis
    ///
    /// # Errors
    ///
    /// Returns error if Redis connection cannot be established
    pub async fn new(redis_url: &str, chain_id: u64) -> eyre::Result<Self> {
        let client = Client::open(redis_url)?;
        let conn = ConnectionManager::new(client).await?;
        Ok(Self { chain_id, conn })
    }

    /// check if a task was already handled by the gateway (exists in Redis SET)
    pub async fn is_seen_by_gateway(&self, task_id: &FixedBytes<32>) -> bool {
        let key = format!("{}:{}", SEEN_TASKS_KEY_PREFIX, self.chain_id);
        let mut conn = self.conn.clone();
        let lookup_start = Instant::now();
        let result = redis::cmd("SISMEMBER")
            .arg(&key)
            .arg(task_id.as_slice())
            .query_async::<bool>(&mut conn)
            .await;
        // Record on all outcomes (hit, miss, error) so dashboards can compare
        // the latency distribution across Redis success and failure paths.
        record_chain_watcher_filter_lookup_duration(self.chain_id, lookup_start.elapsed().as_secs_f64());
        match result {
            Ok(exists) => {
                inc_chain_watcher_redis_lookup(self.chain_id, if exists { "hit" } else { "miss" });
                debug!(
                    chain_id = self.chain_id,
                    task_id = %task_id,
                    seen = exists,
                    "task filter lookup"
                );
                exists
            }
            Err(e) => {
                inc_chain_watcher_redis_lookup(self.chain_id, "error");
                // Redis failure is non-fatal — assume task is NOT seen,
                // gateway will deduplicate via its DashMap if it's a duplicate
                warn!(
                    chain_id = self.chain_id,
                    task_id = %task_id,
                    error = %e,
                    "redis SISMEMBER failed, treating as unseen"
                );
                false
            }
        }
    }

    /// check if a task is a direct on-chain task (NOT seen by gateway)
    pub async fn is_direct_onchain(&self, task_id: &FixedBytes<32>) -> bool {
        !self.is_seen_by_gateway(task_id).await
    }

    /// Marks a task as seen by the gateway.
    ///
    /// In production, the gateway writes directly via `mark_task_seen_by_watcher()`.
    /// This method is exposed for testing and cross-component integration.
    pub async fn mark_seen(&self, task_id: &FixedBytes<32>) -> eyre::Result<()> {
        let key = format!("{}:{}", SEEN_TASKS_KEY_PREFIX, self.chain_id);
        let mut conn = self.conn.clone();
        conn.sadd::<_, _, ()>(&key, task_id.as_slice()).await?;
        conn.expire::<_, ()>(&key, SEEN_TASKS_TTL_SECS as i64).await?;
        Ok(())
    }

    /// returns the Redis key used for a given chain id
    pub fn redis_key(chain_id: u64) -> String {
        format!("{}:{}", SEEN_TASKS_KEY_PREFIX, chain_id)
    }

    /// returns the TTL in seconds for the seen-tasks SET
    pub const fn ttl_secs() -> u64 {
        SEEN_TASKS_TTL_SECS
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn redis_key_format() {
        assert_eq!(TaskFilter::redis_key(31337), "newton:seen-tasks:31337");
        assert_eq!(TaskFilter::redis_key(1), "newton:seen-tasks:1");
    }

    #[test]
    fn ttl_is_one_hour() {
        assert_eq!(TaskFilter::ttl_secs(), 3600);
    }

    /// Requires a running Redis instance at localhost:6379.
    /// Run with: cargo test -p newton-chain-watcher -- --ignored
    #[tokio::test]
    #[ignore]
    async fn mark_seen_then_is_seen_by_gateway_returns_true() {
        let filter = TaskFilter::new("redis://:redis@localhost:6379", 99999)
            .await
            .expect("redis connection");

        let task_id = FixedBytes::from([0xABu8; 32]);

        // Clean up from prior runs
        let key = TaskFilter::redis_key(99999);
        let mut conn = filter.conn.clone();
        let _: Result<(), _> = redis::cmd("DEL").arg(&key).query_async(&mut conn).await;

        // Before marking: task should not be seen
        assert!(!filter.is_seen_by_gateway(&task_id).await);
        assert!(filter.is_direct_onchain(&task_id).await);

        // Mark as seen
        filter.mark_seen(&task_id).await.expect("mark_seen should succeed");

        // After marking: task should be seen
        assert!(filter.is_seen_by_gateway(&task_id).await);
        assert!(!filter.is_direct_onchain(&task_id).await);

        // Clean up
        let _: Result<(), _> = redis::cmd("DEL").arg(&key).query_async(&mut conn).await;
    }

    /// Verifies that unseen tasks are treated as direct on-chain tasks
    #[tokio::test]
    #[ignore]
    async fn unseen_task_is_direct_onchain() {
        let filter = TaskFilter::new("redis://:redis@localhost:6379", 99998)
            .await
            .expect("redis connection");

        // Clean slate
        let key = TaskFilter::redis_key(99998);
        let mut conn = filter.conn.clone();
        let _: Result<(), _> = redis::cmd("DEL").arg(&key).query_async(&mut conn).await;

        let task_id = FixedBytes::from([0xCDu8; 32]);
        assert!(filter.is_direct_onchain(&task_id).await);

        // Clean up
        let _: Result<(), _> = redis::cmd("DEL").arg(&key).query_async(&mut conn).await;
    }

    /// Verifies that different chain IDs use separate Redis keys
    #[tokio::test]
    #[ignore]
    async fn different_chains_have_separate_sets() {
        let filter_a = TaskFilter::new("redis://:redis@localhost:6379", 88881)
            .await
            .expect("redis connection");
        let filter_b = TaskFilter::new("redis://:redis@localhost:6379", 88882)
            .await
            .expect("redis connection");

        let task_id = FixedBytes::from([0xEFu8; 32]);

        // Clean up
        let key_a = TaskFilter::redis_key(88881);
        let key_b = TaskFilter::redis_key(88882);
        let mut conn = filter_a.conn.clone();
        let _: Result<(), _> = redis::cmd("DEL").arg(&key_a).query_async(&mut conn).await;
        let _: Result<(), _> = redis::cmd("DEL").arg(&key_b).query_async(&mut conn).await;

        // Mark on chain A only
        filter_a.mark_seen(&task_id).await.expect("mark_seen");

        // Seen on A, NOT seen on B
        assert!(filter_a.is_seen_by_gateway(&task_id).await);
        assert!(!filter_b.is_seen_by_gateway(&task_id).await);

        // Clean up
        let _: Result<(), _> = redis::cmd("DEL").arg(&key_a).query_async(&mut conn).await;
        let _: Result<(), _> = redis::cmd("DEL").arg(&key_b).query_async(&mut conn).await;
    }

    /// Cross-component test: simulates the gateway's `mark_task_seen_by_watcher()`
    /// writing via raw `SADD` (the same Redis commands the gateway uses), then verifies
    /// the chain watcher's `TaskFilter` reads correctly via `SISMEMBER`.
    #[tokio::test]
    #[ignore]
    async fn gateway_sadd_compatible_with_filter_sismember() {
        let chain_id = 77777u64;
        let key = TaskFilter::redis_key(chain_id);
        let task_id = FixedBytes::from([0x42u8; 32]);

        // --- Simulate gateway side (raw Redis, same pattern as ChainService::mark_task_seen_by_watcher) ---
        let client = redis::Client::open("redis://:redis@localhost:6379").expect("redis client");
        let mut gateway_conn = ConnectionManager::new(client).await.expect("redis conn");

        // Clean up
        let _: Result<(), _> = redis::cmd("DEL").arg(&key).query_async(&mut gateway_conn).await;

        // Gateway writes: SADD + EXPIRE (mirrors handler/mod.rs:124-142)
        let _: () = redis::cmd("SADD")
            .arg(&key)
            .arg(task_id.as_slice())
            .query_async(&mut gateway_conn)
            .await
            .expect("SADD");
        let _: () = redis::cmd("EXPIRE")
            .arg(&key)
            .arg(3600i64)
            .query_async(&mut gateway_conn)
            .await
            .expect("EXPIRE");

        // --- Watcher side: TaskFilter reads via SISMEMBER ---
        let filter = TaskFilter::new("redis://:redis@localhost:6379", chain_id)
            .await
            .expect("filter");

        assert!(
            filter.is_seen_by_gateway(&task_id).await,
            "filter should detect task written by gateway's SADD"
        );
        assert!(
            !filter.is_direct_onchain(&task_id).await,
            "gateway-originated task should NOT be treated as direct on-chain"
        );

        // Unseen task should still be direct
        let unseen = FixedBytes::from([0x99u8; 32]);
        assert!(filter.is_direct_onchain(&unseen).await);

        // Clean up
        let _: Result<(), _> = redis::cmd("DEL").arg(&key).query_async(&mut gateway_conn).await;
    }
}