use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use async_trait::async_trait;
use bytes::Bytes;
use dotenvy::dotenv;
use futures_util::future::BoxFuture;
use http::{Request, Response, StatusCode};
use http_body_util::{Empty, Full};
use sqlx;
use sunocore::middleware::cache::{Cache, CacheEntry, CacheLayer, CachePolicy};
use sunocore::models::studio::StudioProject;
use sunocore::services::studio::{GetStudioProject, GetStudioProjectRequest};
use sunocore::validation::env::{suno_api_key_unwrap, suno_base_url_unwrap};
use tower::{Service, ServiceBuilder, ServiceExt};
use uuid::Uuid;

// In-memory cache implementation for testing
#[derive(Clone)]
struct MemCache {
    store: Arc<Mutex<HashMap<String, CacheEntry>>>,
    /// Track how many times get() was called
    get_count: Arc<Mutex<usize>>,
    /// Track how many times set() was called
    set_count: Arc<Mutex<usize>>,
}

impl MemCache {
    fn new() -> Self {
        Self {
            store: Arc::new(Mutex::new(HashMap::new())),
            get_count: Arc::new(Mutex::new(0)),
            set_count: Arc::new(Mutex::new(0)),
        }
    }

    fn get_stats(&self) -> (usize, usize) {
        let gets = *self.get_count.lock().unwrap();
        let sets = *self.set_count.lock().unwrap();
        (gets, sets)
    }
}

#[async_trait]
impl Cache for MemCache {
    async fn get(&self, key: &str) -> Option<CacheEntry> {
        *self.get_count.lock().unwrap() += 1;
        self.store.lock().unwrap().get(key).cloned()
    }

    async fn set(&self, key: &str, value: Bytes, _policy: CachePolicy) {
        *self.set_count.lock().unwrap() += 1;
        self.store.lock().unwrap().insert(
            key.to_string(),
            CacheEntry {
                data: value,
                inserted_at: Instant::now(),
            },
        );
    }

    async fn invalidate(&self, key: &str) {
        self.store.lock().unwrap().remove(key);
    }

    async fn invalidate_pattern(&self, pattern: &str) {
        let prefix = pattern.trim_end_matches('*');
        self.store
            .lock()
            .unwrap()
            .retain(|k, _| !k.starts_with(prefix));
    }
}

// Small Tower service adapter around reqwest::Client
#[derive(Clone)]
struct ReqwestSvc {
    client: reqwest::Client,
    /// Track how many actual HTTP calls were made
    call_count: Arc<Mutex<usize>>,
}

impl ReqwestSvc {
    fn new(client: reqwest::Client) -> Self {
        Self {
            client,
            call_count: Arc::new(Mutex::new(0)),
        }
    }

    fn get_call_count(&self) -> usize {
        *self.call_count.lock().unwrap()
    }
}

impl Service<Request<Empty<Bytes>>> for ReqwestSvc {
    type Response = Response<Full<Bytes>>;
    type Error = reqwest::Error;
    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;

    fn poll_ready(
        &mut self,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        std::task::Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: Request<Empty<Bytes>>) -> Self::Future {
        *self.call_count.lock().unwrap() += 1;

        let client = self.client.clone();
        let method = req.method().clone();
        let uri = req.uri().to_string();
        let headers = req.headers().clone();

        Box::pin(async move {
            let mut builder = client.request(method, uri);
            for (name, value) in headers.iter() {
                builder = builder.header(name, value);
            }

            let res = builder.send().await?;
            let status = res.status();
            let ct = res.headers().get(http::header::CONTENT_TYPE).cloned();
            let body = res.bytes().await?;

            let mut resp = Response::new(Full::new(Bytes::from(body)));
            *resp.status_mut() = StatusCode::from_u16(status.as_u16()).unwrap();
            if let Some(ct) = ct {
                resp.headers_mut().insert(http::header::CONTENT_TYPE, ct);
            }
            Ok(resp)
        })
    }
}

/// Test that demonstrates cache middleware with immutable project fetching.
/// The first call hits the network, subsequent calls should hit the cache.
#[tokio::test]
async fn test_cache_middleware_with_studio_project() {
    let _ = dotenv();

    let base = suno_base_url_unwrap();
    let project_id = "a6d2bb30-0004-4860-b6bb-1639ca065957".to_string();
    let expected_uuid = Uuid::parse_str(&project_id).expect("valid project id");
    let api_key = suno_api_key_unwrap();

    // Set up the cache and HTTP client with headers
    let cache = MemCache::new();

    let mut headers = http::HeaderMap::new();
    headers.insert(
        http::header::CONTENT_TYPE,
        http::HeaderValue::from_static("application/json"),
    );
    if let Some(key) = api_key.as_ref() {
        headers.insert(
            http::header::AUTHORIZATION,
            http::HeaderValue::from_str(&format!("Bearer {}", key)).expect("valid header"),
        );
        headers.insert(
            "X-Api-Key",
            http::HeaderValue::from_str(key).expect("valid header"),
        );
    }

    let client = reqwest::Client::builder()
        .default_headers(headers)
        .timeout(Duration::from_secs(20))
        .build()
        .expect("build reqwest client");

    let http_svc = ReqwestSvc::new(client);
    let base_svc = GetStudioProject::new(http_svc.clone());

    // Wrap with cache middleware
    let cache_layer = CacheLayer::new(
        cache.clone(),
        // Key function: use project_id as cache key
        |req: &GetStudioProjectRequest| Some(format!("project:{}", req.project_id)),
        // Policy function: projects are immutable, cache forever
        |_req: &GetStudioProjectRequest| CachePolicy::Immutable,
    );

    let mut cached_svc = ServiceBuilder::new().layer(cache_layer).service(base_svc);

    let req = GetStudioProjectRequest {
        base_url: base.clone(),
        project_id: expected_uuid,
        api_key: api_key.clone(),
    };

    // === First call: should hit the network ===
    println!("Making first call (should hit network)...");
    let project1: StudioProject = cached_svc
        .ready()
        .await
        .expect("service ready")
        .call(req.clone())
        .await
        .expect("first call succeeded");

    assert_eq!(project1.id, expected_uuid, "mismatched project id");
    assert!(!project1.title.is_empty(), "title should not be empty");

    // Check that we made 1 HTTP call and 1 cache set
    let http_calls_1 = http_svc.get_call_count();
    let (cache_gets_1, cache_sets_1) = cache.get_stats();
    println!(
        "After first call: http_calls={}, cache_gets={}, cache_sets={}",
        http_calls_1, cache_gets_1, cache_sets_1
    );
    assert_eq!(http_calls_1, 1, "should have made 1 HTTP call");
    assert_eq!(cache_sets_1, 1, "should have set cache once");

    // === Second call: should hit the cache ===
    println!("\nMaking second call (should hit cache)...");
    let project2: StudioProject = cached_svc
        .ready()
        .await
        .expect("service ready")
        .call(req.clone())
        .await
        .expect("second call succeeded");

    assert_eq!(project2.id, expected_uuid, "mismatched project id");
    assert_eq!(project2.title, project1.title, "should be same project");

    // Check that we still only made 1 HTTP call (cached!)
    let http_calls_2 = http_svc.get_call_count();
    let (cache_gets_2, cache_sets_2) = cache.get_stats();
    println!(
        "After second call: http_calls={}, cache_gets={}, cache_sets={}",
        http_calls_2, cache_gets_2, cache_sets_2
    );
    assert_eq!(
        http_calls_2, 1,
        "should still have only 1 HTTP call (cache hit!)"
    );
    assert_eq!(cache_gets_2, 2, "should have checked cache twice");
    assert_eq!(cache_sets_2, 1, "should still have only 1 cache set");

    // === Third call: same result ===
    println!("\nMaking third call (should also hit cache)...");
    let project3: StudioProject = cached_svc
        .ready()
        .await
        .expect("service ready")
        .call(req.clone())
        .await
        .expect("third call succeeded");

    assert_eq!(project3.id, expected_uuid);

    let http_calls_3 = http_svc.get_call_count();
    let (cache_gets_3, _) = cache.get_stats();
    println!(
        "After third call: http_calls={}, cache_gets={}",
        http_calls_3, cache_gets_3
    );
    assert_eq!(http_calls_3, 1, "should still have only 1 HTTP call");
    assert_eq!(cache_gets_3, 3, "should have checked cache three times");

    println!("\n✓ Cache middleware working correctly!");
    println!("  - Made 3 service calls");
    println!("  - Only 1 HTTP request (the rest were cached)");
}

/// Test cache invalidation
#[tokio::test]
async fn test_cache_invalidation() {
    let cache = MemCache::new();

    // Set some entries
    cache
        .set("project:123", Bytes::from("data1"), CachePolicy::Immutable)
        .await;
    cache
        .set("project:456", Bytes::from("data2"), CachePolicy::Immutable)
        .await;
    cache
        .set("user:789", Bytes::from("data3"), CachePolicy::Immutable)
        .await;

    // Verify they exist
    assert!(cache.get("project:123").await.is_some());
    assert!(cache.get("project:456").await.is_some());
    assert!(cache.get("user:789").await.is_some());

    // Invalidate a specific key
    cache.invalidate("project:123").await;
    assert!(cache.get("project:123").await.is_none());
    assert!(cache.get("project:456").await.is_some());

    // Invalidate by pattern
    cache.invalidate_pattern("project:*").await;
    assert!(cache.get("project:456").await.is_none());
    assert!(cache.get("user:789").await.is_some());
}

// Mock service that returns StudioProject and tracks call count
#[derive(Clone)]
struct MockStudioService {
    call_count: Arc<Mutex<usize>>,
}

impl MockStudioService {
    fn new() -> Self {
        Self {
            call_count: Arc::new(Mutex::new(0)),
        }
    }

    fn get_call_count(&self) -> usize {
        *self.call_count.lock().unwrap()
    }
}

impl Service<GetStudioProjectRequest> for MockStudioService {
    type Response = StudioProject;
    type Error = String;
    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;

    fn poll_ready(
        &mut self,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        std::task::Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: GetStudioProjectRequest) -> Self::Future {
        let count = {
            let mut c = self.call_count.lock().unwrap();
            *c += 1;
            *c
        };

        Box::pin(async move {
            // Return a minimal valid StudioProject
            // Each call returns a different title so we can verify cache hit vs miss
            let title = format!("Mock Project Call #{}", count);
            let project_json = format!(
                r#"{{
                    "id": "{}",
                    "title": "{}",
                    "created_at": "2024-01-01T00:00:00Z",
                    "updated_at": "2024-01-01T00:00:00Z",
                    "archived": false,
                    "state": {{
                        "loop": {{"enabled": false, "endBeats": 0.0, "startBeats": 0.0}},
                        "timing": {{"type": "follow-track", "trackId": "{}", "fallbackBPS": 2.0}},
                        "tracks": [],
                        "metronome": {{"enabled": false, "amplitude": 1.0}},
                        "selection": {{
                            "trackIds": [],
                            "focusBeats": 0.0,
                            "anchorBeats": 0.0,
                            "focusedArea": "timeline",
                            "focusedTrackId": "{}"
                        }},
                        "markersRegistry": {{}},
                        "songFadeInBeats": 0.0,
                        "songFadeOutBeats": 0.0,
                        "lyricsCorrectionsByClipId": {{}}
                    }}
                }}"#,
                req.project_id, title, req.project_id, req.project_id
            );

            serde_json::from_str(&project_json).map_err(|e| e.to_string())
        })
    }
}

/// Test TTL policy with a mock service
/// Verifies that:
/// 1. First call hits the service and caches
/// 2. Second call within TTL hits cache (no service call)
/// 3. Third call after TTL expiry hits service again (cache stale)
#[tokio::test]
async fn test_ttl_policy() {
    use tokio::time::sleep;

    let cache = MemCache::new();
    let mock_service = MockStudioService::new();
    let project_id = Uuid::new_v4();

    // Create a cache layer with 1-second TTL
    let cache_layer = CacheLayer::new(
        cache.clone(),
        |req: &GetStudioProjectRequest| Some(format!("project:{}", req.project_id)),
        |_req: &GetStudioProjectRequest| CachePolicy::Ttl(1), // 1 second TTL
    );

    let mut cached_service = ServiceBuilder::new()
        .layer(cache_layer)
        .service(mock_service.clone());

    let req = GetStudioProjectRequest {
        base_url: "http://test".to_string(),
        project_id,
        api_key: None,
    };

    // === First call: should hit the service ===
    println!("First call (should hit service)...");
    let project1: StudioProject = cached_service
        .ready()
        .await
        .expect("service ready")
        .call(req.clone())
        .await
        .expect("first call succeeded");

    assert_eq!(project1.title, "Mock Project Call #1");
    assert_eq!(
        mock_service.get_call_count(),
        1,
        "should have called service once"
    );

    // === Second call within TTL: should hit cache ===
    println!("Second call within TTL (should hit cache)...");
    let project2: StudioProject = cached_service
        .ready()
        .await
        .expect("service ready")
        .call(req.clone())
        .await
        .expect("second call succeeded");

    assert_eq!(
        project2.title, "Mock Project Call #1",
        "should return cached result"
    );
    assert_eq!(
        mock_service.get_call_count(),
        1,
        "should still have only 1 service call"
    );

    // === Wait for TTL to expire ===
    println!("Waiting 1.5 seconds for TTL to expire...");
    sleep(Duration::from_millis(1500)).await;

    // === Third call after TTL: should hit service again ===
    println!("Third call after TTL expiry (should hit service again)...");
    let project3: StudioProject = cached_service
        .ready()
        .await
        .expect("service ready")
        .call(req.clone())
        .await
        .expect("third call succeeded");

    assert_eq!(
        project3.title, "Mock Project Call #2",
        "should have fresh data"
    );
    assert_eq!(
        mock_service.get_call_count(),
        2,
        "should have called service twice"
    );

    println!("\n✓ TTL policy working correctly!");
    println!("  - First call: service invoked, result cached");
    println!("  - Second call within TTL: cache hit");
    println!("  - Third call after TTL: service invoked again, cache refreshed");
}

// SQLite-based cache implementation for testing
#[derive(Clone)]
struct SqliteCache {
    pool: sqlx::SqlitePool,
}

impl SqliteCache {
    async fn new() -> Result<Self, sqlx::Error> {
        // Create an in-memory SQLite database
        let pool = sqlx::SqlitePool::connect("sqlite::memory:").await?;

        // Create the cache table
        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS cache (
                key TEXT PRIMARY KEY,
                value BLOB NOT NULL,
                policy TEXT NOT NULL,
                inserted_at INTEGER NOT NULL
            )
            "#,
        )
        .execute(&pool)
        .await?;

        Ok(Self { pool })
    }
}

#[async_trait]
impl Cache for SqliteCache {
    async fn get(&self, key: &str) -> Option<CacheEntry> {
        let row: Option<(Vec<u8>, i64)> =
            sqlx::query_as("SELECT value, inserted_at FROM cache WHERE key = ?")
                .bind(key)
                .fetch_optional(&self.pool)
                .await
                .ok()?;

        row.map(|(value, inserted_at_ms)| {
            // Convert milliseconds since epoch to Instant
            // Note: This is approximate since Instant doesn't have a stable epoch,
            // but it's fine for testing purposes
            let elapsed_ms = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_millis() as i64
                - inserted_at_ms;

            let inserted_at = if elapsed_ms > 0 {
                Instant::now() - Duration::from_millis(elapsed_ms as u64)
            } else {
                Instant::now()
            };

            CacheEntry {
                data: Bytes::from(value),
                inserted_at,
            }
        })
    }

    async fn set(&self, key: &str, value: Bytes, policy: CachePolicy) {
        let policy_str = match policy {
            CachePolicy::Immutable => "immutable",
            CachePolicy::Ttl(secs) => &format!("ttl:{}", secs),
            CachePolicy::NoCache => "nocache",
        };

        let inserted_at_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as i64;

        let _ = sqlx::query(
            r#"
            INSERT OR REPLACE INTO cache (key, value, policy, inserted_at)
            VALUES (?, ?, ?, ?)
            "#,
        )
        .bind(key)
        .bind(value.as_ref())
        .bind(policy_str)
        .bind(inserted_at_ms)
        .execute(&self.pool)
        .await;
    }

    async fn invalidate(&self, key: &str) {
        let _ = sqlx::query("DELETE FROM cache WHERE key = ?")
            .bind(key)
            .execute(&self.pool)
            .await;
    }

    async fn invalidate_pattern(&self, pattern: &str) {
        // Convert glob pattern to SQL LIKE pattern
        let sql_pattern = pattern.replace('*', "%");
        let _ = sqlx::query("DELETE FROM cache WHERE key LIKE ?")
            .bind(sql_pattern)
            .execute(&self.pool)
            .await;
    }
}

/// Test cache middleware with SQLite backend
/// Verifies that the cache middleware works with a real database backend
#[tokio::test]
async fn test_sqlite_cache_backend() {
    let cache = SqliteCache::new().await.expect("create sqlite cache");
    let mock_service = MockStudioService::new();
    let project_id = Uuid::new_v4();

    // Create cache layer with immutable policy
    let cache_layer = CacheLayer::new(
        cache.clone(),
        |req: &GetStudioProjectRequest| Some(format!("project:{}", req.project_id)),
        |_req: &GetStudioProjectRequest| CachePolicy::Immutable,
    );

    let mut cached_service = ServiceBuilder::new()
        .layer(cache_layer)
        .service(mock_service.clone());

    let req = GetStudioProjectRequest {
        base_url: "http://test".to_string(),
        project_id,
        api_key: None,
    };

    println!("Testing SQLite cache backend...");

    // === First call: should hit service and cache in SQLite ===
    println!("First call (should hit service)...");
    let project1: StudioProject = cached_service
        .ready()
        .await
        .expect("service ready")
        .call(req.clone())
        .await
        .expect("first call succeeded");

    assert_eq!(project1.title, "Mock Project Call #1");
    assert_eq!(mock_service.get_call_count(), 1);

    // === Verify data is in SQLite ===
    let row_count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM cache")
        .fetch_one(&cache.pool)
        .await
        .expect("count cache entries");
    assert_eq!(row_count.0, 1, "should have 1 entry in SQLite");

    // === Second call: should hit SQLite cache ===
    println!("Second call (should hit SQLite cache)...");
    let project2: StudioProject = cached_service
        .ready()
        .await
        .expect("service ready")
        .call(req.clone())
        .await
        .expect("second call succeeded");

    assert_eq!(
        project2.title, "Mock Project Call #1",
        "should return cached result"
    );
    assert_eq!(
        mock_service.get_call_count(),
        1,
        "should still have only 1 service call"
    );

    // === Test manual invalidation ===
    println!("Invalidating cache...");
    cache.invalidate(&format!("project:{}", project_id)).await;

    let row_count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM cache")
        .fetch_one(&cache.pool)
        .await
        .expect("count cache entries");
    assert_eq!(row_count.0, 0, "cache should be empty after invalidation");

    // === Third call: should hit service again ===
    println!("Third call after invalidation (should hit service again)...");
    let project3: StudioProject = cached_service
        .ready()
        .await
        .expect("service ready")
        .call(req.clone())
        .await
        .expect("third call succeeded");

    assert_eq!(
        project3.title, "Mock Project Call #2",
        "should have fresh data"
    );
    assert_eq!(
        mock_service.get_call_count(),
        2,
        "should have called service twice"
    );

    println!("\n✓ SQLite cache backend working correctly!");
    println!("  - First call: service invoked, cached in SQLite");
    println!("  - Second call: retrieved from SQLite cache");
    println!("  - After invalidation: service invoked again");
}

/// Test SQLite cache with pattern invalidation
#[tokio::test]
async fn test_sqlite_cache_pattern_invalidation() {
    let cache = SqliteCache::new().await.expect("create sqlite cache");

    // Insert multiple entries
    cache
        .set("project:123", Bytes::from("data1"), CachePolicy::Immutable)
        .await;
    cache
        .set("project:456", Bytes::from("data2"), CachePolicy::Immutable)
        .await;
    cache
        .set("user:789", Bytes::from("data3"), CachePolicy::Immutable)
        .await;

    // Verify all are present
    assert!(cache.get("project:123").await.is_some());
    assert!(cache.get("project:456").await.is_some());
    assert!(cache.get("user:789").await.is_some());

    // Invalidate by pattern
    cache.invalidate_pattern("project:*").await;

    // Verify only project entries were removed
    assert!(cache.get("project:123").await.is_none());
    assert!(cache.get("project:456").await.is_none());
    assert!(
        cache.get("user:789").await.is_some(),
        "user entry should remain"
    );

    println!("✓ SQLite pattern invalidation working correctly!");
}