// Background daemon: watches for file changes and keeps the structural index up to date.
// Provides a Unix socket interface for status, commands, and event notifications.

use std::fs;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use notify::{RecommendedWatcher, RecursiveMode, Watcher};
use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixListener;
use tokio::sync::{mpsc, watch, Mutex};
use tracing::{info, warn};

use astmap_core::diff::SymbolChange;
use astmap_core::model::ImpactEntry;
use astmap_core::port::{DiffStore, SymbolLookup};

use crate::infra::discovery::IgnoreWalkDiscovery;
use crate::infra::watcher::{classify_event, DEBOUNCE_MS};
use astmap_languages::DefaultLanguageRegistry;

/// Daemon process status.
#[derive(Debug, Serialize)]
pub struct DaemonStatus {
    pub running: bool,
    pub pid: Option<u32>,
    pub socket: Option<PathBuf>,
}

/// Maximum Unix socket path length (macOS sun_path limit is 104 bytes).
const MAX_SOCKET_PATH: usize = 104;

// ── Error type ──

#[derive(Debug, thiserror::Error)]
pub enum DaemonError {
    #[error("daemon already running (pid: {0})")]
    AlreadyRunning(u32),
    #[error("daemon not running")]
    NotRunning,
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
    #[error("scan error: {0}")]
    Scan(#[from] astmap_core::ScanError),
    #[error("notify error: {0}")]
    Notify(#[from] notify::Error),
}

// ── Socket protocol (JSON lines) ──

#[derive(Debug, Deserialize)]
#[serde(tag = "command")]
enum ClientCommand {
    #[serde(rename = "status")]
    Status,
    #[serde(rename = "scan")]
    Scan { full: Option<bool> },
    #[serde(rename = "subscribe")]
    Subscribe { filter: Option<String> },
    #[serde(rename = "shutdown")]
    Shutdown,
}

#[derive(Debug, Serialize)]
struct Event {
    event: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    data: Option<serde_json::Value>,
}

/// Structural change event pushed to subscribed clients.
/// Contains symbol-level diffs and transitive downstream impacts.
#[derive(Debug, Serialize)]
pub struct DaemonEvent {
    pub seq: u64,
    pub event: String,
    pub timestamp: String,
    pub trigger_files: Vec<String>,
    pub changed_symbols: Vec<SymbolChange>,
    pub downstream_impacts: Vec<ImpactEntry>,
    pub impact_count: usize,
}

/// Monotonic event sequence counter (in-memory only, resets on daemon restart).
static EVENT_SEQ: AtomicU64 = AtomicU64::new(1);

fn next_seq() -> u64 {
    EVENT_SEQ.fetch_add(1, Ordering::Relaxed)
}

fn chrono_now() -> String {
    use std::time::SystemTime;
    let now = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .unwrap_or_default();
    let secs = now.as_secs();
    // Simple ISO 8601 UTC timestamp without chrono dependency
    let (s, m, h) = (secs % 60, (secs / 60) % 60, (secs / 3600) % 24);
    let days = secs / 86400;
    // Approximate date calculation (good enough for timestamps)
    let (y, mo, d) = days_to_ymd(days);
    format!("{y:04}-{mo:02}-{d:02}T{h:02}:{m:02}:{s:02}Z")
}

fn days_to_ymd(mut days: u64) -> (u64, u64, u64) {
    // Days since Unix epoch to (year, month, day)
    let mut y = 1970;
    loop {
        let days_in_year = if is_leap(y) { 366 } else { 365 };
        if days < days_in_year {
            break;
        }
        days -= days_in_year;
        y += 1;
    }
    let months: [u64; 12] = if is_leap(y) {
        [31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    } else {
        [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    };
    let mut mo = 1;
    for &m in &months {
        if days < m {
            break;
        }
        days -= m;
        mo += 1;
    }
    (y, mo, days + 1)
}

fn is_leap(y: u64) -> bool {
    (y.is_multiple_of(4) && !y.is_multiple_of(100)) || y.is_multiple_of(400)
}

// ── Client state ──

static CLIENT_ID: AtomicU64 = AtomicU64::new(1);

fn next_client_id() -> u64 {
    CLIENT_ID.fetch_add(1, Ordering::Relaxed)
}

struct Client {
    id: u64,
    tx: mpsc::Sender<String>,
    subscribed: bool,
    filter: Option<glob::Pattern>,
}

// ── Path helpers ──

pub fn socket_path(project_dir: &Path) -> PathBuf {
    project_dir.join(".astmap").join("daemon.sock")
}

fn pid_path(project_dir: &Path) -> PathBuf {
    project_dir.join(".astmap").join("daemon.pid")
}

fn is_process_running(pid: u32) -> bool {
    std::process::Command::new("kill")
        .args(["-0", &pid.to_string()])
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .status()
        .is_ok_and(|s| s.success())
}

// ── Public API ──

/// Check if the daemon is running.
pub fn status(project_dir: &Path) -> DaemonStatus {
    let pid_file = pid_path(project_dir);
    if let Ok(content) = fs::read_to_string(&pid_file) {
        if let Ok(pid) = content.trim().parse::<u32>() {
            if is_process_running(pid) {
                return DaemonStatus {
                    running: true,
                    pid: Some(pid),
                    socket: Some(socket_path(project_dir)),
                };
            }
            // Stale PID file — clean up
            let _ = fs::remove_file(&pid_file);
            let _ = fs::remove_file(socket_path(project_dir));
        }
    }
    DaemonStatus {
        running: false,
        pid: None,
        socket: None,
    }
}

/// Stop the running daemon by sending SIGTERM.
pub fn stop(project_dir: &Path) -> Result<(), DaemonError> {
    let st = status(project_dir);
    if !st.running {
        return Err(DaemonError::NotRunning);
    }
    if let Some(pid) = st.pid {
        let _ = std::process::Command::new("kill")
            .arg(pid.to_string())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status();
        let _ = fs::remove_file(pid_path(project_dir));
        let _ = fs::remove_file(socket_path(project_dir));
    }
    Ok(())
}

/// Start the daemon in the foreground. Blocks until shutdown signal.
pub async fn start(project_dir: &Path) -> Result<(), DaemonError> {
    // Canonicalize to resolve symlinks (e.g. /var -> /private/var on macOS).
    // This ensures notify event paths match our project_dir for strip_prefix.
    let project_dir = project_dir
        .canonicalize()
        .unwrap_or(project_dir.to_path_buf());
    let project_dir = project_dir.as_path();

    // Ensure .astmap directory exists
    fs::create_dir_all(project_dir.join(".astmap"))?;

    // Validate socket path length (macOS sun_path limit: 104 bytes)
    let sock = socket_path(project_dir);
    if sock.as_os_str().len() >= MAX_SOCKET_PATH {
        return Err(DaemonError::Io(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            format!(
                "socket path too long ({} bytes, max {}): {}",
                sock.as_os_str().len(),
                MAX_SOCKET_PATH - 1,
                sock.display()
            ),
        )));
    }

    // Bind socket first — this is the atomic "claim" that prevents start() races.
    let listener = match UnixListener::bind(&sock) {
        Ok(l) => l,
        Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => {
            let st = status(project_dir);
            if st.running {
                return Err(DaemonError::AlreadyRunning(st.pid.unwrap_or(0)));
            }
            // Stale socket from crashed daemon — clean up and retry
            let _ = fs::remove_file(&sock);
            let _ = fs::remove_file(pid_path(project_dir));
            UnixListener::bind(&sock)?
        }
        Err(e) => return Err(DaemonError::Io(e)),
    };

    // Write PID file (socket already bound — we own the daemon slot)
    fs::write(pid_path(project_dir), std::process::id().to_string())?;
    info!("daemon: listening on {}", sock.display());

    // Initial scan with cleanup on failure
    let project = project_dir.to_path_buf();
    {
        let dir = project.clone();
        let scan_result = tokio::task::spawn_blocking(move || {
            let db = crate::infra::sqlite::open(&dir)
                .map_err(|e| DaemonError::Io(std::io::Error::other(e.to_string())))?;
            astmap_core::scanner::scan(
                &dir,
                db.as_ref(),
                false,
                &DefaultLanguageRegistry,
                &IgnoreWalkDiscovery,
            )
            .map_err(DaemonError::Scan)
        })
        .await;
        match scan_result {
            Ok(Ok(result)) => {
                info!(
                    "daemon: initial scan: {} files, {} symbols, {} deps",
                    result.files_total, result.symbols, result.dependencies
                );
            }
            Ok(Err(e)) => {
                let _ = fs::remove_file(&sock);
                let _ = fs::remove_file(pid_path(&project));
                return Err(e);
            }
            Err(e) => {
                let _ = fs::remove_file(&sock);
                let _ = fs::remove_file(pid_path(&project));
                return Err(DaemonError::Io(std::io::Error::other(e)));
            }
        }
    }

    let clients: Arc<Mutex<Vec<Client>>> = Arc::new(Mutex::new(Vec::new()));
    let scan_lock: Arc<tokio::sync::Mutex<()>> = Arc::new(tokio::sync::Mutex::new(()));
    let (shutdown_tx, _) = watch::channel(false);

    // Bridge file system events into async channel (bounded to prevent OOM)
    let (fs_tx, fs_rx) = mpsc::channel::<notify::Event>(256);
    let _watcher = {
        let mut w = RecommendedWatcher::new(
            move |res: Result<notify::Event, notify::Error>| {
                if let Ok(event) = res {
                    // try_send: drop excess events under heavy FS churn (debounce makes this safe)
                    let _ = fs_tx.try_send(event);
                }
            },
            notify::Config::default(),
        )?;
        w.watch(project_dir, RecursiveMode::Recursive)?;
        w // must stay alive
    };

    // Spawn scan pipeline
    spawn_scan_pipeline(
        project.clone(),
        fs_rx,
        clients.clone(),
        scan_lock.clone(),
        shutdown_tx.subscribe(),
    );

    // Spawn signal handler
    {
        let tx = shutdown_tx.clone();
        tokio::spawn(async move {
            let sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate());
            let sigint = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt());
            match (sigterm, sigint) {
                (Ok(mut term), Ok(mut int)) => {
                    tokio::select! {
                        _ = term.recv() => {}
                        _ = int.recv() => {}
                    }
                }
                _ => {
                    warn!("daemon: signal handler registration failed, use socket shutdown");
                    std::future::pending::<()>().await;
                }
            }
            let _ = tx.send(true);
        });
    }

    // Accept client connections until shutdown
    let mut shutdown_rx = shutdown_tx.subscribe();
    loop {
        tokio::select! {
            biased;
            _ = shutdown_rx.changed() => {
                if *shutdown_rx.borrow() { break; }
            }
            result = listener.accept() => {
                if let Ok((stream, _)) = result {
                    let conn_clients = clients.clone();
                    let conn_shutdown = shutdown_tx.clone();
                    let conn_scan_lock = scan_lock.clone();
                    let conn_dir = project.clone();
                    tokio::spawn(async move {
                        handle_client(stream, conn_clients, conn_shutdown, conn_scan_lock, conn_dir).await;
                    });
                }
            }
        }
    }

    // Cleanup
    info!("daemon: shutting down");
    let _ = fs::remove_file(socket_path(&project));
    let _ = fs::remove_file(pid_path(&project));
    Ok(())
}

// ── Scan pipeline ──

/// Convert an absolute path to a project-relative path string.
fn to_rel_path(abs: &Path, project_dir: &Path) -> Option<String> {
    abs.strip_prefix(project_dir)
        .ok()
        .and_then(|rel| rel.to_str())
        .map(|s| s.to_string())
}

/// Branch switch threshold: if more than this many files changed, emit index.rebuilt.
const BRANCH_SWITCH_THRESHOLD: usize = 50;

/// Default max depth for daemon impact analysis.
const DAEMON_IMPACT_DEPTH: i64 = 3;

/// Result of the 4-phase structural diff pipeline.
struct PipelineResult {
    scan_files: i64,
    scan_symbols: i64,
    scan_deps: i64,
    trigger_files: Vec<String>,
    changes: Vec<SymbolChange>,
    impacts: Vec<ImpactEntry>,
    is_branch_switch: bool,
}

fn spawn_scan_pipeline(
    project_dir: PathBuf,
    mut fs_rx: mpsc::Receiver<notify::Event>,
    clients: Arc<Mutex<Vec<Client>>>,
    scan_lock: Arc<tokio::sync::Mutex<()>>,
    mut shutdown_rx: watch::Receiver<bool>,
) {
    tokio::spawn(async move {
        let debounce = std::time::Duration::from_millis(DEBOUNCE_MS);
        let mut pending = false;
        let mut last_event = tokio::time::Instant::now();
        let mut changed_paths: std::collections::HashSet<String> = std::collections::HashSet::new();

        loop {
            tokio::select! {
                biased;
                _ = shutdown_rx.changed() => {
                    if *shutdown_rx.borrow() { break; }
                }
                Some(event) = fs_rx.recv() => {
                    if classify_event(&event, &project_dir).is_some() {
                        // Phase 1: collect changed file paths from notify events
                        for path in &event.paths {
                            if let Some(rel) = to_rel_path(path, &project_dir) {
                                changed_paths.insert(rel);
                            }
                        }
                        pending = true;
                        last_event = tokio::time::Instant::now();
                    }
                }
                _ = tokio::time::sleep(debounce), if pending => {
                    if last_event.elapsed() >= debounce {
                        // Check if any subscribers exist (skip heavy pipeline if none)
                        let has_subscribers = {
                            let cl = clients.lock().await;
                            cl.iter().any(|c| c.subscribed)
                        };

                        let trigger_files: Vec<String> = changed_paths.drain().collect();
                        let _guard = scan_lock.lock().await;
                        let dir = project_dir.clone();
                        let files_for_pipeline = trigger_files.clone();

                        let pipeline_result = tokio::task::spawn_blocking(move || -> Result<PipelineResult, String> {
                            let db = crate::infra::sqlite::open(&dir).map_err(|e| format!("db open: {e}"))?;

                            let is_branch_switch = files_for_pipeline.len() > BRANCH_SWITCH_THRESHOLD;

                            // Phase 2: snapshot BEFORE scan (scan overwrites DB symbols)
                            let old_snapshots: std::collections::HashMap<String, Vec<astmap_core::model::SymbolWithParent>> =
                                if has_subscribers && !is_branch_switch {
                                    files_for_pipeline.iter()
                                        .filter_map(|path| {
                                            db.all_symbols_for_file(path).ok()
                                                .map(|syms| (path.clone(), syms))
                                        })
                                        .collect()
                                } else {
                                    std::collections::HashMap::new()
                                };

                            // Phase 3: scan (incremental, hash-based skip internally)
                            let scan_result = astmap_core::scanner::scan(
                                &dir, db.as_ref(), false,
                                &DefaultLanguageRegistry, &IgnoreWalkDiscovery,
                            ).map_err(|e| format!("scan: {e}"))?;

                            if !has_subscribers || is_branch_switch {
                                return Ok(PipelineResult {
                                    scan_files: scan_result.files_total,
                                    scan_symbols: scan_result.symbols,
                                    scan_deps: scan_result.dependencies,
                                    trigger_files: files_for_pipeline,
                                    changes: Vec::new(),
                                    impacts: Vec::new(),
                                    is_branch_switch,
                                });
                            }

                            // Phase 4: compare + impact
                            let mut all_changes = Vec::new();
                            let mut all_mod_ids = Vec::new();

                            for path in &files_for_pipeline {
                                let old = old_snapshots.get(path).cloned().unwrap_or_default();
                                let current = db.all_symbols_for_file(path).unwrap_or_default();
                                let (changes, mod_ids) = astmap_core::diff::compare_db_symbols(&old, &current, path);
                                all_changes.extend(changes);
                                all_mod_ids.extend(mod_ids);
                            }

                            // Impact analysis for modified symbols
                            let mut impacts = Vec::new();
                            let mut seen_ids = std::collections::HashSet::new();
                            for id in &all_mod_ids {
                                seen_ids.insert(*id);
                            }
                            for sym_id in &all_mod_ids {
                                if let Ok(entries) = db.impact_analysis(*sym_id, DAEMON_IMPACT_DEPTH) {
                                    for entry in entries {
                                        if seen_ids.insert(entry.symbol_id) {
                                            impacts.push(entry);
                                        }
                                    }
                                }
                            }

                            Ok(PipelineResult {
                                scan_files: scan_result.files_total,
                                scan_symbols: scan_result.symbols,
                                scan_deps: scan_result.dependencies,
                                trigger_files: files_for_pipeline,
                                changes: all_changes,
                                impacts,
                                is_branch_switch,
                            })
                        })
                        .await;

                        match pipeline_result {
                            Ok(Ok(result)) => {
                                info!(
                                    "daemon: scan: {} files, {} symbols, {} deps ({} changes, {} impacts)",
                                    result.scan_files, result.scan_symbols, result.scan_deps,
                                    result.changes.len(), result.impacts.len()
                                );

                                // Broadcast to subscribers: DaemonEvent
                                if !result.changes.is_empty() || result.is_branch_switch {
                                    let daemon_evt = DaemonEvent {
                                        seq: next_seq(),
                                        event: if result.is_branch_switch {
                                            "index.rebuilt".to_string()
                                        } else {
                                            "impact.changed".to_string()
                                        },
                                        timestamp: chrono_now(),
                                        trigger_files: result.trigger_files.clone(),
                                        changed_symbols: result.changes,
                                        downstream_impacts: result.impacts,
                                        impact_count: 0, // set below
                                    };
                                    let impact_count = daemon_evt.downstream_impacts.len();
                                    let daemon_evt = DaemonEvent { impact_count, ..daemon_evt };

                                    if let Ok(msg) = serde_json::to_string(&daemon_evt) {
                                        let cl = clients.lock().await;
                                        for c in cl.iter() {
                                            if c.subscribed {
                                                let matches = c.filter.as_ref().is_none_or(|pat| {
                                                    daemon_evt.trigger_files.iter().any(|f| pat.matches(f))
                                                });
                                                if matches {
                                                    let _ = c.tx.try_send(msg.clone());
                                                }
                                            }
                                        }
                                    }
                                }

                                // Broadcast to non-subscribers: index_updated (backwards compat)
                                let evt = Event {
                                    event: "index_updated".to_string(),
                                    data: Some(serde_json::json!({
                                        "files": result.scan_files,
                                        "symbols": result.scan_symbols,
                                        "dependencies": result.scan_deps,
                                    })),
                                };
                                if let Ok(msg) = serde_json::to_string(&evt) {
                                    let mut cl = clients.lock().await;
                                    cl.retain(|c| {
                                        if c.subscribed {
                                            !c.tx.is_closed()
                                        } else {
                                            c.tx.try_send(msg.clone()).is_ok()
                                        }
                                    });
                                }
                            }
                            Ok(Err(e)) => warn!("daemon: {e}"),
                            Err(e) => warn!("daemon: scan task panicked: {e}"),
                        }
                        pending = false;
                    }
                }
            }
        }
    });
}

// ── Client handler ──

async fn handle_client(
    stream: tokio::net::UnixStream,
    clients: Arc<Mutex<Vec<Client>>>,
    shutdown_tx: watch::Sender<bool>,
    scan_lock: Arc<tokio::sync::Mutex<()>>,
    project_dir: PathBuf,
) {
    let (reader, mut writer) = stream.into_split();
    let reader = BufReader::new(reader);
    let (tx, mut rx) = mpsc::channel(64);

    // Register client with unique ID
    let client_id = next_client_id();
    {
        let mut cl = clients.lock().await;
        cl.push(Client {
            id: client_id,
            tx,
            subscribed: false,
            filter: None,
        });
    }

    // Send connected event
    if let Ok(msg) = serde_json::to_string(&Event {
        event: "connected".to_string(),
        data: None,
    }) {
        let _ = writer.write_all(format!("{}\n", msg).as_bytes()).await;
    }

    // Use lines() instead of read_line() — it's cancel-safe in tokio::select!
    let mut lines = reader.lines();

    loop {
        tokio::select! {
            result = lines.next_line() => {
                match result {
                    Ok(None) | Err(_) => break,
                    Ok(Some(line)) => {
                        if let Ok(cmd) = serde_json::from_str::<ClientCommand>(line.trim()) {
                            match cmd {
                                ClientCommand::Status => {
                                    let st = status(&project_dir);
                                    let evt = Event {
                                        event: "status".to_string(),
                                        data: serde_json::to_value(&st).ok(),
                                    };
                                    if let Ok(msg) = serde_json::to_string(&evt) {
                                        let _ = writer
                                            .write_all(format!("{}\n", msg).as_bytes())
                                            .await;
                                    }
                                }
                                ClientCommand::Scan { full } => {
                                    let full = full.unwrap_or(false);
                                    let _guard = scan_lock.lock().await;
                                    let dir = project_dir.clone();
                                    let scan_result = tokio::task::spawn_blocking(move || {
                                        let db = crate::infra::sqlite::open(&dir).ok()?;
                                        astmap_core::scanner::scan(&dir, db.as_ref(), full, &DefaultLanguageRegistry, &IgnoreWalkDiscovery).ok()
                                    })
                                    .await;

                                    let evt = match scan_result {
                                        Ok(Some(r)) => Event {
                                            event: "scan_complete".to_string(),
                                            data: Some(serde_json::json!({
                                                "files": r.files_total,
                                                "symbols": r.symbols,
                                                "dependencies": r.dependencies,
                                            })),
                                        },
                                        _ => Event {
                                            event: "error".to_string(),
                                            data: Some(serde_json::json!({"message": "scan failed"})),
                                        },
                                    };
                                    if let Ok(msg) = serde_json::to_string(&evt) {
                                        let _ = writer
                                            .write_all(format!("{}\n", msg).as_bytes())
                                            .await;
                                    }
                                }
                                ClientCommand::Subscribe { filter } => {
                                    let pattern = filter
                                        .as_deref()
                                        .and_then(|f| glob::Pattern::new(f).ok());
                                    // Mark this client as subscribed (by unique ID)
                                    {
                                        let mut cl = clients.lock().await;
                                        if let Some(c) =
                                            cl.iter_mut().find(|c| c.id == client_id)
                                        {
                                            c.subscribed = true;
                                            c.filter = pattern;
                                        }
                                    }
                                    let evt = Event {
                                        event: "subscribed".to_string(),
                                        data: filter.map(|f| serde_json::json!({"filter": f})),
                                    };
                                    if let Ok(msg) = serde_json::to_string(&evt) {
                                        let _ = writer
                                            .write_all(format!("{}\n", msg).as_bytes())
                                            .await;
                                    }
                                }
                                ClientCommand::Shutdown => {
                                    let _ = shutdown_tx.send(true);
                                    break;
                                }
                            }
                        }
                    }
                }
            }
            Some(msg) = rx.recv() => {
                let _ = writer.write_all(format!("{}\n", msg).as_bytes()).await;
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;
    use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
    use tokio::net::UnixStream;

    fn setup_project() -> tempfile::TempDir {
        let tmp = tempfile::tempdir().unwrap();
        let src = tmp.path().join("src");
        fs::create_dir_all(&src).unwrap();
        fs::write(src.join("main.rs"), "fn main() {}").unwrap();
        fs::create_dir_all(tmp.path().join(".astmap")).unwrap();
        tmp
    }

    #[test]
    fn test_status_not_running() {
        let tmp = setup_project();
        let st = status(tmp.path());
        assert!(!st.running);
        assert!(st.pid.is_none());
    }

    #[test]
    fn test_status_stale_pid() {
        let tmp = setup_project();
        // Write a PID that doesn't exist (99999999)
        fs::write(pid_path(tmp.path()), "99999999").unwrap();
        let st = status(tmp.path());
        assert!(!st.running);
        // Stale PID file should be cleaned up
        assert!(!pid_path(tmp.path()).exists());
    }

    #[test]
    fn test_stop_not_running() {
        let tmp = setup_project();
        let result = stop(tmp.path());
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("not running"));
    }

    #[test]
    fn test_daemon_error_display() {
        let err = DaemonError::AlreadyRunning(1234);
        assert!(err.to_string().contains("1234"));

        let err = DaemonError::NotRunning;
        assert!(err.to_string().contains("not running"));
    }

    #[test]
    fn test_socket_path() {
        let path = socket_path(Path::new("/project"));
        assert_eq!(path, PathBuf::from("/project/.astmap/daemon.sock"));
    }

    #[test]
    fn test_pid_path() {
        let path = pid_path(Path::new("/project"));
        assert_eq!(path, PathBuf::from("/project/.astmap/daemon.pid"));
    }

    #[tokio::test]
    async fn test_daemon_start_and_shutdown() {
        let tmp = setup_project();
        let dir = tmp.path().to_path_buf();

        // Start daemon in background
        let daemon_dir = dir.clone();
        let handle = tokio::spawn(async move { start(&daemon_dir).await });

        // Wait for socket to appear
        let sock = socket_path(&dir);
        for _ in 0..50 {
            if sock.exists() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }
        assert!(sock.exists(), "daemon socket should exist");

        // PID file should exist
        assert!(pid_path(&dir).exists(), "PID file should exist");

        // Connect and send shutdown
        let stream = UnixStream::connect(&sock).await.unwrap();
        let (reader, mut writer) = stream.into_split();
        let mut reader = BufReader::new(reader);

        // Read connected event
        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();
        assert!(line.contains("connected"));

        // Send shutdown
        writer
            .write_all(b"{\"command\":\"shutdown\"}\n")
            .await
            .unwrap();

        // Wait for daemon to exit
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle).await;
        assert!(result.is_ok(), "daemon should exit after shutdown");
        assert!(result.unwrap().unwrap().is_ok());

        // Cleanup files should be removed
        assert!(!sock.exists(), "socket should be cleaned up");
        assert!(!pid_path(&dir).exists(), "PID file should be cleaned up");
    }

    #[tokio::test]
    async fn test_daemon_scan_command() {
        let tmp = setup_project();
        let dir = tmp.path().to_path_buf();

        let daemon_dir = dir.clone();
        let handle = tokio::spawn(async move { start(&daemon_dir).await });

        // Wait for socket
        let sock = socket_path(&dir);
        for _ in 0..50 {
            if sock.exists() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }

        // Connect and send scan command
        let stream = UnixStream::connect(&sock).await.unwrap();
        let (reader, mut writer) = stream.into_split();
        let mut reader = BufReader::new(reader);

        // Read connected event
        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();
        assert!(line.contains("connected"));

        // Send scan command
        writer.write_all(b"{\"command\":\"scan\"}\n").await.unwrap();

        // Read scan result
        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();
        assert!(line.contains("scan_complete"));

        // Shutdown
        writer
            .write_all(b"{\"command\":\"shutdown\"}\n")
            .await
            .unwrap();

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

    #[tokio::test]
    async fn test_daemon_status_command() {
        let tmp = setup_project();
        let dir = tmp.path().to_path_buf();

        let daemon_dir = dir.clone();
        let handle = tokio::spawn(async move { start(&daemon_dir).await });

        let sock = socket_path(&dir);
        for _ in 0..50 {
            if sock.exists() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }

        let stream = UnixStream::connect(&sock).await.unwrap();
        let (reader, mut writer) = stream.into_split();
        let mut reader = BufReader::new(reader);

        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();

        // Send status command
        writer
            .write_all(b"{\"command\":\"status\"}\n")
            .await
            .unwrap();

        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();
        assert!(line.contains("\"running\":true"));

        // Shutdown
        writer
            .write_all(b"{\"command\":\"shutdown\"}\n")
            .await
            .unwrap();

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

    #[tokio::test]
    async fn test_daemon_already_running() {
        let tmp = setup_project();
        let dir = tmp.path().to_path_buf();

        let daemon_dir = dir.clone();
        let handle = tokio::spawn(async move { start(&daemon_dir).await });

        let sock = socket_path(&dir);
        for _ in 0..50 {
            if sock.exists() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }

        // Try to start again — should fail
        let result = start(&dir).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("already running"));

        // Clean up: send shutdown via socket
        let stream = UnixStream::connect(&sock).await.unwrap();
        let (_, mut writer) = stream.into_split();
        writer
            .write_all(b"{\"command\":\"shutdown\"}\n")
            .await
            .unwrap();

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

    #[tokio::test]
    async fn test_daemon_subscribe_command() {
        let tmp = setup_project();
        let dir = tmp.path().to_path_buf();

        let daemon_dir = dir.clone();
        let handle = tokio::spawn(async move { start(&daemon_dir).await });

        let sock = socket_path(&dir);
        for _ in 0..50 {
            if sock.exists() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }

        let stream = UnixStream::connect(&sock).await.unwrap();
        let (reader, mut writer) = stream.into_split();
        let mut reader = BufReader::new(reader);

        // Read connected event
        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();
        assert!(line.contains("connected"));

        // Send subscribe command
        writer
            .write_all(b"{\"command\":\"subscribe\"}\n")
            .await
            .unwrap();

        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();
        assert!(line.contains("subscribed"));

        // Subscribe with filter
        writer
            .write_all(b"{\"command\":\"subscribe\",\"filter\":\"src/**/*.rs\"}\n")
            .await
            .unwrap();

        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();
        assert!(line.contains("subscribed"));
        assert!(line.contains("src/**/*.rs"));

        // Shutdown
        writer
            .write_all(b"{\"command\":\"shutdown\"}\n")
            .await
            .unwrap();

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

    #[tokio::test]
    async fn test_daemon_structural_event() {
        let tmp = setup_project();
        let dir = tmp.path().to_path_buf();

        // Write initial source file
        let src = dir.join("src");
        fs::write(
            src.join("lib.rs"),
            "pub fn hello() -> &'static str { \"hello\" }\n",
        )
        .unwrap();

        // Start daemon (initial scan indexes lib.rs)
        let daemon_dir = dir.clone();
        let handle = tokio::spawn(async move { start(&daemon_dir).await });

        let sock = socket_path(&dir);
        for _ in 0..50 {
            if sock.exists() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }
        assert!(sock.exists(), "daemon socket should exist");

        // Connect subscriber
        let stream = UnixStream::connect(&sock).await.unwrap();
        let (reader, mut writer) = stream.into_split();
        let mut reader = BufReader::new(reader);

        // Read connected event
        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();
        assert!(line.contains("connected"));

        // Send subscribe
        writer
            .write_all(b"{\"command\":\"subscribe\"}\n")
            .await
            .unwrap();
        let mut line = String::new();
        reader.read_line(&mut line).await.unwrap();
        assert!(line.contains("subscribed"));

        // Modify source file: change function signature
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        fs::write(
            src.join("lib.rs"),
            "pub fn hello(name: &str) -> String { format!(\"hello {name}\") }\n",
        )
        .unwrap();

        // Wait for structural event (debounce + scan + compare)
        let event_result = tokio::time::timeout(std::time::Duration::from_secs(10), async {
            loop {
                let mut line = String::new();
                reader.read_line(&mut line).await.unwrap();
                let trimmed = line.trim().to_string();
                if trimmed.contains("impact.changed") {
                    return trimmed;
                }
            }
        })
        .await;

        assert!(
            event_result.is_ok(),
            "should receive impact.changed event within timeout"
        );
        let event_json = event_result.unwrap();

        // Parse and verify the structural event
        let evt: serde_json::Value = serde_json::from_str(&event_json).unwrap();
        assert_eq!(evt["event"], "impact.changed");
        assert!(evt["seq"].as_u64().unwrap() > 0);
        assert!(!evt["trigger_files"].as_array().unwrap().is_empty());

        let symbols = evt["changed_symbols"].as_array().unwrap();
        assert!(!symbols.is_empty(), "should detect changed symbols");

        // Find the hello function change
        let hello_change = symbols.iter().find(|s| s["name"].as_str() == Some("hello"));
        assert!(
            hello_change.is_some(),
            "should detect hello function change: {symbols:?}"
        );
        let hello = hello_change.unwrap();
        assert_eq!(hello["change_type"], "modified");
        assert_eq!(hello["kind"], "function");

        // Shutdown
        writer
            .write_all(b"{\"command\":\"shutdown\"}\n")
            .await
            .unwrap();
        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle).await;
    }
}
