openzeppelin_relayer/utils/

redis.rs

use std::sync::Arc;
use std::time::Duration;

use color_eyre::Result;
use deadpool_redis::{Config, Metrics, Pool, Runtime};
use tracing::{debug, info, warn};

use crate::config::ServerConfig;

// ============================================================================
// Shared Timing Constants for Bootstrap Operations
// ============================================================================

/// Default lock TTL for bootstrap operations (2 minutes).
/// Used as a safety net for crashes during initialization or config processing.
pub const BOOTSTRAP_LOCK_TTL_SECS: u64 = 120;

/// Max wait time when another instance holds the lock.
/// Set slightly longer than lock TTL to handle edge cases.
pub const LOCK_WAIT_MAX_SECS: u64 = 130;

/// Polling interval when waiting for completion.
pub const LOCK_POLL_INTERVAL_MS: u64 = 500;

/// Holds separate connection pools for read and write operations.
///
/// This struct enables optimization for Redis deployments with read replicas,
/// such as AWS ElastiCache. Write operations use the primary pool, while read
/// operations can be distributed across reader replicas.
///
/// When `REDIS_READER_URL` is not configured, both pools point to the same
/// Redis instance (the primary), maintaining backward compatibility.
#[derive(Clone, Debug)]
pub struct RedisConnections {
    /// Pool for write operations (connected to primary endpoint)
    primary_pool: Arc<Pool>,
    /// Pool for read operations (connected to reader endpoint, or primary if not configured)
    reader_pool: Arc<Pool>,
}

impl RedisConnections {
    /// Creates a new `RedisConnections` with a single pool used for both reads and writes.
    ///
    /// This is useful for:
    /// - Testing where read/write separation is not needed
    /// - Simple deployments without read replicas
    /// - Backward compatibility
    pub fn new_single_pool(pool: Arc<Pool>) -> Self {
        Self {
            primary_pool: pool.clone(),
            reader_pool: pool,
        }
    }

    /// Returns the primary pool for write operations.
    ///
    /// Use this for: `create`, `update`, `delete`, and any operation that
    /// modifies data in Redis.
    pub fn primary(&self) -> &Arc<Pool> {
        &self.primary_pool
    }

    /// Returns the reader pool for read operations.
    ///
    /// Use this for: `get_by_id`, `list`, `find_by_*`, `count`, and any
    /// operation that only reads data from Redis.
    ///
    /// If no reader endpoint is configured, this returns the same pool as `primary()`.
    pub fn reader(&self) -> &Arc<Pool> {
        &self.reader_pool
    }
}

/// Creates a Redis connection pool with the specified URL, pool size, and configuration.
async fn create_pool(url: &str, pool_max_size: usize, config: &ServerConfig) -> Result<Arc<Pool>> {
    let cfg = Config::from_url(url);

    let pool = cfg
        .builder()
        .map_err(|e| eyre::eyre!("Failed to create Redis pool builder for {}: {}", url, e))?
        .max_size(pool_max_size)
        .wait_timeout(Some(Duration::from_millis(config.redis_pool_timeout_ms)))
        .create_timeout(Some(Duration::from_millis(
            config.redis_connection_timeout_ms,
        )))
        .recycle_timeout(Some(Duration::from_millis(
            config.redis_connection_timeout_ms,
        )))
        .runtime(Runtime::Tokio1)
        .build()
        .map_err(|e| eyre::eyre!("Failed to build Redis pool for {}: {}", url, e))?;

    // Verify the pool is working by getting a connection
    let conn = pool
        .get()
        .await
        .map_err(|e| eyre::eyre!("Failed to get initial Redis connection from {}: {}", url, e))?;
    drop(conn);

    Ok(Arc::new(pool))
}

/// Initializes Redis connection pools for both primary and reader endpoints.
///
/// # Arguments
///
/// * `config` - The server configuration containing Redis URLs and pool settings.
///
/// # Returns
///
/// A `RedisConnections` struct containing:
/// - `primary_pool`: Connected to `REDIS_URL` (for write operations)
/// - `reader_pool`: Connected to `REDIS_READER_URL` if set, otherwise same as primary
///
/// # Features
///
/// - **Read/Write Separation**: When `REDIS_READER_URL` is configured, read operations
///   can be distributed across read replicas, reducing load on the primary.
/// - **Backward Compatible**: If `REDIS_READER_URL` is not set, both pools use
///   the primary URL, maintaining existing behavior.
/// - **Connection Pooling**: Both pools use deadpool-redis with configurable
///   max size, wait timeout, and connection timeouts.
///
/// # Example Configuration
///
/// ```bash
/// # AWS ElastiCache with read replicas
/// REDIS_URL=redis://my-cluster.xxx.cache.amazonaws.com:6379
/// REDIS_READER_URL=redis://my-cluster-ro.xxx.cache.amazonaws.com:6379
/// ```
pub async fn initialize_redis_connections(config: &ServerConfig) -> Result<Arc<RedisConnections>> {
    let primary_pool_size = config.redis_pool_max_size;
    let primary_pool = create_pool(&config.redis_url, primary_pool_size, config).await?;

    info!(
        primary_url = %config.redis_url,
        primary_pool_size = %primary_pool_size,
        "initializing primary Redis connection pool"
    );
    let reader_pool = match &config.redis_reader_url {
        Some(reader_url) if !reader_url.is_empty() => {
            let reader_pool_size = config.redis_reader_pool_max_size;

            info!(
                primary_url = %config.redis_url,
                reader_url = %reader_url,
                primary_pool_size = %primary_pool_size,
                reader_pool_size = %reader_pool_size,
                "Using separate reader endpoint for read operations"
            );
            create_pool(reader_url, reader_pool_size, config).await?
        }
        _ => {
            debug!(
                primary_url = %config.redis_url,
                pool_size = %primary_pool_size,
                "No reader URL configured, using primary for all operations"
            );
            primary_pool.clone()
        }
    };

    // Bound pool connection lifetimes so fresh connections periodically
    // re-resolve the endpoint DNS and follow it to its current node. A
    // separately configured reader endpoint can repoint the same way (node
    // replacement, scaling, blue/green), so a distinct reader pool gets its
    // own recycler; without one, `reader_pool` aliases `primary_pool` and a
    // single recycler covers both.
    spawn_pool_age_recycler(&primary_pool, config.redis_connection_max_age_ms, "primary");
    if !Arc::ptr_eq(&reader_pool, &primary_pool) {
        spawn_pool_age_recycler(&reader_pool, config.redis_connection_max_age_ms, "reader");
    }

    Ok(Arc::new(RedisConnections {
        primary_pool,
        reader_pool,
    }))
}

/// Age-eviction predicate for `Pool::retain`: keeps a connection while its age
/// is under `max_age`, dropping it once it reaches the bound.
fn should_retain_connection(metrics: &Metrics, max_age: Duration) -> bool {
    metrics.created.elapsed() < max_age
}

/// Spawns a background task that evicts pool connections older than
/// `max_age_ms` via `Pool::retain`; deadpool then lazily opens fresh
/// connections on demand, which re-resolve DNS. `max_age_ms == 0` disables
/// recycling and spawns nothing.
///
/// The sweep runs at half the max age, clamped to `[1s, 30s]`, so a
/// connection can outlive the bound by up to one sweep interval — and for
/// `max_age_ms` under ~2000 the 1s floor means the effective lifetime bound
/// is roughly one second, not the configured value.
///
/// The task holds only a `Weak` reference to the pool and exits once the
/// owning `RedisConnections` (and any checked-out connections) are dropped.
///
/// (`Pool::retain` is used rather than a custom `Manager` because deadpool-redis
/// 0.22 has no hook to swap the manager without changing the concrete `Pool`
/// type that `RedisConnections` holds.)
fn spawn_pool_age_recycler(pool: &Arc<Pool>, max_age_ms: u64, pool_label: &'static str) {
    if max_age_ms == 0 {
        return;
    }

    let max_age = Duration::from_millis(max_age_ms);
    let pool = Arc::downgrade(pool);

    // Sweep at roughly half the max age (bounded to a sane range) so an aged
    // connection is dropped well within one max-age window without hammering.
    let interval = max_age
        .div_f64(2.0)
        .clamp(Duration::from_secs(1), Duration::from_secs(30));

    tokio::spawn(async move {
        loop {
            tokio::time::sleep(interval).await;
            let Some(pool) = pool.upgrade() else {
                break;
            };
            let result = pool.retain(|_, metrics| should_retain_connection(&metrics, max_age));
            if !result.removed.is_empty() {
                debug!(
                    pool = pool_label,
                    removed = result.removed.len(),
                    retained = result.retained,
                    max_age_ms = max_age_ms,
                    "recycled aged Redis connections"
                );
            }
        }
    });
}

/// A distributed lock implementation using Redis SET NX EX pattern.
///
/// This lock is designed for distributed systems where multiple instances
/// need to coordinate exclusive access to shared resources. It uses the
/// Redis SET command with NX (only set if not exists) and EX (expiry time)
/// options to atomically acquire locks.
///
/// # Features
/// - **Atomic acquisition**: Uses Redis SET NX to ensure only one instance can acquire the lock
/// - **Automatic expiry**: Lock automatically expires after TTL to prevent deadlocks
/// - **Unique identifiers**: Each lock acquisition gets a unique ID to prevent accidental release by other instances
/// - **Auto-release on drop**: Lock is released when the guard is dropped (best-effort via spawned task)
///
/// # TTL Considerations
/// The TTL should be set to accommodate the worst-case runtime of the protected operation.
/// If the operation runs longer than TTL, another instance may acquire the lock concurrently.
/// For long-running operations, consider:
/// - Setting a generous TTL (e.g., 2x expected runtime)
/// - Implementing lock refresh/extension during processing
/// - Breaking the operation into smaller locked sections
///
/// # Example
/// ```ignore
/// // Key format: {prefix}:lock:{name}
/// let lock_key = format!("{}:lock:{}", prefix, "my-operation");
/// let lock = DistributedLock::new(redis_client, &lock_key, Duration::from_secs(60));
/// if let Some(guard) = lock.try_acquire().await? {
///     // Do exclusive work here
///     // Lock is automatically released when guard is dropped
/// }
/// ```
#[derive(Clone)]
pub struct DistributedLock {
    pool: Arc<Pool>,
    lock_key: String,
    ttl: Duration,
}

impl DistributedLock {
    /// Creates a new distributed lock instance.
    ///
    /// # Arguments
    /// * `pool` - Redis connection pool
    /// * `lock_key` - Full Redis key for this lock (e.g., "myprefix:lock:cleanup")
    /// * `ttl` - Time-to-live for the lock. Lock will automatically expire after this duration
    ///   to prevent deadlocks if the holder crashes.
    pub fn new(pool: Arc<Pool>, lock_key: &str, ttl: Duration) -> Self {
        Self {
            pool,
            lock_key: lock_key.to_string(),
            ttl,
        }
    }

    /// Attempts to acquire the distributed lock.
    ///
    /// This is a non-blocking operation. If the lock is already held by another
    /// instance, it returns `Ok(None)` immediately without waiting.
    ///
    /// # Returns
    /// * `Ok(Some(LockGuard))` - Lock was successfully acquired
    /// * `Ok(None)` - Lock is already held by another instance
    /// * `Err(_)` - Redis communication error
    ///
    /// # Lock Semantics
    /// The lock is implemented using Redis `SET key value NX EX ttl`:
    /// - `NX`: Only set if key does not exist (atomic check-and-set)
    /// - `EX`: Set expiry time in seconds
    pub async fn try_acquire(&self) -> Result<Option<LockGuard>> {
        let lock_value = generate_lock_value();
        let mut conn = self.pool.get().await?;

        // Use SET NX EX for atomic lock acquisition with expiry
        let result: Option<String> = redis::cmd("SET")
            .arg(&self.lock_key)
            .arg(&lock_value)
            .arg("NX")
            .arg("EX")
            .arg(self.ttl.as_secs())
            .query_async(&mut conn)
            .await?;

        match result {
            Some(_) => {
                debug!(
                    lock_key = %self.lock_key,
                    ttl_secs = %self.ttl.as_secs(),
                    "distributed lock acquired"
                );
                Ok(Some(LockGuard {
                    release_info: Some(LockReleaseInfo {
                        pool: self.pool.clone(),
                        lock_key: self.lock_key.clone(),
                        lock_value,
                    }),
                }))
            }
            None => {
                debug!(
                    lock_key = %self.lock_key,
                    "distributed lock already held by another instance"
                );
                Ok(None)
            }
        }
    }
}

/// A guard that represents an acquired distributed lock.
///
/// The lock is automatically released when this guard is dropped. This ensures
/// the lock is released regardless of how the protected code exits (normal return,
/// early return via `?`, or panic).
///
/// The release is performed via a spawned task to avoid blocking in `Drop`.
/// If you need to confirm the release succeeded, call `release()` explicitly instead.
///
/// # Drop Behavior
/// When dropped, the guard spawns an async task to release the lock. This is
/// best-effort: if the task fails (e.g., Redis unavailable), the lock will
/// still expire after TTL.
pub struct LockGuard {
    /// Release info is wrapped in Option to support both explicit release and Drop.
    /// When `release()` is called, this is taken (set to None) to prevent double-release.
    /// When Drop runs, if this is Some, we spawn a release task.
    release_info: Option<LockReleaseInfo>,
}

/// Internal struct holding the information needed to release a lock.
struct LockReleaseInfo {
    pool: Arc<Pool>,
    lock_key: String,
    lock_value: String,
}

impl LockGuard {
    /// Explicitly releases the lock before TTL expiry.
    ///
    /// This uses a Lua script to ensure we only delete the lock if we still own it.
    /// This prevents accidentally releasing a lock that was already expired and
    /// re-acquired by another instance.
    ///
    /// Calling this method consumes the guard and prevents the Drop-based release.
    ///
    /// # Returns
    /// * `Ok(true)` - Lock was successfully released
    /// * `Ok(false)` - Lock was not released (already expired or owned by another instance)
    /// * `Err(_)` - Redis communication error
    pub async fn release(mut self) -> Result<bool> {
        // Take the release info to prevent Drop from also trying to release
        let info = self
            .release_info
            .take()
            .expect("release_info should always be Some before release");

        Self::do_release(info).await
    }

    /// Internal async release implementation.
    async fn do_release(info: LockReleaseInfo) -> Result<bool> {
        let mut conn = info.pool.get().await?;

        // Lua script to atomically check and delete
        // Only delete if the value matches (we still own the lock)
        let script = r#"
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("DEL", KEYS[1])
            else
                return 0
            end
        "#;

        let result: i32 = redis::Script::new(script)
            .key(&info.lock_key)
            .arg(&info.lock_value)
            .invoke_async(&mut conn)
            .await?;

        if result == 1 {
            debug!(lock_key = %info.lock_key, "distributed lock released");
            Ok(true)
        } else {
            warn!(
                lock_key = %info.lock_key,
                "distributed lock release failed - lock already expired or owned by another instance"
            );
            Ok(false)
        }
    }
}

impl Drop for LockGuard {
    fn drop(&mut self) {
        // If release_info is still Some, we need to release the lock
        if let Some(info) = self.release_info.take() {
            // Spawn a task to release the lock asynchronously
            // This is best-effort; if it fails, the lock will expire after TTL
            tokio::spawn(async move {
                if let Err(e) = LockGuard::do_release(info).await {
                    warn!(error = %e, "failed to release distributed lock in drop, will expire after TTL");
                }
            });
        }
    }
}

/// Generates a unique value for lock ownership verification.
///
/// Uses a combination of process ID, timestamp, and a monotonic counter
/// to create a unique identifier for this lock acquisition. This avoids
/// collisions in the same process even when calls share a timestamp.
fn generate_lock_value() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};

    static LOCK_VALUE_COUNTER: AtomicU64 = AtomicU64::new(0);
    let process_id = std::process::id();
    let timestamp = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let counter = LOCK_VALUE_COUNTER.fetch_add(1, Ordering::Relaxed);

    format!("{process_id}:{timestamp}:{counter}")
}

// ============================================================================
// Relayer Sync Metadata Functions
// ============================================================================
//
// These functions track when relayers were last synchronized/initialized.
// This allows multiple instances to skip redundant initialization when
// a relayer was recently synced by another instance.

/// The Redis hash key suffix for storing relayer sync metadata.
const RELAYER_SYNC_META_KEY: &str = "relayer_sync_meta";

/// The hash field for global initialization completion timestamp.
const GLOBAL_INIT_FIELD: &str = "global:init_completed";

/// The Redis hash key suffix for storing bootstrap process metadata.
const BOOTSTRAP_META_KEY: &str = "bootstrap_meta";

/// Hash field storing config processing status.
const CONFIG_PROCESSING_STATUS_FIELD: &str = "config_processing:status";

/// Hash field storing config processing start timestamp.
const CONFIG_PROCESSING_STARTED_AT_FIELD: &str = "config_processing:started_at";

/// Hash field storing config processing completion timestamp.
const CONFIG_PROCESSING_COMPLETED_AT_FIELD: &str = "config_processing:completed_at";

const CONFIG_PROCESSING_STATUS_IN_PROGRESS: &str = "in_progress";
const CONFIG_PROCESSING_STATUS_COMPLETED: &str = "completed";

/// Marks config processing as in progress.
///
/// This explicit marker is used by distributed bootstrap coordination so
/// waiters do not infer completion from partially written repository state.
pub async fn set_config_processing_in_progress(pool: &Arc<Pool>, prefix: &str) -> Result<()> {
    use chrono::Utc;
    use redis::AsyncCommands;

    let mut conn = pool.get().await?;
    let hash_key = format!("{prefix}:{BOOTSTRAP_META_KEY}");
    let timestamp = Utc::now().timestamp();

    conn.hset_multiple::<_, _, _, ()>(
        &hash_key,
        &[
            (
                CONFIG_PROCESSING_STATUS_FIELD,
                CONFIG_PROCESSING_STATUS_IN_PROGRESS,
            ),
            (CONFIG_PROCESSING_STARTED_AT_FIELD, &timestamp.to_string()),
        ],
    )
    .await
    .map_err(|e| eyre::eyre!("Failed to set config processing in progress: {}", e))?;

    conn.hdel::<_, _, ()>(&hash_key, CONFIG_PROCESSING_COMPLETED_AT_FIELD)
        .await
        .map_err(|e| eyre::eyre!("Failed to clear config processing completion time: {}", e))?;

    debug!(
        timestamp = %timestamp,
        "recorded config processing in-progress marker"
    );

    Ok(())
}

/// Marks config processing as completed.
pub async fn set_config_processing_completed(pool: &Arc<Pool>, prefix: &str) -> Result<()> {
    use chrono::Utc;
    use redis::AsyncCommands;

    let mut conn = pool.get().await?;
    let hash_key = format!("{prefix}:{BOOTSTRAP_META_KEY}");
    let timestamp = Utc::now().timestamp();

    conn.hset_multiple::<_, _, _, ()>(
        &hash_key,
        &[
            (
                CONFIG_PROCESSING_STATUS_FIELD,
                CONFIG_PROCESSING_STATUS_COMPLETED,
            ),
            (CONFIG_PROCESSING_COMPLETED_AT_FIELD, &timestamp.to_string()),
        ],
    )
    .await
    .map_err(|e| eyre::eyre!("Failed to set config processing completed: {}", e))?;

    debug!(
        timestamp = %timestamp,
        "recorded config processing completed marker"
    );

    Ok(())
}

/// Returns whether config processing has been explicitly marked as completed.
pub async fn is_config_processing_completed(pool: &Arc<Pool>, prefix: &str) -> Result<bool> {
    use redis::AsyncCommands;

    let mut conn = pool.get().await?;
    let hash_key = format!("{prefix}:{BOOTSTRAP_META_KEY}");

    let status: Option<String> = conn
        .hget(&hash_key, CONFIG_PROCESSING_STATUS_FIELD)
        .await
        .map_err(|e| eyre::eyre!("Failed to get config processing status: {}", e))?;

    Ok(matches!(
        status.as_deref(),
        Some(CONFIG_PROCESSING_STATUS_COMPLETED)
    ))
}

/// Returns whether config processing is explicitly marked as in progress.
pub async fn is_config_processing_in_progress(pool: &Arc<Pool>, prefix: &str) -> Result<bool> {
    use redis::AsyncCommands;

    let mut conn = pool.get().await?;
    let hash_key = format!("{prefix}:{BOOTSTRAP_META_KEY}");

    let status: Option<String> = conn
        .hget(&hash_key, CONFIG_PROCESSING_STATUS_FIELD)
        .await
        .map_err(|e| eyre::eyre!("Failed to get config processing status: {}", e))?;

    Ok(matches!(
        status.as_deref(),
        Some(CONFIG_PROCESSING_STATUS_IN_PROGRESS)
    ))
}

/// Sets the last sync timestamp for a relayer to the current time.
///
/// This should be called after a relayer has been successfully initialized
/// to record when the initialization occurred.
///
/// # Arguments
/// * `conn` - Redis connection manager
/// * `prefix` - Key prefix for multi-tenant support
/// * `relayer_id` - The relayer's unique identifier
///
/// # Redis Key Format
/// Hash key: `{prefix}:relayer_sync_meta`
/// Hash field: `{relayer_id}:last_sync`
/// Value: Unix timestamp in seconds
pub async fn set_relayer_last_sync(pool: &Arc<Pool>, prefix: &str, relayer_id: &str) -> Result<()> {
    use chrono::Utc;
    use redis::AsyncCommands;

    let mut conn = pool.get().await?;
    let hash_key = format!("{prefix}:{RELAYER_SYNC_META_KEY}");
    let field = format!("{relayer_id}:last_sync");
    let timestamp = Utc::now().timestamp();

    conn.hset::<_, _, _, ()>(&hash_key, &field, timestamp)
        .await
        .map_err(|e| eyre::eyre!("Failed to set relayer last sync: {}", e))?;

    debug!(
        relayer_id = %relayer_id,
        timestamp = %timestamp,
        "recorded relayer last sync time"
    );

    Ok(())
}

/// Gets the last sync timestamp for a relayer.
///
/// # Arguments
/// * `conn` - Redis connection manager
/// * `prefix` - Key prefix for multi-tenant support
/// * `relayer_id` - The relayer's unique identifier
///
/// # Returns
/// * `Ok(Some(DateTime))` - The last sync time if recorded
/// * `Ok(None)` - If the relayer has never been synced
/// * `Err(_)` - Redis communication error
pub async fn get_relayer_last_sync(
    pool: &Arc<Pool>,
    prefix: &str,
    relayer_id: &str,
) -> Result<Option<chrono::DateTime<chrono::Utc>>> {
    use chrono::{TimeZone, Utc};
    use redis::AsyncCommands;

    let mut conn = pool.get().await?;
    let hash_key = format!("{prefix}:{RELAYER_SYNC_META_KEY}");
    let field = format!("{relayer_id}:last_sync");

    let timestamp: Option<i64> = conn
        .hget(&hash_key, &field)
        .await
        .map_err(|e| eyre::eyre!("Failed to get relayer last sync: {}", e))?;

    Ok(timestamp.and_then(|ts| Utc.timestamp_opt(ts, 0).single()))
}

/// Checks if a relayer was recently synced within the given threshold.
///
/// This is used to skip initialization for relayers that were recently
/// initialized by another instance (e.g., during rolling restarts).
///
/// # Arguments
/// * `conn` - Redis connection manager
/// * `prefix` - Key prefix for multi-tenant support
/// * `relayer_id` - The relayer's unique identifier
/// * `threshold_secs` - Number of seconds to consider as "recent"
///
/// # Returns
/// * `Ok(true)` - If the relayer was synced within the threshold
/// * `Ok(false)` - If the relayer was not synced or sync is stale
/// * `Err(_)` - Redis communication error
pub async fn is_relayer_recently_synced(
    pool: &Arc<Pool>,
    prefix: &str,
    relayer_id: &str,
    threshold_secs: u64,
) -> Result<bool> {
    use chrono::Utc;

    let last_sync = get_relayer_last_sync(pool, prefix, relayer_id).await?;

    match last_sync {
        Some(sync_time) => {
            let elapsed = Utc::now().signed_duration_since(sync_time);
            let is_recent = elapsed.num_seconds() < threshold_secs as i64;

            if is_recent {
                debug!(
                    relayer_id = %relayer_id,
                    last_sync = %sync_time.to_rfc3339(),
                    elapsed_secs = %elapsed.num_seconds(),
                    threshold_secs = %threshold_secs,
                    "relayer was recently synced"
                );
            }

            Ok(is_recent)
        }
        None => Ok(false),
    }
}

// ============================================================================
// Global Initialization Tracking Functions
// ============================================================================
//
// These functions track when the global relayer initialization was last completed.
// This allows multiple instances to skip redundant initialization when
// initialization was recently completed by another instance.

/// Sets the global initialization completion timestamp to the current time.
///
/// This should be called after all relayers have been successfully initialized
/// to record when the initialization occurred.
///
/// # Arguments
/// * `conn` - Redis connection manager
/// * `prefix` - Key prefix for multi-tenant support
///
/// # Redis Key Format
/// Hash key: `{prefix}:relayer_sync_meta`
/// Hash field: `global:init_completed`
/// Value: Unix timestamp in seconds
pub async fn set_global_init_completed(pool: &Arc<Pool>, prefix: &str) -> Result<()> {
    use chrono::Utc;
    use redis::AsyncCommands;

    let mut conn = pool.get().await?;
    let hash_key = format!("{prefix}:{RELAYER_SYNC_META_KEY}");
    let timestamp = Utc::now().timestamp();

    conn.hset::<_, _, _, ()>(&hash_key, GLOBAL_INIT_FIELD, timestamp)
        .await
        .map_err(|e| eyre::eyre!("Failed to set global init completed: {}", e))?;

    debug!(timestamp = %timestamp, "recorded global initialization completion time");
    Ok(())
}

/// Checks if global initialization was recently completed within the given threshold.
///
/// This is used to skip initialization when another instance recently completed
/// initialization (e.g., during rolling restarts).
///
/// # Arguments
/// * `conn` - Redis connection manager
/// * `prefix` - Key prefix for multi-tenant support
/// * `threshold_secs` - Number of seconds to consider as "recent"
///
/// # Returns
/// * `Ok(true)` - If initialization was completed within the threshold
/// * `Ok(false)` - If initialization was not completed or is stale
/// * `Err(_)` - Redis communication error
pub async fn is_global_init_recently_completed(
    pool: &Arc<Pool>,
    prefix: &str,
    threshold_secs: u64,
) -> Result<bool> {
    use chrono::Utc;
    use redis::AsyncCommands;

    let mut conn = pool.get().await?;
    let hash_key = format!("{prefix}:{RELAYER_SYNC_META_KEY}");

    let timestamp: Option<i64> = conn
        .hget(&hash_key, GLOBAL_INIT_FIELD)
        .await
        .map_err(|e| eyre::eyre!("Failed to get global init time: {}", e))?;

    match timestamp {
        Some(ts) => {
            let now = Utc::now().timestamp();
            let elapsed = now - ts;
            let is_recent = elapsed < threshold_secs as i64;

            if is_recent {
                debug!(
                    elapsed_secs = %elapsed,
                    threshold_secs = %threshold_secs,
                    "global initialization recently completed"
                );
            }

            Ok(is_recent)
        }
        None => Ok(false),
    }
}

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

    #[test]
    fn test_generate_lock_value_is_unique() {
        let value1 = generate_lock_value();
        let value2 = generate_lock_value();

        // Values should be different due to monotonic counter
        assert_ne!(value1, value2);
    }

    #[test]
    fn test_generate_lock_value_contains_process_id() {
        let value = generate_lock_value();

        // Should contain a colon separator
        assert!(value.contains(':'));

        // Should have two parts: process_id and timestamp
        let parts: Vec<&str> = value.split(':').collect();
        assert_eq!(parts.len(), 3);

        // First part should be parseable as u32 (process ID)
        assert!(parts[0].parse::<u32>().is_ok());
    }

    #[test]
    fn test_should_retain_connection_keeps_young_connection() {
        // A freshly created connection is younger than any positive max age.
        let metrics = Metrics::default();
        assert!(
            should_retain_connection(&metrics, Duration::from_secs(60)),
            "young connection should be retained"
        );
    }

    #[test]
    fn test_should_retain_connection_drops_over_age_connection() {
        // Simulate a connection created well in the past → over max age → dropped.
        // checked_sub: a bare `Instant - Duration` panics on hosts whose
        // monotonic clock started less than 120s ago (fresh VMs/containers).
        let Some(past) = std::time::Instant::now().checked_sub(Duration::from_secs(120)) else {
            return;
        };
        let mut metrics = Metrics::default();
        metrics.created = past;
        assert!(
            !should_retain_connection(&metrics, Duration::from_secs(60)),
            "over-age connection should be dropped"
        );
    }

    #[test]
    fn test_distributed_lock_key_format() {
        // Verify the lock key format: {prefix}:lock:{name}
        let prefix = "myrelayer";
        let lock_name = "transaction_cleanup";
        let lock_key = format!("{prefix}:lock:{lock_name}");
        assert_eq!(lock_key, "myrelayer:lock:transaction_cleanup");
    }

    #[test]
    fn test_distributed_lock_key_format_with_complex_prefix() {
        // Test with a more realistic prefix
        let prefix = "oz-relayer-prod";
        let lock_name = "transaction_cleanup";
        let lock_key = format!("{prefix}:lock:{lock_name}");
        assert_eq!(lock_key, "oz-relayer-prod:lock:transaction_cleanup");
    }

    #[test]
    fn test_distributed_lock_key_uses_exact_key() {
        // DistributedLock now uses the exact key provided (no automatic prefix)
        let lock_key = "myprefix:lock:myoperation";
        // The lock will use this key exactly as provided
        assert_eq!(lock_key, "myprefix:lock:myoperation");
    }

    // =========================================================================
    // RedisConnections tests
    // =========================================================================

    mod redis_connections_tests {
        use super::*;

        #[test]
        fn test_new_single_pool_returns_same_pool_for_both() {
            // This test verifies the backward-compatible single pool mode
            // When no REDIS_READER_URL is set, both pools should be the same
            let cfg = Config::from_url("redis://127.0.0.1:6379");
            let pool = cfg
                .builder()
                .expect("Failed to create pool builder")
                .max_size(16)
                .runtime(Runtime::Tokio1)
                .build()
                .expect("Failed to build pool");
            let pool = Arc::new(pool);

            let connections = RedisConnections::new_single_pool(pool.clone());

            // Both primary and reader should return the same pool
            assert!(Arc::ptr_eq(connections.primary(), &pool));
            assert!(Arc::ptr_eq(connections.reader(), &pool));
            assert!(Arc::ptr_eq(connections.primary(), connections.reader()));
        }

        #[test]
        fn test_redis_connections_clone() {
            // Verify that RedisConnections can be cloned
            let cfg = Config::from_url("redis://127.0.0.1:6379");
            let pool = cfg
                .builder()
                .expect("Failed to create pool builder")
                .max_size(16)
                .runtime(Runtime::Tokio1)
                .build()
                .expect("Failed to build pool");
            let pool = Arc::new(pool);

            let connections = RedisConnections::new_single_pool(pool);
            let cloned = connections.clone();

            // Both should point to the same pools
            assert!(Arc::ptr_eq(connections.primary(), cloned.primary()));
            assert!(Arc::ptr_eq(connections.reader(), cloned.reader()));
        }

        #[test]
        fn test_redis_connections_debug() {
            // Verify Debug implementation exists
            let cfg = Config::from_url("redis://127.0.0.1:6379");
            let pool = cfg
                .builder()
                .expect("Failed to create pool builder")
                .max_size(16)
                .runtime(Runtime::Tokio1)
                .build()
                .expect("Failed to build pool");
            let pool = Arc::new(pool);

            let connections = RedisConnections::new_single_pool(pool);
            let debug_str = format!("{connections:?}");

            assert!(debug_str.contains("RedisConnections"));
        }
    }

    // =========================================================================
    // Integration tests - require a running Redis instance
    // Run with: cargo test --lib redis::tests::integration -- --ignored
    // =========================================================================

    /// Helper to create a Redis connection for integration tests.
    /// Expects Redis to be running on localhost:6379.
    async fn create_test_redis_connection() -> Option<Arc<Pool>> {
        let cfg = Config::from_url("redis://127.0.0.1:6379");
        let pool = cfg
            .builder()
            .ok()?
            .max_size(16)
            .runtime(Runtime::Tokio1)
            .build()
            .ok()?;
        Some(Arc::new(pool))
    }

    mod integration {
        use super::*;

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_distributed_lock_acquire_and_release() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let lock =
                DistributedLock::new(conn, "test:lock:acquire_release", Duration::from_secs(60));

            // Should be able to acquire the lock
            let guard = lock.try_acquire().await.expect("Redis error");
            assert!(guard.is_some(), "Should acquire lock when not held");

            // Release the lock
            let released = guard.unwrap().release().await.expect("Redis error");
            assert!(released, "Should successfully release the lock");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_distributed_lock_prevents_double_acquisition() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let lock1 = DistributedLock::new(
                conn.clone(),
                "test:lock:double_acquire",
                Duration::from_secs(60),
            );
            let lock2 =
                DistributedLock::new(conn, "test:lock:double_acquire", Duration::from_secs(60));

            // First acquisition should succeed
            let guard1 = lock1.try_acquire().await.expect("Redis error");
            assert!(guard1.is_some(), "First acquisition should succeed");

            // Second acquisition should fail (lock already held)
            let guard2 = lock2.try_acquire().await.expect("Redis error");
            assert!(
                guard2.is_none(),
                "Second acquisition should fail - lock already held"
            );

            // Release the first lock
            guard1.unwrap().release().await.expect("Redis error");

            // Now second acquisition should succeed
            let guard2_retry = lock2.try_acquire().await.expect("Redis error");
            assert!(guard2_retry.is_some(), "Should acquire lock after release");

            // Cleanup
            guard2_retry.unwrap().release().await.expect("Redis error");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_distributed_lock_expires_after_ttl() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            // Use a very short TTL for testing
            let lock =
                DistributedLock::new(conn.clone(), "test:lock:ttl_expiry", Duration::from_secs(1));

            // Acquire the lock
            let guard = lock.try_acquire().await.expect("Redis error");
            assert!(guard.is_some(), "Should acquire lock");

            // Don't release - let it expire
            drop(guard);

            // Wait for TTL to expire
            tokio::time::sleep(Duration::from_secs(2)).await;

            // Should be able to acquire again after expiry
            let lock2 = DistributedLock::new(conn, "test:lock:ttl_expiry", Duration::from_secs(60));
            let guard2 = lock2.try_acquire().await.expect("Redis error");
            assert!(guard2.is_some(), "Should acquire lock after TTL expiry");

            // Cleanup
            if let Some(g) = guard2 {
                g.release().await.expect("Redis error");
            }
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_distributed_lock_release_only_own_lock() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            // Use a short TTL
            let lock = DistributedLock::new(
                conn.clone(),
                "test:lock:release_own",
                Duration::from_secs(1),
            );

            // Acquire the lock
            let guard = lock.try_acquire().await.expect("Redis error");
            assert!(guard.is_some(), "Should acquire lock");

            // Wait for lock to expire
            tokio::time::sleep(Duration::from_secs(2)).await;

            // Try to release expired lock - should return false (not owned anymore)
            let released = guard.unwrap().release().await.expect("Redis error");
            assert!(!released, "Should not release expired lock");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_distributed_lock_with_prefix() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            // Simulate prefixed lock key like in transaction cleanup
            // Format: {prefix}:lock:{name}
            let prefix = "test_prefix";
            let lock_name = "cleanup";
            let lock_key = format!("{prefix}:lock:{lock_name}");

            let lock = DistributedLock::new(conn, &lock_key, Duration::from_secs(60));

            let guard = lock.try_acquire().await.expect("Redis error");
            assert!(guard.is_some(), "Should acquire prefixed lock");

            // Cleanup
            guard.unwrap().release().await.expect("Redis error");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_distributed_lock_drop_releases_lock() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let lock_key = "test:lock:drop_release";

            // Acquire lock in inner scope
            {
                let lock = DistributedLock::new(conn.clone(), lock_key, Duration::from_secs(60));
                let guard = lock.try_acquire().await.expect("Redis error");
                assert!(guard.is_some(), "Should acquire lock");

                // Guard is dropped here without explicit release()
            }

            // Give the spawned release task time to complete
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Should be able to acquire again because Drop released the lock
            let lock2 = DistributedLock::new(conn, lock_key, Duration::from_secs(60));
            let guard2 = lock2.try_acquire().await.expect("Redis error");
            assert!(
                guard2.is_some(),
                "Should acquire lock after previous guard was dropped"
            );

            // Cleanup
            if let Some(g) = guard2 {
                g.release().await.expect("Redis error");
            }
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_distributed_lock_explicit_release_prevents_double_release() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let lock_key = "test:lock:no_double_release";
            let lock = DistributedLock::new(conn.clone(), lock_key, Duration::from_secs(60));

            let guard = lock.try_acquire().await.expect("Redis error");
            assert!(guard.is_some(), "Should acquire lock");

            // Explicitly release
            let released = guard.unwrap().release().await.expect("Redis error");
            assert!(released, "Should release successfully");

            // Guard is now consumed by release(), Drop won't run
            // Verify lock is still released (not double-locked)
            tokio::time::sleep(Duration::from_millis(50)).await;

            let lock2 = DistributedLock::new(conn, lock_key, Duration::from_secs(60));
            let guard2 = lock2.try_acquire().await.expect("Redis error");
            assert!(
                guard2.is_some(),
                "Should acquire lock - no double-release issue"
            );

            // Cleanup
            if let Some(g) = guard2 {
                g.release().await.expect("Redis error");
            }
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_distributed_lock_drop_on_early_return() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let lock_key = "test:lock:early_return";

            // Simulate a function that returns early (like ? operator)
            async fn simulated_work_with_early_return(
                conn: Arc<Pool>,
                lock_key: &str,
            ) -> Result<(), &'static str> {
                let lock = DistributedLock::new(conn, lock_key, Duration::from_secs(60));
                let _guard = lock
                    .try_acquire()
                    .await
                    .map_err(|_| "lock error")?
                    .ok_or("lock held")?;

                // Simulate early return (error path)
                Err("simulated error")

                // _guard is dropped here due to early return
            }

            // Call the function that returns early
            let result = simulated_work_with_early_return(conn.clone(), lock_key).await;
            assert!(result.is_err(), "Should have returned early with error");

            // Give the spawned release task time to complete
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Lock should be released despite the early return
            let lock2 = DistributedLock::new(conn, lock_key, Duration::from_secs(60));
            let guard2 = lock2.try_acquire().await.expect("Redis error");
            assert!(
                guard2.is_some(),
                "Should acquire lock after early return released it"
            );

            // Cleanup
            if let Some(g) = guard2 {
                g.release().await.expect("Redis error");
            }
        }

        // =========================================================================
        // RedisConnections integration tests
        // =========================================================================

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_redis_connections_single_pool_operations() {
            let pool = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let connections = RedisConnections::new_single_pool(pool);

            // Test that we can get connections from both pools
            let mut primary_conn = connections
                .primary()
                .get()
                .await
                .expect("Failed to get primary connection");
            let mut reader_conn = connections
                .reader()
                .get()
                .await
                .expect("Failed to get reader connection");

            // Test basic operations on both connections
            let _: () = redis::cmd("SET")
                .arg("test:connections:key")
                .arg("test_value")
                .query_async(&mut primary_conn)
                .await
                .expect("Failed to SET");

            let value: String = redis::cmd("GET")
                .arg("test:connections:key")
                .query_async(&mut reader_conn)
                .await
                .expect("Failed to GET");

            assert_eq!(value, "test_value");

            // Cleanup
            let _: () = redis::cmd("DEL")
                .arg("test:connections:key")
                .query_async(&mut primary_conn)
                .await
                .expect("Failed to DEL");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_redis_connections_backward_compatible() {
            // Verify that single pool mode (no reader URL) works correctly
            let pool = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let connections = Arc::new(RedisConnections::new_single_pool(pool));

            // Multiple repositories should be able to share the connections
            let conn1 = connections.clone();
            let conn2 = connections.clone();

            // Both should be able to get connections
            let _primary1 = conn1.primary().get().await.expect("Failed to get primary");
            let _reader1 = conn1.reader().get().await.expect("Failed to get reader");
            let _primary2 = conn2.primary().get().await.expect("Failed to get primary");
            let _reader2 = conn2.reader().get().await.expect("Failed to get reader");

            // All should work without issues (backward compatible)
        }
        // =====================================================================
        // Relayer Sync Metadata Tests
        // =====================================================================

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_set_and_get_relayer_last_sync() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_sync";
            let relayer_id = "test-relayer-sync";

            // Set the last sync time
            set_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should set last sync time");

            // Get the last sync time
            let last_sync = get_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should get last sync time");

            assert!(last_sync.is_some(), "Should have a last sync time");

            // Verify the timestamp is recent (within last minute)
            let elapsed = chrono::Utc::now()
                .signed_duration_since(last_sync.unwrap())
                .num_seconds();
            assert!(elapsed < 60, "Last sync should be within last minute");

            // Cleanup: delete the hash
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_get_relayer_last_sync_returns_none_when_not_set() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_sync_none";
            let relayer_id = "nonexistent-relayer";

            // Get the last sync time for a relayer that hasn't been synced
            let last_sync = get_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should not error");

            assert!(
                last_sync.is_none(),
                "Should return None for unsynced relayer"
            );
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_is_relayer_recently_synced_returns_true_for_recent_sync() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_recent_sync";
            let relayer_id = "recent-relayer";

            // Set the last sync time
            set_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should set last sync time");

            // Check if recently synced (within 5 minutes)
            let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 300)
                .await
                .expect("Should check recent sync");

            assert!(is_recent, "Should be recently synced");

            // Cleanup
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_is_relayer_recently_synced_returns_false_when_not_set() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_not_synced";
            let relayer_id = "never-synced-relayer";

            // Check if recently synced for a relayer that hasn't been synced
            let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 300)
                .await
                .expect("Should check recent sync");

            assert!(!is_recent, "Should not be recently synced");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_is_relayer_recently_synced_returns_false_for_stale_sync() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_stale_sync";
            let relayer_id = "stale-relayer";

            // Manually set an old timestamp (10 minutes ago)
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let field = format!("{relayer_id}:last_sync");
            let old_timestamp = chrono::Utc::now().timestamp() - 600; // 10 minutes ago

            redis::AsyncCommands::hset::<_, _, _, ()>(
                &mut conn_clone,
                &hash_key,
                &field,
                old_timestamp,
            )
            .await
            .expect("Should set old timestamp");

            // Check if recently synced with 5 minute threshold
            let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 300)
                .await
                .expect("Should check recent sync");

            assert!(!is_recent, "Should not be recently synced (stale)");

            // Cleanup
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_get_relayer_last_sync_multiple_relayers() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_multi_sync";

            // Set sync times for multiple relayers
            set_relayer_last_sync(&conn, prefix, "relayer-1")
                .await
                .expect("Should set sync time");
            tokio::time::sleep(Duration::from_millis(10)).await;
            set_relayer_last_sync(&conn, prefix, "relayer-2")
                .await
                .expect("Should set sync time");

            let sync1 = get_relayer_last_sync(&conn, prefix, "relayer-1")
                .await
                .expect("Should get sync time");
            let sync2 = get_relayer_last_sync(&conn, prefix, "relayer-2")
                .await
                .expect("Should get sync time");

            assert!(sync1.is_some(), "Relayer 1 should have sync time");
            assert!(sync2.is_some(), "Relayer 2 should have sync time");
            assert!(
                sync2.unwrap() >= sync1.unwrap(),
                "Relayer 2 should be synced at same time or after relayer 1"
            );

            // Cleanup
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_get_relayer_last_sync_update_existing() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_update_sync";
            let relayer_id = "update-relayer";

            // Set initial sync time
            set_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should set sync time");

            let first_sync = get_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should get sync time")
                .expect("Should have sync time");

            // Wait and update
            tokio::time::sleep(Duration::from_millis(100)).await;

            set_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should update sync time");

            let second_sync = get_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should get sync time")
                .expect("Should have sync time");

            assert!(
                second_sync >= first_sync,
                "Updated sync time should be at the same time or later than first"
            );

            // Cleanup
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_is_relayer_recently_synced_threshold_boundary() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_boundary_sync";
            let relayer_id = "boundary-relayer";

            // Set timestamp exactly at threshold (should be considered NOT recent)
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let field = format!("{relayer_id}:last_sync");
            let threshold_secs: u64 = 60;
            let boundary_timestamp = chrono::Utc::now().timestamp() - (threshold_secs as i64);

            redis::AsyncCommands::hset::<_, _, _, ()>(
                &mut conn_clone,
                &hash_key,
                &field,
                boundary_timestamp,
            )
            .await
            .expect("Should set boundary timestamp");

            let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, threshold_secs)
                .await
                .expect("Should check recent sync");

            assert!(!is_recent, "Should not be recent at exact threshold");

            // Cleanup
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_is_relayer_recently_synced_just_before_threshold() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_just_before_sync";
            let relayer_id = "just-before-relayer";

            // Set timestamp just before threshold (should be considered recent)
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let field = format!("{relayer_id}:last_sync");
            let threshold_secs: u64 = 60;
            let just_before_timestamp =
                chrono::Utc::now().timestamp() - (threshold_secs as i64) + 5;

            redis::AsyncCommands::hset::<_, _, _, ()>(
                &mut conn_clone,
                &hash_key,
                &field,
                just_before_timestamp,
            )
            .await
            .expect("Should set just-before timestamp");

            let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, threshold_secs)
                .await
                .expect("Should check recent sync");

            assert!(is_recent, "Should be recent just before threshold");

            // Cleanup
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_is_relayer_recently_synced_different_prefixes() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let relayer_id = "shared-relayer";

            // Set sync for prefix1 only
            set_relayer_last_sync(&conn, "prefix1", relayer_id)
                .await
                .expect("Should set sync time");

            let is_recent_prefix1 = is_relayer_recently_synced(&conn, "prefix1", relayer_id, 300)
                .await
                .expect("Should check sync");
            let is_recent_prefix2 = is_relayer_recently_synced(&conn, "prefix2", relayer_id, 300)
                .await
                .expect("Should check sync");

            assert!(is_recent_prefix1, "Should be recent for prefix1");
            assert!(!is_recent_prefix2, "Should not be recent for prefix2");

            // Cleanup
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = "prefix1:relayer_sync_meta";
            let _: () = redis::AsyncCommands::del(&mut conn_clone, hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_is_relayer_recently_synced_zero_threshold() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_zero_threshold";
            let relayer_id = "zero-threshold-relayer";

            set_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should set sync time");

            // With zero threshold, even immediate sync should be considered stale
            let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 0)
                .await
                .expect("Should check sync");

            assert!(!is_recent, "Should not be recent with zero threshold");

            // Cleanup
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        // =====================================================================
        // Global Initialization Tracking Tests
        // =====================================================================

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_set_and_check_global_init_completed() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_global_init";

            // Set global init completed
            set_global_init_completed(&conn, prefix)
                .await
                .expect("Should set global init completed");

            // Check if recently completed (within 5 minutes)
            let is_recent = is_global_init_recently_completed(&conn, prefix, 300)
                .await
                .expect("Should check global init");

            assert!(is_recent, "Should be recently completed");

            // Cleanup
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_is_global_init_recently_completed_returns_false_when_not_set() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_global_init_not_set";

            // Check without setting - should return false
            let is_recent = is_global_init_recently_completed(&conn, prefix, 300)
                .await
                .expect("Should check global init");

            assert!(!is_recent, "Should not be recently completed when not set");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_is_global_init_recently_completed_returns_false_when_stale() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_global_init_stale";

            // Manually set an old timestamp (10 minutes ago)
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let old_timestamp = chrono::Utc::now().timestamp() - 600; // 10 minutes ago

            redis::AsyncCommands::hset::<_, _, _, ()>(
                &mut conn_clone,
                &hash_key,
                "global:init_completed",
                old_timestamp,
            )
            .await
            .expect("Should set old timestamp");

            // Check with 5 minute threshold - should be stale
            let is_recent = is_global_init_recently_completed(&conn, prefix, 300)
                .await
                .expect("Should check global init");

            assert!(!is_recent, "Should not be recent when stale");

            // Cleanup
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_global_init_different_prefixes() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            // Set global init for prefix1 only
            set_global_init_completed(&conn, "global_prefix1")
                .await
                .expect("Should set global init");

            let is_recent_prefix1 = is_global_init_recently_completed(&conn, "global_prefix1", 300)
                .await
                .expect("Should check global init");
            let is_recent_prefix2 = is_global_init_recently_completed(&conn, "global_prefix2", 300)
                .await
                .expect("Should check global init");

            assert!(is_recent_prefix1, "Should be recent for prefix1");
            assert!(!is_recent_prefix2, "Should not be recent for prefix2");

            // Cleanup
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = "global_prefix1:relayer_sync_meta";
            let _: () = redis::AsyncCommands::del(&mut conn_clone, hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_global_init_update_existing() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_global_init_update";

            // Set initial timestamp
            set_global_init_completed(&conn, prefix)
                .await
                .expect("Should set global init");

            // Wait a bit
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Update timestamp
            set_global_init_completed(&conn, prefix)
                .await
                .expect("Should update global init");

            // Should still be recent
            let is_recent = is_global_init_recently_completed(&conn, prefix, 300)
                .await
                .expect("Should check global init");

            assert!(is_recent, "Should be recent after update");

            // Cleanup
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_global_init_coexists_with_relayer_sync() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_coexist";
            let relayer_id = "test-relayer";

            // Set both global init and relayer sync
            set_global_init_completed(&conn, prefix)
                .await
                .expect("Should set global init");
            set_relayer_last_sync(&conn, prefix, relayer_id)
                .await
                .expect("Should set relayer sync");

            // Both should be checkable
            let global_recent = is_global_init_recently_completed(&conn, prefix, 300)
                .await
                .expect("Should check global init");
            let relayer_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 300)
                .await
                .expect("Should check relayer sync");

            assert!(global_recent, "Global init should be recent");
            assert!(relayer_recent, "Relayer sync should be recent");

            // Cleanup
            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:relayer_sync_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        // =====================================================================
        // Config Processing Marker Tests
        // =====================================================================

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_config_processing_in_progress_is_not_completed() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_config_processing_in_progress";

            set_config_processing_in_progress(&conn, prefix)
                .await
                .expect("Should set in-progress marker");

            let completed = is_config_processing_completed(&conn, prefix)
                .await
                .expect("Should read config processing status");

            assert!(
                !completed,
                "In-progress config processing must not be treated as completed"
            );

            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:bootstrap_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }

        #[tokio::test]
        #[ignore] // Requires running Redis instance
        async fn test_config_processing_completed_requires_explicit_completion_marker() {
            let conn = create_test_redis_connection()
                .await
                .expect("Redis connection required for this test");

            let prefix = "test_config_processing_completed";

            set_config_processing_in_progress(&conn, prefix)
                .await
                .expect("Should set in-progress marker");
            set_config_processing_completed(&conn, prefix)
                .await
                .expect("Should set completed marker");

            let completed = is_config_processing_completed(&conn, prefix)
                .await
                .expect("Should read config processing status");

            assert!(
                completed,
                "Completed config processing must require the explicit completion marker"
            );

            let mut conn_clone = conn.get().await.expect("Failed to get connection");
            let hash_key = format!("{prefix}:bootstrap_meta");
            let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key)
                .await
                .expect("Cleanup failed");
        }
    }
}