//! End-to-end smoke test that does NOT require Chrome.
//! Runs the daemon in a background task and exercises lifecycle + error paths.

mod support;

use night_fury_daemon_core::cli::make_req;
use night_fury_daemon_core::client;
use serde_json::json;
use tempfile::tempdir;

#[tokio::test]
async fn daemon_start_status_shutdown_roundtrip() {
    // Initialise the registry so the debug_assert in TailFinSession::launch
    // doesn't fire. An empty registry still returns the expected "unknown site"
    // error for "bogus", which is what the assertion below tests.
    tail_fin_daemon::driver::register_sites(vec![]);

    let dir = tempdir().unwrap();
    let sock = dir.path().join("tfd.sock").to_string_lossy().to_string();

    let sock_bg = sock.clone();
    let daemon = tokio::spawn(async move {
        let _ = tail_fin_daemon::server::serve(&sock_bg, 300, 900, 3).await;
    });

    // Wait for socket to appear.
    for _ in 0..30 {
        if tokio::net::UnixStream::connect(&sock).await.is_ok() {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }

    let status_req = make_req("daemon.status", None, json!({}));
    let status = client::send(&sock, &status_req).await.unwrap();
    assert!(status.ok, "status response not ok: {:?}", status.error);
    assert_eq!(
        status
            .data
            .as_ref()
            .and_then(|d| d.get("sessions"))
            .and_then(|v| v.as_u64()),
        Some(0)
    );

    // Asking for an unknown site should return a clear error.
    let bad_acquire = make_req(
        "session.acquire",
        None,
        json!({"site": "bogus", "host": "127.0.0.1:9222"}),
    );
    let resp = client::send(&sock, &bad_acquire).await.unwrap();
    assert!(!resp.ok);
    assert!(resp.error.as_deref().unwrap_or("").contains("unknown site"));

    // Shutdown cleanly.
    let shutdown = make_req("daemon.shutdown", None, json!({}));
    let resp = client::send(&sock, &shutdown).await.unwrap();
    assert!(resp.ok);

    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn session_acquire_creates_new_session() {
    let dir = tempdir().unwrap();
    let (sock, daemon) = support::spawn_mock_daemon(dir.path(), 300, 900, 3).await;

    let req = make_req(
        "session.acquire",
        None,
        json!({"site": "sa", "host": "127.0.0.1:9222", "mode": "in_use"}),
    );
    let resp = client::send(&sock, &req).await.unwrap();
    assert!(resp.ok, "acquire failed: {:?}", resp.error);
    let data = resp.data.unwrap();
    assert!(data.get("session_id").is_some());
    assert_eq!(data.get("reused").and_then(|v| v.as_bool()), Some(false));

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn session_acquire_reuses_idle() {
    let dir = tempdir().unwrap();
    let (sock, daemon) = support::spawn_mock_daemon(dir.path(), 300, 900, 3).await;

    // First acquire creates a new session, mark it in_use...
    let r1 = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "127.0.0.1:9222", "mode": "in_use"}),
        ),
    )
    .await
    .unwrap();
    let sid1 = r1.data.unwrap()["session_id"].as_str().unwrap().to_string();

    // ...then release it back to idle.
    let rel = client::send(&sock, &make_req("session.release", Some(&sid1), json!({})))
        .await
        .unwrap();
    assert!(rel.ok);

    // Second acquire should pick up the same session.
    let r2 = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "127.0.0.1:9222", "mode": "in_use"}),
        ),
    )
    .await
    .unwrap();
    let data = r2.data.unwrap();
    assert_eq!(data["session_id"].as_str().unwrap(), &sid1);
    assert_eq!(data["reused"].as_bool(), Some(true));

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn session_acquire_busy_waits_until_release() {
    let dir = tempdir().unwrap();
    // max_per_pool = 1 so the second acquire must wait.
    let (sock, daemon) = support::spawn_mock_daemon(dir.path(), 300, 900, 1).await;

    // Hold one session in_use.
    let _r1 = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "127.0.0.1:9222", "mode": "in_use"}),
        ),
    )
    .await
    .unwrap();

    // Fire a second acquire in the background — it should block on the busy
    // loop (250ms poll, 30s deadline).
    let sock_for_wait = sock.clone();
    let wait_task = tokio::spawn(async move {
        client::send(
            &sock_for_wait,
            &make_req(
                "session.acquire",
                None,
                json!({"site": "sa", "host": "127.0.0.1:9222", "mode": "in_use"}),
            ),
        )
        .await
        .unwrap()
    });

    // Let it block for ~500ms.
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    assert!(
        !wait_task.is_finished(),
        "acquire should still be blocked on a full pool"
    );

    // Grab the sid of the first session via session.list, then release it.
    let list = client::send(&sock, &make_req("session.list", None, json!({})))
        .await
        .unwrap();
    let sid1 = list.data.unwrap()["sessions"][0]["session_id"]
        .as_str()
        .unwrap()
        .to_string();
    let _ = client::send(&sock, &make_req("session.release", Some(&sid1), json!({}))).await;

    // The waiting acquire should now unblock and complete successfully.
    let r2 = tokio::time::timeout(std::time::Duration::from_secs(5), wait_task)
        .await
        .expect("second acquire task did not finish within 5s")
        .unwrap();
    assert!(r2.ok, "second acquire should succeed after release");

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn session_release_marks_idle() {
    let dir = tempdir().unwrap();
    let (sock, daemon) = support::spawn_mock_daemon(dir.path(), 300, 900, 3).await;

    let r = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "127.0.0.1:9222", "mode": "acquired"}),
        ),
    )
    .await
    .unwrap();
    let sid = r.data.unwrap()["session_id"].as_str().unwrap().to_string();

    let rel = client::send(&sock, &make_req("session.release", Some(&sid), json!({})))
        .await
        .unwrap();
    assert!(rel.ok);
    assert_eq!(rel.data.unwrap()["released"].as_str(), Some(sid.as_str()));

    // session.list should now show state: idle
    let list = client::send(&sock, &make_req("session.list", None, json!({})))
        .await
        .unwrap();
    let state = list.data.unwrap()["sessions"][0]["state"]
        .as_str()
        .map(String::from)
        .unwrap();
    assert_eq!(state, "idle");

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn session_list_shows_state_snake_case() {
    let dir = tempdir().unwrap();
    let (sock, daemon) = support::spawn_mock_daemon(dir.path(), 300, 900, 3).await;

    // Acquired mode → state "acquired" in snake_case
    let r = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "h", "mode": "acquired"}),
        ),
    )
    .await
    .unwrap();
    assert!(r.ok);

    let list = client::send(&sock, &make_req("session.list", None, json!({})))
        .await
        .unwrap();
    let sessions = list.data.unwrap()["sessions"].as_array().unwrap().clone();
    assert_eq!(sessions.len(), 1);
    let state = sessions[0]["state"].as_str().unwrap();
    assert_eq!(state, "acquired", "state should be snake_case, got {state}");

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn session_destroy_removes_from_registry_and_pool() {
    let dir = tempdir().unwrap();
    let (sock, daemon) = support::spawn_mock_daemon(dir.path(), 300, 900, 3).await;

    let r = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "h", "mode": "acquired"}),
        ),
    )
    .await
    .unwrap();
    let sid = r.data.unwrap()["session_id"].as_str().unwrap().to_string();

    // daemon.status reports 1 session.
    let pre = client::send(&sock, &make_req("daemon.status", None, json!({})))
        .await
        .unwrap();
    assert_eq!(pre.data.unwrap()["sessions"].as_u64(), Some(1));

    // Destroy.
    let destroy = client::send(&sock, &make_req("session.destroy", Some(&sid), json!({})))
        .await
        .unwrap();
    assert!(destroy.ok);

    // Now daemon.status → 0 sessions, session.list → empty.
    let post_status = client::send(&sock, &make_req("daemon.status", None, json!({})))
        .await
        .unwrap();
    assert_eq!(post_status.data.unwrap()["sessions"].as_u64(), Some(0));
    let post_list = client::send(&sock, &make_req("session.list", None, json!({})))
        .await
        .unwrap();
    assert_eq!(
        post_list.data.unwrap()["sessions"]
            .as_array()
            .unwrap()
            .len(),
        0
    );

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn business_cmd_routes_to_driver_handle() {
    let dir = tempdir().unwrap();
    let (sock, daemon) = support::spawn_mock_daemon(dir.path(), 300, 900, 3).await;

    let r = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "h", "mode": "acquired"}),
        ),
    )
    .await
    .unwrap();
    let sid = r.data.unwrap()["session_id"].as_str().unwrap().to_string();

    // Send a business cmd; MockDriver echoes the cmd name back in data.
    let biz = client::send(
        &sock,
        &make_req("some.business.cmd", Some(&sid), json!({"arg": 1})),
    )
    .await
    .unwrap();
    assert!(biz.ok, "business cmd error: {:?}", biz.error);
    let data = biz.data.unwrap();
    assert_eq!(data["mock_ok"].as_bool(), Some(true));
    assert_eq!(data["cmd"].as_str(), Some("some.business.cmd"));

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn business_cmd_after_destroy_errors_and_leaves_pool_clean() {
    // Covers the "session not found" branch of handle_request, which is one
    // path into the dead-session eviction logic. The direct driver-error
    // branch (MockDriver.fail_handle = true) requires propagating extra params
    // through session.acquire and is deferred to a follow-up.
    let dir = tempdir().unwrap();
    let (sock, daemon) = support::spawn_mock_daemon(dir.path(), 300, 900, 3).await;

    let r = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "h", "mode": "acquired"}),
        ),
    )
    .await
    .unwrap();
    let sid = r.data.unwrap()["session_id"].as_str().unwrap().to_string();

    // Destroy it out from under ourselves.
    let _ = client::send(&sock, &make_req("session.destroy", Some(&sid), json!({}))).await;

    // Now send a business cmd — this should return an error because the
    // session is gone from the registry.
    let biz = client::send(&sock, &make_req("some.biz.cmd", Some(&sid), json!({})))
        .await
        .unwrap();
    assert!(!biz.ok);
    assert!(biz.error.unwrap_or_default().contains("not found"));

    // Confirm the pool/registry are clean.
    let list = client::send(&sock, &make_req("session.list", None, json!({})))
        .await
        .unwrap();
    assert_eq!(
        list.data.unwrap()["sessions"].as_array().unwrap().len(),
        0,
        "pool should be empty after destroy"
    );

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn acquired_session_survives_idle_timeout() {
    let dir = tempdir().unwrap();

    // idle_timeout = 3s so a stale session would be reaped after ~3s.
    // keepalive fires every 500ms, so an Acquired session gets touched
    // ~12 times during the 6s test and never ages past the threshold.
    let (sock, daemon) = support::spawn_mock_daemon_with_keepalive(
        dir.path(),
        /* idle_timeout */ 3,
        /* daemon_idle */ 900,
        /* max_per_pool */ 3,
        /* keepalive_period */ std::time::Duration::from_millis(500),
    )
    .await;

    // Acquire as Acquired (long-lived).
    let r = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "h", "mode": "acquired"}),
        ),
    )
    .await
    .unwrap();
    assert!(r.ok);
    let sid = r.data.unwrap()["session_id"].as_str().unwrap().to_string();

    // Wait longer than idle_timeout. Without keepalive, the session would be
    // reaped by the 1s reaper tick. With keepalive at 500ms, it should stay
    // alive the whole time.
    tokio::time::sleep(std::time::Duration::from_secs(6)).await;

    // Verify the session is STILL in the registry (not just in the pool — the
    // pool would retain stale entries even if the reaper destroyed the registry
    // entry, which is the "divergence" bug noted as a follow-up).
    let status = client::send(&sock, &make_req("daemon.status", None, json!({})))
        .await
        .unwrap();
    let registry_count = status
        .data
        .as_ref()
        .and_then(|d| d.get("sessions"))
        .and_then(|v| v.as_u64())
        .unwrap_or(0);
    assert_eq!(
        registry_count, 1,
        "Acquired session was reaped from the registry despite keepalive"
    );

    // And belt-and-suspenders: the pool should still show it as 'acquired'.
    let list = client::send(&sock, &make_req("session.list", None, json!({})))
        .await
        .unwrap();
    let sessions = list.data.unwrap()["sessions"].as_array().unwrap().clone();
    let entry = sessions
        .iter()
        .find(|s| s["session_id"].as_str() == Some(&sid))
        .expect("pool lost our session");
    assert_eq!(
        entry["state"].as_str(),
        Some("acquired"),
        "session state changed unexpectedly"
    );

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test]
async fn idle_session_gets_reaped_without_keepalive_touch() {
    let dir = tempdir().unwrap();

    // Same timings but the acquired session is released immediately, so it
    // becomes Idle. Keepalive only touches Acquired sessions, so Idle should
    // be reaped after ~3s idle_timeout.
    let (sock, daemon) = support::spawn_mock_daemon_with_keepalive(
        dir.path(),
        /* idle_timeout */ 3,
        /* daemon_idle */ 900,
        /* max_per_pool */ 3,
        /* keepalive_period */ std::time::Duration::from_millis(500),
    )
    .await;

    let r = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "sa", "host": "h", "mode": "acquired"}),
        ),
    )
    .await
    .unwrap();
    let sid = r.data.unwrap()["session_id"].as_str().unwrap().to_string();

    // Release immediately → session becomes Idle (not protected by keepalive).
    let _ = client::send(&sock, &make_req("session.release", Some(&sid), json!({}))).await;

    // Wait long enough for the reaper to destroy it.
    tokio::time::sleep(std::time::Duration::from_secs(6)).await;

    // The reaper removes sessions from the registry; daemon.status reflects
    // the registry count (not the pool snapshot). An Idle session that has
    // aged past idle_timeout should be gone from the registry.
    let status = client::send(&sock, &make_req("daemon.status", None, json!({})))
        .await
        .unwrap();
    let count = status.data.unwrap()["sessions"].as_u64().unwrap_or(1);
    assert_eq!(
        count, 0,
        "Idle session should have been reaped from the registry after idle_timeout"
    );

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_acquires_respect_pool_cap() {
    let dir = tempdir().unwrap();
    let (sock, daemon) = support::spawn_mock_daemon(dir.path(), 300, 900, 3).await;

    // Fire 10 parallel acquires for the same key.
    // With max_per_pool = 3, the pool SHOULD hand out at most 3 distinct
    // session IDs. However, concurrent `CreateNew` outcomes from `acquire_atomic`
    // can transiently exceed the cap by up to (N - max) when N concurrent
    // callers hit an empty pool — this is a documented pre-existing race
    // (see PR #88 review).
    //
    // This test verifies the invariant is "bounded" (no runaway session
    // creation) rather than exactly `== max`.
    let mut tasks = Vec::new();
    for _ in 0..10 {
        let sock = sock.clone();
        tasks.push(tokio::spawn(async move {
            let resp = client::send(
                &sock,
                &make_req(
                    "session.acquire",
                    None,
                    json!({"site": "sa", "host": "h", "mode": "in_use"}),
                ),
            )
            .await;
            resp.ok()
        }));
    }

    // Give the bursty acquires ~500ms to all get their initial responses.
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;

    // Release everything currently in the pool so any still-waiting acquires
    // can proceed (reuse path).
    let list = client::send(&sock, &make_req("session.list", None, json!({})))
        .await
        .unwrap();
    let sessions = list.data.unwrap()["sessions"].as_array().unwrap().clone();
    for s in &sessions {
        if let Some(sid) = s["session_id"].as_str() {
            let _ = client::send(&sock, &make_req("session.release", Some(sid), json!({}))).await;
        }
    }

    // Collect results. Some tasks may still be looping on the busy path
    // (250ms poll, 30s deadline) — give them time to complete via reuse.
    let mut ok_count = 0usize;
    let mut session_ids = std::collections::HashSet::new();
    for task in tasks {
        if let Ok(Some(resp)) = task.await {
            if resp.ok {
                if let Some(sid) = resp.data.and_then(|d| {
                    d.get("session_id")
                        .and_then(|v| v.as_str())
                        .map(String::from)
                }) {
                    session_ids.insert(sid);
                    ok_count += 1;
                }
            }
        }
    }

    assert!(ok_count > 0, "no acquires succeeded");
    // max=3, plus generous slack of 3 for the CreateNew race = 6 upper bound.
    // In practice we usually see 3-4 distinct IDs.
    assert!(
        session_ids.len() <= 6,
        "too many distinct sessions handed out: {} (cap=3, slack ≤3 for CreateNew race)",
        session_ids.len()
    );

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}

// ── Pool/registry self-heal: a reaped-but-still-pooled session must NOT be
// handed back out (the reaper removes idle sessions from the registry but not
// the pool; acquire must reconcile rather than poison the pool key forever).
#[tokio::test]
async fn reaped_idle_session_self_heals_on_reacquire() {
    let dir = tempdir().unwrap();
    // idle_timeout=1s, reaper every 1s, large keepalive (only touches Acquired,
    // and our session is released to Idle, so it ages out and gets reaped).
    let (sock, daemon) = support::spawn_mock_daemon_with_keepalive(
        dir.path(),
        1,
        900,
        0,
        std::time::Duration::from_secs(60),
    )
    .await;

    // Acquire (in_use), run a command (touches the registry), release → Idle.
    let r1 = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "mock", "host": "h", "mode": "in_use"}),
        ),
    )
    .await
    .unwrap();
    assert!(r1.ok, "first acquire failed: {:?}", r1.error);
    let sid1 = r1.data.unwrap()["session_id"].as_str().unwrap().to_string();
    let _ = client::send(&sock, &make_req("mock.ping", Some(&sid1), json!({}))).await;
    let _ = client::send(&sock, &make_req("session.release", Some(&sid1), json!({}))).await;

    // Wait for the reaper to remove sid1 from the registry (idle_timeout 1s +
    // reap interval 1s, with margin).
    tokio::time::sleep(std::time::Duration::from_secs(3)).await;

    // Re-acquire: must hand out a LIVE session, not the reaped sid1.
    let r2 = client::send(
        &sock,
        &make_req(
            "session.acquire",
            None,
            json!({"site": "mock", "host": "h", "mode": "in_use"}),
        ),
    )
    .await
    .unwrap();
    assert!(r2.ok, "re-acquire failed: {:?}", r2.error);
    let sid2 = r2.data.unwrap()["session_id"].as_str().unwrap().to_string();
    assert_ne!(sid2, sid1, "re-acquire returned the reaped session_id");

    // The decisive check: a command on the re-acquired session must succeed
    // (before the fix, sid2 == the dead sid1 and this returns "session not found").
    let biz = client::send(&sock, &make_req("mock.ping", Some(&sid2), json!({})))
        .await
        .unwrap();
    assert!(
        biz.ok,
        "command on re-acquired session failed: {:?}",
        biz.error
    );

    let _ = client::send(&sock, &make_req("daemon.shutdown", None, json!({}))).await;
    let _ = tokio::time::timeout(std::time::Duration::from_secs(3), daemon).await;
}
