use redis_rs::{server::Server, options::DBOption};
use redis::{Client, Commands, Connection};
use std::time::Duration;
use tokio::time::{sleep, timeout};
use tokio::sync::oneshot;

// Helper function to start a test server with clean data directory
async fn start_test_server(test_name: &str) -> (Server, u16) {
    use std::sync::atomic::{AtomicU16, Ordering};
    static PORT_COUNTER: AtomicU16 = AtomicU16::new(16400);
    
    // Get a unique port for this test
    let port = PORT_COUNTER.fetch_add(1, Ordering::SeqCst);
    
    // Ensure port is available by trying to bind to it first
    let mut attempts = 0;
    let final_port = loop {
        let test_port = port + attempts;
        match tokio::net::TcpListener::bind(format!("127.0.0.1:{}", test_port)).await {
            Ok(_) => break test_port,
            Err(_) => {
                attempts += 1;
                if attempts > 100 {
                    panic!("Could not find available port after 100 attempts");
                }
            }
        }
    };
    let test_dir = format!("/tmp/herodb_test_{}", test_name);
    
    // Clean up any existing test data
    let _ = std::fs::remove_dir_all(&test_dir);
    std::fs::create_dir_all(&test_dir).unwrap();
    
    let option = DBOption {
        dir: test_dir,
        port: final_port,
    };
    
    let server = Server::new(option).await;
    (server, final_port)
}

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

#[tokio::test]
async fn test_basic_ping() {
    let (mut server, port) = start_test_server("ping").await;
    
    // Start server in background
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    let result: String = redis::cmd("PING").query(&mut conn).unwrap();
    assert_eq!(result, "PONG");
}

#[tokio::test]
async fn test_string_operations() {
    let (mut server, port) = start_test_server("string").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // 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);
}

#[tokio::test]
async fn test_incr_operations() {
    let (mut server, port) = start_test_server("incr").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // Test INCR on non-existent key
    let result: i32 = conn.incr("counter", 1).unwrap();
    assert_eq!(result, 1);
    
    // Test INCR on existing key
    let result: i32 = conn.incr("counter", 1).unwrap();
    assert_eq!(result, 2);
    
    // Test INCR on string value (should fail)
    let _: () = conn.set("string", "hello").unwrap();
    let result: Result<i32, _> = conn.incr("string", 1);
    assert!(result.is_err());
}

#[tokio::test]
async fn test_hash_operations() {
    let (mut server, port) = start_test_server("hash").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // 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"]);
}

#[tokio::test]
async fn test_expiration() {
    let (mut server, port) = start_test_server("expiration").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // 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);
}

#[tokio::test]
async fn test_scan_operations() {
    let (mut server, port) = start_test_server("scan").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // 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(&mut 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(&mut conn).unwrap();
    result.sort();
    assert_eq!(result, vec!["key0", "key1", "key2", "key3", "key4"]);
}

#[tokio::test]
async fn test_hscan_operations() {
    let (mut server, port) = start_test_server("hscan").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // 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(&mut 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
}

#[tokio::test]
async fn test_transaction_operations() {
    let (mut server, port) = start_test_server("transaction").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // Test MULTI/EXEC
    let _: () = redis::cmd("MULTI").query(&mut conn).unwrap();
    let _: () = redis::cmd("SET").arg("key1").arg("value1").query(&mut conn).unwrap();
    let _: () = redis::cmd("SET").arg("key2").arg("value2").query(&mut conn).unwrap();
    let _: Vec<String> = redis::cmd("EXEC").query(&mut 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");
}

#[tokio::test]
async fn test_discard_transaction() {
    let (mut server, port) = start_test_server("discard").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // Test MULTI/DISCARD
    let _: () = redis::cmd("MULTI").query(&mut conn).unwrap();
    let _: () = redis::cmd("SET").arg("discard").arg("value").query(&mut conn).unwrap();
    let _: () = redis::cmd("DISCARD").query(&mut conn).unwrap();
    
    // Verify command was not executed
    let result: Option<String> = conn.get("discard").unwrap();
    assert_eq!(result, None);
}

#[tokio::test]
async fn test_type_command() {
    let (mut server, port) = start_test_server("type").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // Test string type
    let _: () = conn.set("string", "value").unwrap();
    let result: String = redis::cmd("TYPE").arg("string").query(&mut 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(&mut conn).unwrap();
    assert_eq!(result, "hash");
    
    // Test non-existent key
    let result: String = redis::cmd("TYPE").arg("noexist").query(&mut conn).unwrap();
    assert_eq!(result, "none");
}

#[tokio::test]
async fn test_config_commands() {
    let (mut server, port) = start_test_server("config").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // Test CONFIG GET databases
    let result: Vec<String> = redis::cmd("CONFIG")
        .arg("GET")
        .arg("databases")
        .query(&mut conn)
        .unwrap();
    assert_eq!(result, vec!["databases", "16"]);
    
    // Test CONFIG GET dir
    let result: Vec<String> = redis::cmd("CONFIG")
        .arg("GET")
        .arg("dir")
        .query(&mut conn)
        .unwrap();
    assert_eq!(result[0], "dir");
    assert!(result[1].contains("/tmp/herodb_test_config"));
}

#[tokio::test]
async fn test_info_command() {
    let (mut server, port) = start_test_server("info").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // Test INFO
    let result: String = redis::cmd("INFO").query(&mut conn).unwrap();
    assert!(result.contains("redis_version"));
    
    // Test INFO replication
    let result: String = redis::cmd("INFO").arg("replication").query(&mut conn).unwrap();
    assert!(result.contains("role:master"));
}

#[tokio::test]
async fn test_error_handling() {
    let (mut server, port) = start_test_server("error").await;
    
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{}", port))
            .await
            .unwrap();
        
        loop {
            if let Ok((stream, _)) = listener.accept().await {
                let _ = server.handle(stream).await;
            }
        }
    });
    
    sleep(Duration::from_millis(200)).await;
    
    let mut conn = get_redis_connection(port);
    
    // 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(&mut conn);
    assert!(result.is_err());
    
    // Test EXEC without MULTI
    let result: Result<Vec<String>, _> = redis::cmd("EXEC").query(&mut conn);
    assert!(result.is_err());
    
    // Test DISCARD without MULTI
    let result: Result<(), _> = redis::cmd("DISCARD").query(&mut conn);
    assert!(result.is_err());
}