//! Shared integration tests for the redis_interface crate.
//!
//! These tests exercise the **public API** only — `RedisConnectionPool`,
//! `RedisKey`, `RedisEntryId`, `RedisValue`, etc. — and therefore
//! validate both the `fred` and `redis-rs` module with the same
//! test suite.
//!
//! Run with redis-rs (default):
//!     cargo test -p redis_interface
//!
//! Run with fred:
//!     cargo test -p redis_interface --no-default-features --features fred

use std::collections::HashMap;

use crate::{
    ConsumerGroupDestroyReply, DelReply, HsetnxReply, MsetnxReply, RedisConnectionPool,
    RedisConnectionWithContext, RedisEntryId, RedisKey, RedisSettings, RedisValue, SaddReply,
    SetGetReply, SetnxReply, StreamCapKind, StreamCapTrim, StreamTrimConfig,
};

// ─── Helpers ──────────────────────────────────────────────────────────────────

async fn test_connection(
    settings: &RedisSettings,
) -> error_stack::Result<RedisConnectionWithContext, crate::errors::RedisError> {
    let pool = RedisConnectionPool::new_without_event_emitter(settings)
        .await
        .map(std::sync::Arc::new)?;

    Ok(RedisConnectionWithContext::new_without_context(pool))
}

/// Generate a unique ID for test key isolation.
/// Uses PID + nanoseconds + counter for global uniqueness across parallel runs.
fn unique_test_id() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let counter = COUNTER.fetch_add(1, Ordering::SeqCst);
    let pid = std::process::id();
    let millis = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis())
        .unwrap_or(0);
    format!("{pid}_{millis}_{counter}")
}

/// Convert a `RedisValue` to `Option<String>`.
/// Works for both module because `RedisValue::as_string()` is a shared method.
fn redis_value_to_option_string(v: &RedisValue) -> Option<String> {
    v.as_string()
}

/// Create a cluster `RedisConnectionPool` if `REDIS_CLUSTER_URLS` env var is set.
/// Returns `None` if no cluster is configured (test should skip, not fail).
async fn cluster_pool() -> Option<RedisConnectionWithContext> {
    let cluster_urls_str = std::env::var("REDIS_CLUSTER_URLS").ok()?;
    let cluster_urls: Vec<String> = cluster_urls_str
        .split(',')
        .map(|url| url.trim().to_string())
        .filter(|url| !url.is_empty())
        .collect();

    if cluster_urls.is_empty() {
        return None;
    }

    let first_url = cluster_urls.first()?;
    let (host, port) = if first_url.starts_with("redis://") {
        let without_scheme = first_url.trim_start_matches("redis://");
        let parts: Vec<&str> = without_scheme.split(':').collect();
        (
            parts.first()?.to_string(),
            parts.get(1)?.parse::<u16>().ok()?,
        )
    } else {
        let parts: Vec<&str> = first_url.split(':').collect();
        (
            parts.first()?.to_string(),
            parts.get(1)?.parse::<u16>().ok()?,
        )
    };

    let settings = RedisSettings {
        host,
        port,
        cluster_enabled: true,
        cluster_urls,
        ..RedisSettings::default()
    };

    test_connection(&settings).await.ok()
}

/// Helper: get a cluster pool or skip the test.
async fn get_cluster_pool_or_skip() -> Option<RedisConnectionWithContext> {
    let pool = cluster_pool().await;
    if pool.is_none() {
        tracing::warn!(
            "SKIP: Cluster test skipped — set REDIS_CLUSTER_URLS to enable. \
             Example: REDIS_CLUSTER_URLS=redis://localhost:7000,redis://localhost:7001"
        );
    }
    pool
}

async fn get_cluster_pool_with_uid() -> Option<(RedisConnectionWithContext, String)> {
    let pool = get_cluster_pool_or_skip().await?;
    let unique_id = unique_test_id();
    Some((pool, unique_id))
}

// ─── Consumer Group Validation ────────────────────────────────────────────────

#[tokio::test]
async fn test_consumer_group_create() {
    let is_invalid_redis_entry_error = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let redis_conn = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");

            let result1 = redis_conn
                .consumer_group_create(&"TEST1".into(), "GTEST", &RedisEntryId::AutoGeneratedID)
                .await;

            let result2 = redis_conn
                .consumer_group_create(&"TEST3".into(), "GTEST", &RedisEntryId::UndeliveredEntryID)
                .await;

            *result1.unwrap_err().current_context()
                == crate::errors::RedisError::InvalidRedisEntryId
                && *result2.unwrap_err().current_context()
                    == crate::errors::RedisError::InvalidRedisEntryId
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_invalid_redis_entry_error);
}

// ─── Key Commands ─────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_delete_existing_key_success() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let _ = pool.set_key(&"key".into(), "value".to_string()).await;
            pool.delete_key(&"key".into()).await.is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_delete_non_existing_key_success() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            pool.delete_key(&"key not exists".into()).await.is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_delete_key_returns_correct_reply() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let existing_key: RedisKey = format!("test_del_exist_{uid}").into();
            let non_existing_key: RedisKey = format!("test_del_miss_{uid}").into();

            let _ = pool.set_key(&existing_key, "value".to_string()).await;

            let delete_existing = pool.delete_key(&existing_key).await;
            let delete_non_existing = pool.delete_key(&non_existing_key).await;

            matches!(delete_existing, Ok(DelReply::KeyDeleted))
                && matches!(delete_non_existing, Ok(DelReply::KeyNotDeleted))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_delete_key_reply_semantics() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let existing_key: RedisKey = format!("test_del_existing_{uid}").into();
            let non_existing_key: RedisKey = format!("test_del_nonexisting_{uid}").into();

            let _ = pool.set_key(&existing_key, "value".to_string()).await;

            let delete_existing = pool.delete_key(&existing_key).await;
            let delete_non_existing = pool.delete_key(&non_existing_key).await;

            matches!(delete_existing, Ok(DelReply::KeyDeleted))
                && matches!(delete_non_existing, Ok(DelReply::KeyNotDeleted))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_delete_multiple_keys() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key1: RedisKey = format!("test_del_multi_a_{uid}").into();
            let key2: RedisKey = format!("test_del_multi_b_{uid}").into();
            let key3: RedisKey = format!("test_del_multi_c_{uid}").into();

            let _ = pool.set_key(&key1, "val1".to_string()).await;
            let _ = pool.set_key(&key2, "val2".to_string()).await;

            let result = pool.delete_multiple_keys(&[key1, key2, key3]).await;

            match result {
                Ok(replies) => {
                    replies.len() == 3
                        && replies.first() == Some(&DelReply::KeyDeleted)
                        && replies.get(1) == Some(&DelReply::KeyDeleted)
                        && replies.get(2) == Some(&DelReply::KeyNotDeleted)
                }
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_exists_key() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let existing_key: RedisKey = format!("test_exists_yes_{uid}").into();
            let missing_key: RedisKey = format!("test_exists_no_{uid}").into();

            let _ = pool.set_key(&existing_key, "val".to_string()).await;

            let exists_result: Result<bool, _> = pool.exists::<()>(&existing_key).await;
            let missing_result: Result<bool, _> = pool.exists::<()>(&missing_key).await;

            matches!(exists_result, Ok(true)) && matches!(missing_result, Ok(false))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_key_with_expiry() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = format!("test_set_expiry_{}", unique_test_id()).into();
            pool.set_key_with_expiry(&key, "test_value".to_string(), 60)
                .await
                .is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_key_without_modifying_ttl() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = format!("test_keepttl_{}", unique_test_id()).into();
            let initial_value = "initial".to_string();
            let new_value = "new_value".to_string();

            let _ = pool.set_key_with_expiry(&key, initial_value, 60).await;

            pool.set_key_without_modifying_ttl(&key, new_value.clone())
                .await
                .is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_key_if_not_exists_with_expiry_new_key() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = format!("test_setnx_new_{}", unique_test_id()).into();
            let value = "test_value".to_string();

            let result = pool
                .set_key_if_not_exists_with_expiry(&key, value.clone(), Some(60))
                .await;

            matches!(result, Ok(SetnxReply::KeySet))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_key_if_not_exists_with_expiry_existing_key() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = format!("test_setnx_exist_{}", unique_test_id()).into();
            let initial_value = "initial".to_string();

            let _ = pool.set_key(&key, initial_value.clone()).await;

            let result = pool
                .set_key_if_not_exists_with_expiry(&key, "new_value".to_string(), Some(60))
                .await;

            matches!(result, Ok(SetnxReply::KeyNotSet))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_key_if_not_exists_and_get_value_new_key() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = format!("test_new_key_string_{}", unique_test_id()).into();
            let value = "test_value".to_string();

            let result = pool
                .set_key_if_not_exists_and_get_value(&key, value.clone(), Some(30))
                .await;

            match result {
                Ok(SetGetReply::ValueSet(returned_value)) => returned_value == value,
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_key_if_not_exists_and_get_value_existing_key() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = format!("test_existing_key_string_{}", unique_test_id()).into();
            let initial_value = "initial_value".to_string();
            let new_value = "new_value".to_string();

            let _ = pool.set_key(&key, initial_value.clone()).await;

            let result = pool
                .set_key_if_not_exists_and_get_value(&key, new_value, Some(30))
                .await;

            match result {
                Ok(SetGetReply::ValueExists(returned_value)) => returned_value == initial_value,
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_key_if_not_exists_and_get_value_with_default_ttl() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = format!("test_default_ttl_key_string_{}", unique_test_id()).into();
            let value = "test_value".to_string();

            let result = pool
                .set_key_if_not_exists_and_get_value(&key, value.clone(), None)
                .await;

            match result {
                Ok(SetGetReply::ValueSet(returned_value)) => returned_value == value,
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_key_if_not_exists_and_get_value_concurrent_access() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key_name = &format!("test_concurrent_key_string_{}", unique_test_id());
            let value1 = "value1".to_string();
            let value2 = "value2".to_string();

            let conn1 = pool.clone();
            let conn2 = pool.clone();
            let key1 = key_name.into();
            let key2 = key_name.into();

            let (result1, result2) = tokio::join!(
                conn1.set_key_if_not_exists_and_get_value(&key1, value1, Some(30)),
                conn2.set_key_if_not_exists_and_get_value(&key2, value2, Some(30))
            );

            let result1_is_set = matches!(result1, Ok(SetGetReply::ValueSet(_)));
            let result2_is_set = matches!(result2, Ok(SetGetReply::ValueSet(_)));
            let result1_is_exists = matches!(result1, Ok(SetGetReply::ValueExists(_)));
            let result2_is_exists = matches!(result2, Ok(SetGetReply::ValueExists(_)));

            (result1_is_set && result2_is_exists) || (result1_is_exists && result2_is_set)
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_resp3_set_and_get() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = format!("test_resp3_{}", unique_test_id()).into();
            let value = "resp3_value".to_string();
            let _ = pool.set_key(&key, value.clone()).await;
            let result: Result<String, _> = pool.get_key(&key).await;
            result.is_ok() && result.unwrap() == value
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_connection_with_custom_config() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let settings = RedisSettings {
                reconnect_max_attempts: 10,
                reconnect_delay: 100,
                default_command_timeout: 60,
                max_in_flight_commands: 10000,
                ..RedisSettings::default()
            };

            let pool = test_connection(&settings)
                .await
                .expect("failed to create redis connection pool with custom config");

            let key: RedisKey = format!("test_config_{}", unique_test_id()).into();
            let value = "custom_config_value".to_string();
            let _ = pool.set_key(&key, value.clone()).await;
            let result: Result<String, _> = pool.get_key(&key).await;
            result.is_ok() && result.unwrap() == value
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── Get / Multiple Keys ──────────────────────────────────────────────────────

#[tokio::test]
async fn test_get_multiple_keys_success() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");

            let keys = vec![
                "multi_test_key1".into(),
                "multi_test_key2".into(),
                "multi_test_key3".into(),
            ];
            let values = ["value1", "value2", "value3"];

            for (key, value) in keys.iter().zip(values.iter()) {
                let _ = pool.set_key(key, value.to_string()).await;
            }

            let result = pool.get_multiple_keys::<String>(&keys).await;

            match result {
                Ok(retrieved_values) => {
                    retrieved_values.len() == 3
                        && retrieved_values.first() == Some(&Some("value1".to_string()))
                        && retrieved_values.get(1) == Some(&Some("value2".to_string()))
                        && retrieved_values.get(2) == Some(&Some("value3".to_string()))
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_get_multiple_keys_with_missing_keys() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");

            let keys = vec![
                "existing_key".into(),
                "non_existing_key".into(),
                "another_existing_key".into(),
            ];

            let _ = pool
                .set_key(
                    keys.first().expect("should not be none"),
                    "value1".to_string(),
                )
                .await;
            let _ = pool
                .set_key(
                    keys.get(2).expect("should not be none"),
                    "value3".to_string(),
                )
                .await;

            let result = pool.get_multiple_keys::<String>(&keys).await;

            match result {
                Ok(retrieved_values) => {
                    retrieved_values.len() == 3
                        && *retrieved_values.first().expect("should not be none")
                            == Some("value1".to_string())
                        && retrieved_values.get(1).is_some_and(|v| v.is_none())
                        && *retrieved_values.get(2).expect("should not be none")
                            == Some("value3".to_string())
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_get_multiple_keys_empty_input() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");

            let keys: Vec<RedisKey> = vec![];
            let result = pool.get_multiple_keys::<String>(&keys).await;

            match result {
                Ok(retrieved_values) => retrieved_values.is_empty(),
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_get_and_deserialize_key() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_deser_single_{uid}").into();

            #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug)]
            struct Data {
                x: i32,
            }

            let data = Data { x: 42 };
            let _ = pool.serialize_and_set_key(&key, &data).await;
            let result = pool.get_and_deserialize_key::<Data>(&key, "Data").await;

            match result {
                Ok(retrieved) => retrieved == data,
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_get_and_deserialize_key_not_found() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = "nonexistent_deser_key".into();

            let result: Result<String, _> = pool.get_and_deserialize_key(&key, "String").await;
            result.is_err()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_get_and_deserialize_multiple_keys() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");

            #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug, Clone)]
            struct TestData {
                id: u32,
                name: String,
            }

            let test_data = [
                TestData {
                    id: 1,
                    name: "test1".to_string(),
                },
                TestData {
                    id: 2,
                    name: "test2".to_string(),
                },
            ];

            let keys = vec![
                "serialize_test_key1".into(),
                "serialize_test_key2".into(),
                "non_existing_serialize_key".into(),
            ];

            for (i, data) in test_data.iter().enumerate() {
                let _ = pool
                    .serialize_and_set_key(keys.get(i).expect("should not be none"), data)
                    .await;
            }

            let result = pool
                .get_and_deserialize_multiple_keys::<TestData>(&keys, "TestData")
                .await;

            match result {
                Ok(retrieved_data) => {
                    retrieved_data.len() == 3
                        && retrieved_data.first() == Some(&Some(test_data[0].clone()))
                        && retrieved_data.get(1) == Some(&Some(test_data[1].clone()))
                        && retrieved_data.get(2) == Some(&None)
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── Serialization ────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_serialize_and_set_key_if_not_exist() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_ser_setnx_{uid}").into();

            #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug)]
            struct Data {
                id: u32,
            }

            let data = Data { id: 99 };

            let set_result = pool
                .serialize_and_set_key_if_not_exist(&key, &data, Some(60))
                .await;

            let dup_result = pool
                .serialize_and_set_key_if_not_exist(&key, &Data { id: 100 }, Some(60))
                .await;

            matches!(
                (set_result, dup_result),
                (Ok(SetnxReply::KeySet), Ok(SetnxReply::KeyNotSet),)
            )
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_serialize_and_set_key_with_expiry() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_ser_setex_{uid}").into();

            #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug)]
            struct Data {
                name: String,
            }

            let data = Data {
                name: "test".to_string(),
            };

            let set_result = pool
                .serialize_and_set_key_with_expiry(&key, &data, 120)
                .await;

            let get_result = pool.get_and_deserialize_key::<Data>(&key, "Data").await;

            match (set_result, get_result) {
                (Ok(()), Ok(retrieved)) => retrieved == data,
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_serialize_and_set_key_without_modifying_ttl() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_ser_keepttl_{uid}").into();

            #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug)]
            struct Data {
                val: i32,
            }

            let _ = pool
                .serialize_and_set_key_with_expiry(&key, &Data { val: 1 }, 120)
                .await;

            let update_result = pool
                .serialize_and_set_key_without_modifying_ttl(&key, &Data { val: 2 })
                .await;

            let get_result = pool.get_and_deserialize_key::<Data>(&key, "Data").await;

            match (update_result, get_result) {
                (Ok(()), Ok(retrieved)) => retrieved.val == 2,
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── TTL ──────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_set_expiry_and_get_ttl() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_ttl_{uid}").into();

            let _ = pool.set_key(&key, "value".to_string()).await;

            let set_expiry_result = pool.set_expiry(&key, 120).await;
            let ttl_result = pool.get_ttl(&key).await;

            match (set_expiry_result, ttl_result) {
                (Ok(()), Ok(ttl)) => ttl > 0 && ttl <= 120,
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_expire_at_and_get_ttl() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_expireat_{uid}").into();

            let _ = pool.set_key(&key, "value".to_string()).await;

            let future_ts = i64::try_from(
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_secs(),
            )
            .unwrap()
                + 120;
            let set_result = pool.set_expire_at(&key, future_ts).await;
            let ttl_result = pool.get_ttl(&key).await;

            match (set_result, ttl_result) {
                (Ok(()), Ok(ttl)) => ttl > 0 && ttl <= 120,
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── Hash Commands ────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_set_hash_fields() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let key: RedisKey = format!("test_hash_{}", unique_test_id()).into();
            let fields: Vec<(&str, &str)> = vec![
                ("field1", "value1"),
                ("field2", "value2"),
                ("field3", "value3"),
            ];

            pool.set_hash_fields(&key, fields, Some(60)).await.is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_hash_field_if_not_exist_and_get() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_hsetnx_{uid}").into();

            let set_result = pool
                .set_hash_field_if_not_exist(&key, "field1", "value1", None)
                .await;

            let dup_result = pool
                .set_hash_field_if_not_exist(&key, "field1", "value2", None)
                .await;

            let get_result: Result<String, _> = pool.get_hash_field(&key, "field1").await;

            let all_fields: Result<HashMap<String, String>, _> = pool.get_hash_fields(&key).await;

            match (set_result, dup_result, get_result, all_fields) {
                (Ok(HsetnxReply::KeySet), Ok(HsetnxReply::KeyNotSet), Ok(val), Ok(map)) => {
                    val == "value1" && map.get("field1") == Some(&"value1".to_string())
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_increment_fields_in_hash() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_hincr_{uid}").into();

            let fields: Vec<(&str, &str)> = vec![("counter", "10")];
            let _ = pool.set_hash_fields(&key, fields, Some(60)).await;

            let result = pool
                .increment_fields_in_hash(&key, &[("counter".to_string(), 5)])
                .await;

            match result {
                Ok(values) => values.len() == 1 && values.first() == Some(&15),
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_hscan_returns_values() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_hscan_{uid}").into();

            let fields: Vec<(&str, &str)> = vec![
                ("prefix_field1", "val1"),
                ("prefix_field2", "val2"),
                ("other_field", "val3"),
            ];
            let _ = pool.set_hash_fields(&key, fields, Some(60)).await;

            let scan_result = pool.hscan(&key, "prefix_*", None).await;

            match scan_result {
                Ok(values) => {
                    values.contains(&"val1".to_string())
                        && values.contains(&"val2".to_string())
                        && !values.contains(&"val3".to_string())
                }
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_hscan_and_deserialize() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_hscan_deser_{uid}").into();

            let fields: Vec<(&str, &str)> = vec![
                ("item_a", "\"alpha\""),
                ("item_b", "\"beta\""),
                ("item_c", "\"gamma\""),
            ];
            let _ = pool.set_hash_fields(&key, fields, Some(60)).await;

            let result = pool
                .hscan_and_deserialize::<String>(&key, "item_*", None)
                .await;

            match result {
                Ok(values) => {
                    values.contains(&"alpha".to_string())
                        && values.contains(&"beta".to_string())
                        && values.contains(&"gamma".to_string())
                }
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_sadd_and_get_hash_field_and_deserialize() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let set_key: RedisKey = format!("test_sadd_{uid}").into();
            let hash_key: RedisKey = format!("test_hget_deser_{uid}").into();

            // SADD
            let sadd_result = pool.sadd(&set_key, "member1").await;
            let sadd_dup = pool.sadd(&set_key, "member1").await;

            // get_hash_field_and_deserialize
            #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug)]
            struct TestData {
                id: u32,
            }
            let data = TestData { id: 42 };
            let _ = pool
                .serialize_and_set_hash_field_if_not_exist(&hash_key, "data", &data, None)
                .await;

            let deser_result = pool
                .get_hash_field_and_deserialize::<TestData>(&hash_key, "data", "TestData")
                .await;

            match (sadd_result, sadd_dup, deser_result) {
                (Ok(SaddReply::KeySet(1)), Ok(SaddReply::KeyNotSet), Ok(deserialized)) => {
                    deserialized == data
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_serialize_and_set_multiple_hash_field_if_not_exist() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key1: RedisKey = format!("test_ser_hsetnx_multi_a_{uid}").into();
            let key2: RedisKey = format!("test_ser_hsetnx_multi_b_{uid}").into();

            #[derive(serde::Serialize, Debug)]
            struct Data {
                id: u32,
            }

            let result = pool
                .serialize_and_set_multiple_hash_field_if_not_exist(
                    &[(&key1, Data { id: 1 }), (&key2, Data { id: 2 })],
                    "data",
                    None,
                )
                .await;

            match result {
                Ok(replies) => {
                    replies.len() == 2
                        && replies.first() == Some(&HsetnxReply::KeySet)
                        && replies.get(1) == Some(&HsetnxReply::KeySet)
                }
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── MSETNX ───────────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_set_multiple_keys_if_not_exist_msetnx() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key1: RedisKey = format!("test_msetnx_a_{uid}").into();
            let key2: RedisKey = format!("test_msetnx_b_{uid}").into();

            let result = pool
                .set_multiple_keys_if_not_exist(&[
                    (key1.tenant_aware_key(&pool.redis_conn), "val1"),
                    (key2.tenant_aware_key(&pool.redis_conn), "val2"),
                ])
                .await;

            matches!(result, Ok(MsetnxReply::KeysSet))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_multiple_keys_if_not_exist_with_existing_key() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key1: RedisKey = format!("test_msetnx_exist_a_{uid}").into();
            let key2: RedisKey = format!("test_msetnx_exist_b_{uid}").into();

            let _ = pool.set_key(&key1, "existing".to_string()).await;

            let result = pool
                .set_multiple_keys_if_not_exist(&[
                    (key1.tenant_aware_key(&pool.redis_conn), "new1"),
                    (key2.tenant_aware_key(&pool.redis_conn), "new2"),
                ])
                .await;

            matches!(result, Ok(MsetnxReply::KeysNotSet))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_set_multiple_keys_if_not_exists_and_get_values() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key1: RedisKey = format!("test_setget_multi_a_{uid}").into();
            let key2: RedisKey = format!("test_setget_multi_b_{uid}").into();

            let result = pool
                .set_multiple_keys_if_not_exists_and_get_values(
                    &[(key1, "val1".to_string()), (key2, "val2".to_string())],
                    Some(30),
                )
                .await;

            match result {
                Ok(replies) => {
                    replies.len() == 2
                        && matches!(replies.first(), Some(SetGetReply::ValueSet(ref value)) if value == "val1")
                        && matches!(replies.get(1), Some(SetGetReply::ValueSet(ref value)) if value == "val2")
                }
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── Stream Commands ──────────────────────────────────────────────────────────

#[tokio::test]
async fn test_stream_append_and_get_length() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let stream: RedisKey = format!("test_stream_append_{}", unique_test_id()).into();
            let fields: Vec<(&str, &str)> = vec![("field1", "value1")];

            let append_result = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                .await;

            let length_result = pool.stream_get_length(&stream).await;

            append_result.is_ok() && length_result.is_ok() && length_result.unwrap() >= 1
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_stream_read_entries_xread() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let stream: RedisKey = format!("test_stream_read_entries_{uid}").into();

            let fields1: Vec<(&str, &str)> = vec![("field1", "value1")];
            let fields2: Vec<(&str, &str)> = vec![("field2", "value2")];
            let _ = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields1)
                .await;
            let _ = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields2)
                .await;

            let result = pool
                .stream_read_entries(
                    std::slice::from_ref(&stream),
                    vec!["0-0".to_string()],
                    Some(10),
                )
                .await;

            match result {
                Ok(grouped) => {
                    let stream_key = pool.add_prefix(&format!("test_stream_read_entries_{uid}"));
                    let entries = grouped.get(&stream_key);
                    entries.is_some() && entries.unwrap().len() == 2
                }
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_stream_read_entries() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let stream: RedisKey = format!("test_xread_{uid}").into();

            let fields: Vec<(&str, &str)> = vec![("f", "v")];
            let _ = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                .await;

            let result = pool
                .stream_read_entries(
                    std::slice::from_ref(&stream),
                    vec!["0-0".to_string()],
                    Some(10),
                )
                .await;

            match result {
                Ok(reply) => {
                    reply.len() == 1
                        && reply
                            .values()
                            .next()
                            .is_some_and(|entries| !entries.is_empty())
                }
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_stream_read_with_options_xreadgroup() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let stream: RedisKey = format!("test_xreadgroup_{uid}").into();
            let group = format!("test_grp_{uid}");
            let consumer = format!("test_consumer_{uid}");

            let fields: Vec<(&str, &str)> = vec![("task", "process_payment")];
            pool.stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                .await
                .expect("failed to append");

            pool.consumer_group_create(
                &stream,
                &group,
                &RedisEntryId::UserSpecifiedID {
                    milliseconds: "0".to_string(),
                    sequence_number: "0".to_string(),
                },
            )
            .await
            .expect("failed to create group");

            let result = pool
                .stream_read_with_options(
                    &[stream],
                    vec![RedisEntryId::UndeliveredEntryID.to_stream_id()],
                    Some(1),
                    None,
                    Some((&group, &consumer)),
                )
                .await;

            match result {
                Ok(reply) => {
                    reply.len() == 1
                        && reply
                            .values()
                            .next()
                            .is_some_and(|entries| entries.len() == 1)
                        && reply
                            .values()
                            .next()
                            .and_then(|entries| entries.first())
                            .is_some_and(|(_, fields)| fields.contains_key("task"))
                }
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_stream_acknowledge_entries() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let stream: RedisKey = format!("test_stream_ack_{}", unique_test_id()).into();
            let group = "test_ack_group";
            let fields: Vec<(&str, &str)> = vec![("field1", "value1")];

            let _ = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                .await;
            let _ = pool
                .consumer_group_create(
                    &stream,
                    group,
                    &RedisEntryId::UserSpecifiedID {
                        milliseconds: "0".to_string(),
                        sequence_number: "0".to_string(),
                    },
                )
                .await;

            let ack_result = pool
                .stream_acknowledge_entries(&stream, group, vec!["0-1".to_string()])
                .await;

            ack_result.is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_stream_delete_entries() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let stream: RedisKey = format!("test_xdel_{uid}").into();

            let fields: Vec<(&str, &str)> = vec![("f", "v")];
            let _ = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                .await;

            let read_result = pool
                .stream_read_entries(
                    std::slice::from_ref(&stream),
                    vec!["0-0".to_string()],
                    Some(1),
                )
                .await
                .expect("failed to read");

            let stream_key = pool.add_prefix(&format!("test_xdel_{uid}"));
            let entries = read_result.get(&stream_key).expect("should have entries");
            if entries.is_empty() {
                return false;
            }
            let entry_id = entries.first().expect("checked non-empty").0.clone();

            let delete_result = pool.stream_delete_entries(&stream, vec![entry_id]).await;

            matches!(delete_result, Ok(count) if count >= 1)
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_stream_trim_entries() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let stream: RedisKey = format!("test_trim_{uid}").into();

            for i in 0..5 {
                let i_str = i.to_string();
                let fields: Vec<(&str, &str)> = vec![("idx", i_str.as_str())];
                let _ = pool
                    .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                    .await;
            }

            let len_before = pool.stream_get_length(&stream).await.unwrap();
            if len_before < 5 {
                return false;
            }

            let read_result = pool
                .stream_read_entries(
                    std::slice::from_ref(&stream),
                    vec!["0-0".to_string()],
                    Some(10),
                )
                .await
                .expect("failed to read stream");

            let stream_key = pool.add_prefix(&format!("test_trim_{uid}"));
            let entries = read_result.get(&stream_key).expect("should have entries");

            if entries.len() >= 3 {
                let trim_id = &entries.get(1).expect("checked len >= 3").0;
                let trim_result = pool
                    .stream_trim_entries(
                        &stream,
                        StreamTrimConfig::new(
                            StreamCapKind::MinID,
                            StreamCapTrim::Exact,
                            trim_id.as_str(),
                        ),
                    )
                    .await;

                match trim_result {
                    Ok(trimmed_count) => trimmed_count >= 1,
                    Err(e) => {
                        eprintln!("trim returned error: {:?}", e);
                        false
                    }
                }
            } else {
                false
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── Consumer Group Commands ──────────────────────────────────────────────────

#[tokio::test]
async fn test_consumer_group_operations() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let stream: RedisKey = format!("test_stream_group_ops_{uid}").into();
            let group = format!("test_group_{uid}");
            let fields: Vec<(&str, &str)> = vec![("field1", "value1")];

            let _ = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                .await;

            let create_result = pool
                .consumer_group_create(
                    &stream,
                    &group,
                    &RedisEntryId::UserSpecifiedID {
                        milliseconds: "0".to_string(),
                        sequence_number: "0".to_string(),
                    },
                )
                .await;

            create_result.is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_consumer_group_destroy() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let stream: RedisKey = format!("test_grp_destroy_{uid}").into();
            let group = format!("test_destroy_grp_{uid}");

            let fields: Vec<(&str, &str)> = vec![("f", "v")];
            let _ = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                .await;
            pool.consumer_group_create(
                &stream,
                &group,
                &RedisEntryId::UserSpecifiedID {
                    milliseconds: "0".to_string(),
                    sequence_number: "0".to_string(),
                },
            )
            .await
            .expect("failed to create group");

            let destroy_result = pool.consumer_group_destroy(&stream, &group).await;

            matches!(destroy_result, Ok(ConsumerGroupDestroyReply::Destroyed))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_consumer_group_delete_consumer() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let stream: RedisKey = format!("test_xdelconsumer_{uid}").into();
            let group = format!("test_grp_{uid}");
            let consumer = format!("test_consumer_{uid}");

            let fields: Vec<(&str, &str)> = vec![("f", "v")];
            let _ = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                .await;
            let _ = pool
                .consumer_group_create(
                    &stream,
                    &group,
                    &RedisEntryId::UserSpecifiedID {
                        milliseconds: "0".to_string(),
                        sequence_number: "0".to_string(),
                    },
                )
                .await;

            let _ = pool
                .stream_read_with_options(
                    std::slice::from_ref(&stream),
                    vec![RedisEntryId::UndeliveredEntryID.to_stream_id()],
                    Some(1),
                    None,
                    Some((&group, &consumer)),
                )
                .await;

            let result = pool
                .consumer_group_delete_consumer(&stream, &group, &consumer)
                .await;

            result.is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_consumer_group_set_last_id() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let stream: RedisKey = format!("test_xsetid_{uid}").into();
            let group = format!("test_grp_{uid}");

            let fields: Vec<(&str, &str)> = vec![("f", "v")];
            let _ = pool
                .stream_append_entry(&stream, &RedisEntryId::AutoGeneratedID, fields)
                .await;
            let _ = pool
                .consumer_group_create(
                    &stream,
                    &group,
                    &RedisEntryId::UserSpecifiedID {
                        milliseconds: "0".to_string(),
                        sequence_number: "0".to_string(),
                    },
                )
                .await;

            let result = pool
                .consumer_group_set_last_id(
                    &stream,
                    &group,
                    &RedisEntryId::UserSpecifiedID {
                        milliseconds: "0".to_string(),
                        sequence_number: "1".to_string(),
                    },
                )
                .await;

            matches!(result, Ok(id) if id == "0-1")
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── List Commands ────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_list_operations() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key: RedisKey = format!("test_list_{uid}").into();

            let append_result = pool.append_elements_to_list(&key, &["a", "b", "c"]).await;

            let length_result = pool.get_list_length(&key).await;

            let elements_result: Result<Vec<String>, _> = pool.get_list_elements(&key, 0, -1).await;

            let pop_result = pool.lpop_list_elements(&key, Some(1)).await;

            match (append_result, length_result, elements_result, pop_result) {
                (Ok(()), Ok(3), Ok(elems), Ok(popped)) => {
                    elems == vec!["a".to_string(), "b".to_string(), "c".to_string()]
                        && popped == vec!["a".to_string()]
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── Lua Scripts ──────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_setting_keys_using_scripts() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let lua_script = r#"
                for i = 1, #KEYS do
                    redis.call("INCRBY", KEYS[i], ARGV[i])
                end
                return
                "#;
            let mut keys_and_values = HashMap::new();
            for i in 0..10 {
                keys_and_values.insert(format!("key{i}"), i);
            }

            let key = keys_and_values.keys().cloned().collect::<Vec<_>>();
            let values = keys_and_values
                .values()
                .map(|val| val.to_string())
                .collect::<Vec<String>>();

            pool.evaluate_redis_script::<_, ()>(lua_script, key, values)
                .await
                .is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_getting_keys_using_scripts() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");

            for i in 0..3 {
                let key = format!("script_test_key{i}").into();
                let _ = pool.set_key(&key, format!("value{i}")).await;
            }

            let lua_script = r#"
                local results = {}
                for i = 1, #KEYS do
                    results[i] = redis.call("GET", KEYS[i])
                end
                return results
                "#;

            let keys = vec![
                "script_test_key0".to_string(),
                "script_test_key1".to_string(),
                "script_test_key2".to_string(),
            ];

            pool.evaluate_redis_script::<_, Vec<String>>(lua_script, keys, vec![""])
                .await
                .is_ok()
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── SCAN ─────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_scan_returns_matching_keys() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let uid = unique_test_id();
            let key1: RedisKey = format!("test_scan_foo_{uid}").into();
            let key2: RedisKey = format!("test_scan_bar_{uid}").into();

            let _ = pool.set_key(&key1, "v1".to_string()).await;
            let _ = pool.set_key(&key2, "v2".to_string()).await;

            let scan_pattern: RedisKey = format!("*test_scan_foo_{uid}*").into();
            let result = pool.scan(&scan_pattern, None, None).await;

            match result {
                Ok(keys) => {
                    let found_foo = keys
                        .iter()
                        .any(|k| k.contains(&format!("test_scan_foo_{uid}")));
                    let found_bar = keys
                        .iter()
                        .any(|k| k.contains(&format!("test_scan_bar_{uid}")));
                    found_foo && !found_bar
                }
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── Pub/Sub ──────────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_pubsub_standalone_publish_and_receive() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let channel = "test_pubsub_channel";
            let test_message = "test_message_value";

            pool.redis_conn
                .subscriber
                .subscribe(channel)
                .await
                .expect("failed to subscribe");

            let mut receiver = pool.redis_conn.subscriber.message_rx();

            tokio::time::sleep(std::time::Duration::from_millis(100)).await;

            pool.redis_conn
                .publisher
                .publish(channel, RedisValue::from_string(test_message.to_string()))
                .await
                .expect("failed to publish");

            let received =
                tokio::time::timeout(std::time::Duration::from_secs(5), receiver.recv()).await;

            match received {
                Ok(Ok(msg)) => {
                    let value_str = redis_value_to_option_string(&msg.value);
                    msg.channel == channel && value_str.as_deref() == Some(test_message)
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_subscriber_unsubscribe() {
    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let pool = test_connection(&RedisSettings::default())
                .await
                .expect("failed to create redis connection pool");
            let channel = "test_unsub_channel";

            pool.redis_conn
                .subscriber
                .subscribe(channel)
                .await
                .expect("failed to subscribe");

            let unsub_result = pool.redis_conn.subscriber.unsubscribe(channel).await;

            matches!(unsub_result, Ok(()))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

// ─── on_error / max_failure_threshold_seconds ─────────────────────────────────

#[tokio::test]
async fn test_on_error_triggers_shutdown_when_redis_unreachable() {
    let settings = RedisSettings {
        host: "192.0.2.1".to_string(), // RFC 5737 test address, guaranteed unreachable
        port: 1,
        unresponsive_check_interval: 1,
        max_failure_threshold_seconds: 2,
        reconnect_max_attempts: 1,
        default_command_timeout: 1,
        ..RedisSettings::default()
    };

    let pool = test_connection(&settings).await;
    let pool = match pool {
        Ok(pool) => pool,
        Err(_) => return, // If connection fails entirely, skip
    };

    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();

    tokio::spawn(async move {
        pool.redis_conn.on_error(shutdown_tx).await;
    });

    let result = tokio::time::timeout(std::time::Duration::from_secs(10), shutdown_rx).await;

    assert!(
        result.is_ok() && result.unwrap().is_ok(),
        "on_error should trigger shutdown signal when Redis is unreachable"
    );
}

#[tokio::test]
async fn test_on_error_keeps_redis_available_when_healthy() {
    let pool = test_connection(&RedisSettings::default())
        .await
        .expect("failed to create redis connection pool");

    let initial_state = pool
        .redis_conn
        .is_redis_available
        .load(std::sync::atomic::Ordering::SeqCst);

    let key: RedisKey = format!("test_health_{}", unique_test_id()).into();
    let set_result = pool.set_key(&key, "ok".to_string()).await;
    let get_result: Result<String, _> = pool.get_key(&key).await;

    if set_result.is_err() || get_result.is_err() {
        return;
    }

    assert!(initial_state, "Redis should be available when healthy");
}

#[tokio::test]
async fn test_on_error_marks_unavailable_after_threshold() {
    let settings = RedisSettings {
        host: "192.0.2.1".to_string(),
        port: 1,
        unresponsive_check_interval: 1,
        max_failure_threshold_seconds: 2,
        reconnect_max_attempts: 1,
        default_command_timeout: 1,
        ..RedisSettings::default()
    };

    let pool = test_connection(&settings).await;
    let pool = match pool {
        Ok(pool) => pool,
        Err(_) => return,
    };

    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
    let is_available = pool.redis_conn.is_redis_available.clone();

    tokio::spawn(async move {
        pool.redis_conn.on_error(shutdown_tx).await;
    });

    let result = tokio::time::timeout(std::time::Duration::from_secs(10), shutdown_rx).await;

    if result.is_ok() && result.unwrap().is_ok() {
        assert!(
            !is_available.load(std::sync::atomic::Ordering::SeqCst),
            "is_redis_available should be false after on_error triggers shutdown"
        );
    }
}

// ─── Cluster-mode tests ──────────────────────────────────────────────────────
// These tests run only when `REDIS_CLUSTER_URLS` env var is set.

#[tokio::test]
async fn test_cluster_set_get_delete() {
    let (pool, uid) = match get_cluster_pool_with_uid().await {
        Some(result) => result,
        None => return,
    };

    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let key: RedisKey = format!("test_cluster_sgd_{uid}").into();
            let value = "cluster_value".to_string();

            let set_result = pool.set_key(&key, value.clone()).await;
            let get_result: Result<String, _> = pool.get_key(&key).await;
            let del_result = pool.delete_key(&key).await;

            matches!(set_result, Ok(()))
                && matches!(get_result, Ok(v) if v == value)
                && matches!(del_result, Ok(DelReply::KeyDeleted))
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_cluster_hash_operations() {
    let (pool, uid) = match get_cluster_pool_with_uid().await {
        Some(result) => result,
        None => return,
    };

    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let key: RedisKey = format!("test_cluster_hash_{uid}").into();

            let set_result = pool
                .set_hash_fields(&key, vec![("field1", "val1"), ("field2", "val2")], Some(60))
                .await;

            let hsetnx_new = pool
                .set_hash_field_if_not_exist(&key, "field3", "val3", None)
                .await;

            let hsetnx_dup = pool
                .set_hash_field_if_not_exist(&key, "field1", "new_val", None)
                .await;

            let hget_result: Result<String, _> = pool.get_hash_field(&key, "field1").await;

            let hgetall_result: Result<HashMap<String, String>, _> =
                pool.get_hash_fields(&key).await;

            match (
                set_result,
                hsetnx_new,
                hsetnx_dup,
                hget_result,
                hgetall_result,
            ) {
                (Ok(()), Ok(HsetnxReply::KeySet), Ok(HsetnxReply::KeyNotSet), Ok(val), Ok(map)) => {
                    val == "val1" && map.contains_key("field1") && map.contains_key("field3")
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_cluster_stream_operations() {
    let (pool, uid) = match get_cluster_pool_with_uid().await {
        Some(result) => result,
        None => return,
    };

    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let stream: RedisKey = format!("test_cluster_stream_{uid}").into();
            let group = format!("test_cluster_grp_{uid}");

            let append_result = pool
                .stream_append_entry(
                    &stream,
                    &RedisEntryId::AutoGeneratedID,
                    vec![("task", "process")],
                )
                .await;

            let len_result = pool.stream_get_length(&stream).await;

            let group_result = pool
                .consumer_group_create(
                    &stream,
                    &group,
                    &RedisEntryId::UserSpecifiedID {
                        milliseconds: "0".to_string(),
                        sequence_number: "0".to_string(),
                    },
                )
                .await;

            let read_result = pool
                .stream_read_with_options(
                    std::slice::from_ref(&stream),
                    vec![RedisEntryId::UndeliveredEntryID.to_stream_id()],
                    Some(1),
                    None,
                    Some((&group, &format!("consumer_{uid}"))),
                )
                .await;

            let destroy_result = pool.consumer_group_destroy(&stream, &group).await;

            match (
                append_result,
                len_result,
                group_result,
                read_result,
                destroy_result,
            ) {
                (Ok(()), Ok(len), Ok(()), Ok(reply), Ok(_)) => {
                    len >= 1
                        && reply.len() == 1
                        && reply
                            .values()
                            .next()
                            .is_some_and(|entries| !entries.is_empty())
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_cluster_set_key_if_not_exists_and_get_value() {
    let (pool, uid) = match get_cluster_pool_with_uid().await {
        Some(result) => result,
        None => return,
    };

    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let key: RedisKey = format!("test_cluster_setnx_get_{uid}").into();

            let result_new = pool
                .set_key_if_not_exists_and_get_value(&key, "val1".to_string(), Some(30))
                .await;

            let result_exist = pool
                .set_key_if_not_exists_and_get_value(&key, "val2".to_string(), Some(30))
                .await;

            match (result_new, result_exist) {
                (Ok(SetGetReply::ValueSet(v1)), Ok(SetGetReply::ValueExists(v2))) => {
                    v1 == "val1" && v2 == "val1"
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_cluster_scan() {
    let (pool, uid) = match get_cluster_pool_with_uid().await {
        Some(result) => result,
        None => return,
    };

    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let key: RedisKey = format!("test_cluster_scan_{uid}").into();

            let _ = pool.set_key(&key, "v".to_string()).await;

            let pattern: RedisKey = format!("*test_cluster_scan_{uid}*").into();
            let result = pool.scan(&pattern, None, None).await;

            match result {
                Ok(keys) => keys
                    .iter()
                    .any(|k| k.contains(&format!("test_cluster_scan_{uid}"))),
                Err(_) => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}

#[tokio::test]
async fn test_cluster_pubsub() {
    let (pool, uid) = match get_cluster_pool_with_uid().await {
        Some(result) => result,
        None => return,
    };

    let is_success = tokio::task::spawn_blocking(move || {
        futures::executor::block_on(async {
            let channel = format!("test_cluster_pubsub_{uid}");
            let test_message = "cluster_message";

            pool.redis_conn
                .subscriber
                .subscribe(&channel)
                .await
                .expect("failed to subscribe on cluster");

            let mut receiver = pool.redis_conn.subscriber.message_rx();

            tokio::time::sleep(std::time::Duration::from_millis(100)).await;

            pool.redis_conn
                .publisher
                .publish(&channel, RedisValue::from_string(test_message.to_string()))
                .await
                .expect("failed to publish on cluster");

            let received =
                tokio::time::timeout(std::time::Duration::from_secs(5), receiver.recv()).await;

            match received {
                Ok(Ok(msg)) => {
                    let value_str = redis_value_to_option_string(&msg.value);
                    msg.channel == channel && value_str.as_deref() == Some(test_message)
                }
                _ => false,
            }
        })
    })
    .await
    .expect("Spawn block failure");

    assert!(is_success);
}
