use redis::{Client, Commands, Connection};
use std::process::{Child, Command};
use std::time::Duration;
use tokio::time::sleep;

// Helper function to get Redis connection, retrying until successful
fn get_redis_connection(port: u16) -> Connection {
    let connection_info = format!("redis://127.0.0.1:{}", port);
    let client = Client::open(connection_info).unwrap();
    let mut attempts = 0;
    loop {
        match client.get_connection() {
            Ok(mut conn) => {
                if redis::cmd("PING").query::<String>(&mut conn).is_ok() {
                    return conn;
                }
            }
            Err(e) => {
                if attempts >= 20 {
                    panic!(
                        "Failed to connect to Redis server after 20 attempts: {}",
                        e
                    );
                }
            }
        }
        attempts += 1;
        std::thread::sleep(Duration::from_millis(100));
    }
}

// A guard to ensure the server process is killed when it goes out of scope
struct ServerProcessGuard {
    process: Child,
    test_dir: String,
}

impl Drop for ServerProcessGuard {
    fn drop(&mut self) {
        println!("Killing server process (pid: {})...", self.process.id());
        if let Err(e) = self.process.kill() {
            eprintln!("Failed to kill server process: {}", e);
        }
        match self.process.wait() {
            Ok(status) => println!("Server process exited with: {}", status),
            Err(e) => eprintln!("Failed to wait on server process: {}", e),
        }

        // Clean up the specific test directory
        println!("Cleaning up test directory: {}", self.test_dir);
        if let Err(e) = std::fs::remove_dir_all(&self.test_dir) {
            eprintln!("Failed to clean up test directory: {}", e);
        }
    }
}

// Helper to set up the server and return a connection
fn setup_server() -> (ServerProcessGuard, u16) {
    use std::sync::atomic::{AtomicU16, Ordering};
    static PORT_COUNTER: AtomicU16 = AtomicU16::new(16400);
    let port = PORT_COUNTER.fetch_add(1, Ordering::SeqCst);

    let test_dir = format!("/tmp/herodb_test_{}", port);

    // Clean up previous test data
    if std::path::Path::new(&test_dir).exists() {
        let _ = std::fs::remove_dir_all(&test_dir);
    }
    std::fs::create_dir_all(&test_dir).unwrap();

    // Start the server in a subprocess
    let child = Command::new("cargo")
        .args(&[
            "run",
            "--",
            "--dir",
            &test_dir,
            "--port",
            &port.to_string(),
        ])
        .spawn()
        .expect("Failed to start server process");
    
    // Create a new guard that also owns the test directory path
    let guard = ServerProcessGuard {
        process: child,
        test_dir,
    };
    
    // Give the server a moment to start
    std::thread::sleep(Duration::from_millis(100));

    (guard, port)
}

#[tokio::test]
async fn all_tests() {
    let (_server_guard, port) = setup_server();
    let mut conn = get_redis_connection(port);

    // Run all tests using the same connection
    cleanup_keys(&mut conn).await;
    test_basic_ping(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_string_operations(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_incr_operations(&mut conn).await;
    // cleanup_keys(&mut conn).await;
    // test_hash_operations(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_expiration(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_scan_operations(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_scan_with_count(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_hscan_operations(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_transaction_operations(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_discard_transaction(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_type_command(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_config_commands(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_info_command(&mut conn).await;
    cleanup_keys(&mut conn).await;
    test_error_handling(&mut conn).await;
    
    // Clean up keys after all tests
    cleanup_keys(&mut conn).await;
}

async fn cleanup_keys(conn: &mut Connection) {
    let keys: Vec<String> = redis::cmd("KEYS").arg("*").query(conn).unwrap();
    if !keys.is_empty() {
        let _: () = redis::cmd("DEL").arg(keys).query(conn).unwrap();
    }
}

async fn test_basic_ping(conn: &mut Connection) {
    let result: String = redis::cmd("PING").query(conn).unwrap();
    assert_eq!(result, "PONG");

}

async fn test_string_operations(conn: &mut Connection) {
    // Test SET
    let _: () = conn.set("key", "value").unwrap();

    // Test GET
    let result: String = conn.get("key").unwrap();
    assert_eq!(result, "value");

    // Test GET non-existent key
    let result: Option<String> = conn.get("noexist").unwrap();
    assert_eq!(result, None);

    // Test DEL
    let deleted: i32 = conn.del("key").unwrap();
    assert_eq!(deleted, 1);

    // Test GET after DEL
    let result: Option<String> = conn.get("key").unwrap();
    assert_eq!(result, None);
}

async fn test_incr_operations(conn: &mut Connection) {
    // Test INCR on non-existent key
    let result: i32 = redis::cmd("INCR").arg("counter").query(conn).unwrap();
    assert_eq!(result, 1);

    // Test INCR on existing key
    let result: i32 = redis::cmd("INCR").arg("counter").query(conn).unwrap();
    assert_eq!(result, 2);

    // Test INCR on string value (should fail)
    let _: () = conn.set("string", "hello").unwrap();
    let result: Result<i32, _> = redis::cmd("INCR").arg("string").query(conn);
    assert!(result.is_err());
}

async fn test_hash_operations(conn: &mut Connection) {
    // Test HSET
    let result: i32 = conn.hset("hash", "field1", "value1").unwrap();
    assert_eq!(result, 1); // 1 new field

    // Test HGET
    let result: String = conn.hget("hash", "field1").unwrap();
    assert_eq!(result, "value1");

    // Test HSET multiple fields
    let _: () = conn.hset_multiple("hash", &[("field2", "value2"), ("field3", "value3")]).unwrap();

    // Test HGETALL
    let result: std::collections::HashMap<String, String> = conn.hgetall("hash").unwrap();
    assert_eq!(result.len(), 3);
    assert_eq!(result.get("field1").unwrap(), "value1");
    assert_eq!(result.get("field2").unwrap(), "value2");
    assert_eq!(result.get("field3").unwrap(), "value3");

    // Test HLEN
    let result: i32 = conn.hlen("hash").unwrap();
    assert_eq!(result, 3);

    // Test HEXISTS
    let result: bool = conn.hexists("hash", "field1").unwrap();
    assert_eq!(result, true);

    let result: bool = conn.hexists("hash", "noexist").unwrap();
    assert_eq!(result, false);

    // Test HDEL
    let result: i32 = conn.hdel("hash", "field1").unwrap();
    assert_eq!(result, 1);

    // Test HKEYS
    let mut result: Vec<String> = conn.hkeys("hash").unwrap();
    result.sort();
    assert_eq!(result, vec!["field2", "field3"]);

    // Test HVALS
    let mut result: Vec<String> = conn.hvals("hash").unwrap();
    result.sort();
    assert_eq!(result, vec!["value2", "value3"]);
}

async fn test_expiration(conn: &mut Connection) {
    // Test SETEX (expire in 1 second)
    let _: () = conn.set_ex("expkey", "value", 1).unwrap();

    // Test TTL
    let result: i32 = conn.ttl("expkey").unwrap();
    assert!(result == 1 || result == 0); // Should be 1 or 0 seconds

    // Test EXISTS
    let result: bool = conn.exists("expkey").unwrap();
    assert_eq!(result, true);

    // Wait for expiration
    sleep(Duration::from_millis(1100)).await;

    // Test GET after expiration
    let result: Option<String> = conn.get("expkey").unwrap();
    assert_eq!(result, None);

    // Test TTL after expiration
    let result: i32 = conn.ttl("expkey").unwrap();
    assert_eq!(result, -2); // Key doesn't exist

    // Test EXISTS after expiration
    let result: bool = conn.exists("expkey").unwrap();
    assert_eq!(result, false);
}

async fn test_scan_operations(conn: &mut Connection) {
    // Set up test data
    for i in 0..5 {
        let _: () = conn.set(format!("key{}", i), format!("value{}", i)).unwrap();
    }

    // Test SCAN
    let result: (u64, Vec<String>) = redis::cmd("SCAN")
        .arg(0)
        .arg("MATCH")
        .arg("*")
        .arg("COUNT")
        .arg(10)
        .query(conn)
        .unwrap();

    let (cursor, keys) = result;
    assert_eq!(cursor, 0); // Should complete in one scan
    assert_eq!(keys.len(), 5);

    // Test KEYS
    let mut result: Vec<String> = redis::cmd("KEYS").arg("*").query(conn).unwrap();
    result.sort();
    assert_eq!(result, vec!["key0", "key1", "key2", "key3", "key4"]);
}

async fn test_scan_with_count(conn: &mut Connection) {
    // Clean up previous keys
    let keys: Vec<String> = redis::cmd("KEYS").arg("scan_key*").query(conn).unwrap();
    if !keys.is_empty() {
        let _: () = redis::cmd("DEL").arg(keys).query(conn).unwrap();
    }

    // Set up test data
    for i in 0..15 {
        let _: () = conn.set(format!("scan_key{}", i), i).unwrap();
    }

    let mut cursor = 0;
    let mut all_keys = std::collections::HashSet::new();

    // First SCAN
    let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
        .arg(cursor)
        .arg("MATCH")
        .arg("scan_key*")
        .arg("COUNT")
        .arg(5)
        .query(conn)
        .unwrap();

    assert_ne!(next_cursor, 0);
    assert_eq!(keys.len(), 5);
    for key in keys {
        all_keys.insert(key);
    }
    cursor = next_cursor;

    // Second SCAN
    let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
        .arg(cursor)
        .arg("MATCH")
        .arg("scan_key*")
        .arg("COUNT")
        .arg(5)
        .query(conn)
        .unwrap();

    assert_ne!(next_cursor, 0);
    assert_eq!(keys.len(), 5);
    for key in keys {
        all_keys.insert(key);
    }
    cursor = next_cursor;

    // Final SCAN
    let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
        .arg(cursor)
        .arg("MATCH")
        .arg("scan_key*")
        .arg("COUNT")
        .arg(5)
        .query(conn)
        .unwrap();

    assert_eq!(next_cursor, 0);
    assert_eq!(keys.len(), 5);
    for key in keys {
        all_keys.insert(key);
    }

    assert_eq!(all_keys.len(), 15);
}

async fn test_hscan_operations(conn: &mut Connection) {
    // Set up hash data
    for i in 0..3 {
        let _: () = conn.hset("testhash", format!("field{}", i), format!("value{}", i)).unwrap();
    }

    // Test HSCAN
    let result: (u64, Vec<String>) = redis::cmd("HSCAN")
        .arg("testhash")
        .arg(0)
        .arg("MATCH")
        .arg("*")
        .arg("COUNT")
        .arg(10)
        .query(conn)
        .unwrap();

    let (cursor, fields) = result;
    assert_eq!(cursor, 0); // Should complete in one scan
    assert_eq!(fields.len(), 6); // 3 field-value pairs = 6 elements
}

async fn test_transaction_operations(conn: &mut Connection) {
    // Test MULTI/EXEC
    let _: () = redis::cmd("MULTI").query(conn).unwrap();
    let _: () = redis::cmd("SET").arg("key1").arg("value1").query(conn).unwrap();
    let _: () = redis::cmd("SET").arg("key2").arg("value2").query(conn).unwrap();
    let _: Vec<String> = redis::cmd("EXEC").query(conn).unwrap();

    // Verify commands were executed
    let result: String = conn.get("key1").unwrap();
    assert_eq!(result, "value1");

    let result: String = conn.get("key2").unwrap();
    assert_eq!(result, "value2");
}

async fn test_discard_transaction(conn: &mut Connection) {
    // Test MULTI/DISCARD
    let _: () = redis::cmd("MULTI").query(conn).unwrap();
    let _: () = redis::cmd("SET").arg("discard").arg("value").query(conn).unwrap();
    let _: () = redis::cmd("DISCARD").query(conn).unwrap();

    // Verify command was not executed
    let result: Option<String> = conn.get("discard").unwrap();
    assert_eq!(result, None);
}

async fn test_type_command(conn: &mut Connection) {
    // Test string type
    let _: () = conn.set("string", "value").unwrap();
    let result: String = redis::cmd("TYPE").arg("string").query(conn).unwrap();
    assert_eq!(result, "string");

    // Test hash type
    let _: () = conn.hset("hash", "field", "value").unwrap();
    let result: String = redis::cmd("TYPE").arg("hash").query(conn).unwrap();
    assert_eq!(result, "hash");

    // Test non-existent key
    let result: String = redis::cmd("TYPE").arg("noexist").query(conn).unwrap();
    assert_eq!(result, "none");
}

async fn test_config_commands(conn: &mut Connection) {
    // Test CONFIG GET databases
    let result: Vec<String> = redis::cmd("CONFIG")
        .arg("GET")
        .arg("databases")
        .query(conn)
        .unwrap();
    assert_eq!(result, vec!["databases", "16"]);

    // Test CONFIG GET dir
    let result: Vec<String> = redis::cmd("CONFIG")
        .arg("GET")
        .arg("dir")
        .query(conn)
        .unwrap();
    assert_eq!(result[0], "dir");
    assert!(result[1].contains("/tmp/herodb_test_"));
}

async fn test_info_command(conn: &mut Connection) {
    // Test INFO
    let result: String = redis::cmd("INFO").query(conn).unwrap();
    assert!(result.contains("redis_version"));

    // Test INFO replication
    let result: String = redis::cmd("INFO").arg("replication").query(conn).unwrap();
    assert!(result.contains("role:master"));
}

async fn test_error_handling(conn: &mut Connection) {
    // Test WRONGTYPE error - try to use hash command on string
    let _: () = conn.set("string", "value").unwrap();
    let result: Result<String, _> = conn.hget("string", "field");
    assert!(result.is_err());

    // Test unknown command
    let result: Result<String, _> = redis::cmd("UNKNOWN").query(conn);
    assert!(result.is_err());

    // Test EXEC without MULTI
    let result: Result<Vec<String>, _> = redis::cmd("EXEC").query(conn);
    assert!(result.is_err());

    // Test DISCARD without MULTI
    let result: Result<(), _> = redis::cmd("DISCARD").query(conn);
    assert!(result.is_err());
}