use std::path::{Path, PathBuf};
use std::sync::atomic::AtomicU64;
use std::sync::Arc;
use std::time::Duration;

use notify::{Config, Event, RecommendedWatcher, RecursiveMode, Watcher};
use tokio::sync::{mpsc, Semaphore};
use tracing::{debug, info, warn};

use crate::actors::ActorRegistry;
use crate::cache::NclCache;
use crate::config::DaemonNclConfig;
use crate::config_reload::ReloadAction;
use crate::notifications::NotificationStore;
use crate::registry::{KeyEntry, ProjectContext, ProjectRegistry};

/// Directories to watch for NCL changes relative to a project root.
const WATCH_DIRS: &[&str] = &[".ontology", "adrs", "reflection", "ontology"];

/// File watcher that invalidates the NCL cache on filesystem changes
/// and pushes notifications to the notification store.
pub struct FileWatcher {
    _watcher: RecommendedWatcher,
    debounce_task: tokio::task::JoinHandle<()>,
}

impl FileWatcher {
    /// Cancel the debounce task and stop watching.
    ///
    /// The underlying `notify` watcher drops when `FileWatcher` is dropped.
    /// Calling `abort()` before drop ensures the async task is cancelled
    /// immediately rather than running until its next yield point.
    pub fn abort(&self) {
        self.debounce_task.abort();
    }
}

/// Optional dependencies injected into the file watcher.
pub struct WatcherDeps {
    /// Project slug used for DB seeding, NATS events, and notification scoping.
    /// Falls back to the directory name of `project_root` when empty.
    pub slug: String,
    #[cfg(feature = "db")]
    pub db: Option<Arc<stratum_db::StratumDb>>,
    pub import_path: Option<String>,
    pub notifications: Arc<NotificationStore>,
    pub actors: Arc<ActorRegistry>,
    #[cfg(feature = "nats")]
    pub nats: Option<Arc<crate::nats::NatsPublisher>>,
    /// Shared with `ProjectContext` — prevents concurrent re-seeds per project.
    pub seed_lock: Arc<Semaphore>,
    /// Shared with `ProjectContext` — incremented after each successful seed.
    pub ontology_version: Arc<AtomicU64>,
    /// Shared with `ProjectContext` — per-file change counters, keyed by
    /// canonical path. Incremented unconditionally on every cache invalidation.
    pub file_versions: Arc<dashmap::DashMap<std::path::PathBuf, u64>>,
}

impl FileWatcher {
    /// Start watching NCL-relevant directories under `project_root`.
    ///
    /// Changes are debounced (200ms) before invalidating the cache.
    /// A periodic full invalidation runs every `full_invalidation_secs` as
    /// safety net.
    pub fn start(
        project_root: &Path,
        cache: Arc<NclCache>,
        full_invalidation_secs: u64,
        deps: WatcherDeps,
    ) -> std::result::Result<Self, crate::error::DaemonError> {
        let (tx, rx) = mpsc::channel::<Vec<PathBuf>>(256);
        let project_root_owned = project_root
            .canonicalize()
            .unwrap_or_else(|_| project_root.to_path_buf());

        let tx_notify = tx.clone();
        let mut watcher = RecommendedWatcher::new(
            move |res: std::result::Result<Event, notify::Error>| match res {
                Ok(event) => {
                    let ncl_paths: Vec<PathBuf> = event
                        .paths
                        .into_iter()
                        .filter(|p| {
                            p.extension()
                                .is_some_and(|ext| ext == "ncl" || ext == "jsonl")
                        })
                        .collect();
                    if !ncl_paths.is_empty() {
                        let _ = tx_notify.try_send(ncl_paths);
                    }
                }
                Err(e) => warn!(error = %e, "file watcher error"),
            },
            Config::default(),
        )
        .map_err(|e| crate::error::DaemonError::Watcher(e.to_string()))?;

        let mut watched_count = 0;
        for dir_name in WATCH_DIRS {
            let dir = project_root.join(dir_name);
            if dir.is_dir() {
                if let Err(e) = watcher.watch(&dir, RecursiveMode::Recursive) {
                    warn!(dir = %dir.display(), error = %e, "failed to watch directory");
                } else {
                    info!(dir = %dir.display(), "watching directory");
                    watched_count += 1;
                }
            }
        }
        info!(watched_count, "file watcher started");

        let debounce_task = tokio::spawn(debounce_loop(
            rx,
            cache,
            project_root_owned,
            full_invalidation_secs,
            deps,
        ));

        Ok(Self {
            _watcher: watcher,
            debounce_task,
        })
    }
}

/// Debounce filesystem events: collect paths over 200ms windows, then
/// invalidate once. Also runs periodic full invalidation as safety net.
/// Pushes notifications to the store and optionally publishes via NATS.
async fn debounce_loop(
    mut rx: mpsc::Receiver<Vec<PathBuf>>,
    cache: Arc<NclCache>,
    project_root: PathBuf,
    full_invalidation_secs: u64,
    deps: WatcherDeps,
) {
    let debounce = Duration::from_millis(200);
    let effective_secs = if full_invalidation_secs == 0 {
        60
    } else {
        full_invalidation_secs
    };
    let mut full_tick = tokio::time::interval(Duration::from_secs(effective_secs));
    full_tick.tick().await; // consume immediate first tick

    let project_name = if deps.slug.is_empty() {
        project_root
            .file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("unknown")
            .to_string()
    } else {
        deps.slug.clone()
    };

    loop {
        tokio::select! {
            recv = rx.recv() => match recv {
                None => {
                    debug!("watcher channel closed — debounce task exiting");
                    return;
                }
                Some(paths) => {
                    // Collect all events within debounce window
                    let mut all_paths = paths;
                    tokio::time::sleep(debounce).await;
                    while let Ok(more) = rx.try_recv() {
                        all_paths.extend(more);
                    }

                    // Canonicalize, deduplicate, and invalidate.
                    let mut canonical: Vec<PathBuf> = all_paths
                        .into_iter()
                        .filter_map(|p| p.canonicalize().ok())
                        .collect();
                    canonical.sort();
                    canonical.dedup();

                    let file_names: Vec<String> = canonical
                        .iter()
                        .filter_map(|p| p.file_name())
                        .map(|n| n.to_string_lossy().to_string())
                        .collect();

                    for path in &canonical {
                        cache.invalidate_file(path);
                        *deps.file_versions.entry(path.clone()).or_insert(0) += 1;
                    }

                    info!(
                        files = canonical.len(),
                        names = %file_names.join(", "),
                        "cache invalidated — files changed"
                    );

                    // Convert to relative paths for notification matching
                    let relative_paths: Vec<String> = canonical
                        .iter()
                        .filter_map(|p| {
                            p.strip_prefix(&project_root)
                                .ok()
                                .map(|rel| rel.to_string_lossy().to_string())
                        })
                        .collect();

                    // Publish general file.changed event via NATS (all files, not just ack-required)
                    #[cfg(feature = "nats")]
                    {
                        if !relative_paths.is_empty() {
                            if let Some(ref nats) = deps.nats {
                                if let Err(e) = nats.publish_file_changed(&project_name, &relative_paths).await {
                                    warn!(error = %e, "NATS file.changed publish failed");
                                }
                            }
                        }
                    }

                    // Push notifications — one per event type, actors need to ack
                    if !relative_paths.is_empty() {
                        let notification_ids = deps.notifications.push(
                            &project_name,
                            relative_paths.clone(),
                            None, // source_actor unknown from fs event
                        );

                        if !notification_ids.is_empty() {
                            let actor_tokens = deps.actors.tokens_for_project(&project_name);

                            // Increment pending count on each actor for each notification
                            for token in &actor_tokens {
                                for _ in &notification_ids {
                                    deps.actors.increment_pending(token);
                                }
                            }

                            info!(
                                notifications = notification_ids.len(),
                                project = %project_name,
                                actors = actor_tokens.len(),
                                "notifications pushed"
                            );

                            // Publish via NATS — derive events from the file paths directly
                            #[cfg(feature = "nats")]
                            {
                                if let Some(ref nats) = deps.nats {
                                    let mut published_events = std::collections::HashSet::new();
                                    for file in &relative_paths {
                                        if let Some(event) = crate::notifications::NotificationEvent::from_path(file) {
                                            if published_events.insert(event) {
                                                let event_files: Vec<String> = relative_paths
                                                    .iter()
                                                    .filter(|f| crate::notifications::NotificationEvent::from_path(f) == Some(event))
                                                    .cloned()
                                                    .collect();
                                                if let Err(e) = nats.publish_notification(
                                                    &project_name,
                                                    &event,
                                                    &event_files,
                                                ).await {
                                                    warn!(error = %e, "NATS notification publish failed");
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }

                    // Re-seed DB if ontology files changed.
                    // The seed_lock semaphore (permits=1) prevents concurrent re-seeds
                    // caused by rapid successive debounce windows on the same project.
                    #[cfg(feature = "db")]
                    {
                        let ontology_changed = canonical.iter().any(|p| {
                            p.components().any(|c| {
                                c.as_os_str() == std::ffi::OsStr::new(".ontology")
                            })
                        });
                        if ontology_changed {
                            if let Some(ref db) = deps.db {
                                let _permit = deps.seed_lock.acquire().await;
                                info!("re-seeding ontology tables from changed files");
                                crate::seed::seed_ontology(
                                    db,
                                    &project_name,
                                    &project_root,
                                    &cache,
                                    deps.import_path.as_deref(),
                                )
                                .await;
                                deps.ontology_version
                                    .fetch_add(1, std::sync::atomic::Ordering::Release);
                                // _permit drops here, releasing the semaphore
                            }
                        }
                    }
                }
            },
            _ = full_tick.tick() => {
                // Periodic full invalidation as safety net against missed events.
                let before = cache.len();
                cache.invalidate_all();
                if before > 0 {
                    info!(evicted = before, "periodic full cache invalidation");
                }
            }
        }
    }
}

// ── Config Hot-Reload
// ─────────────────────────────────────────────────────────

/// Watches `keys-overlay.json` and hot-applies credential changes to all
/// registered `ProjectContext`s and the primary project's key set.
///
/// The parent directory is watched (not the file itself) so atomic editor
/// saves — which create a new inode and rename it — are detected reliably.
pub struct ConfigWatcher {
    _watcher: RecommendedWatcher,
    _task: tokio::task::JoinHandle<()>,
}

impl ConfigWatcher {
    /// Start watching `overlay_path` for changes.
    ///
    /// On any write, re-reads the file and applies keys for every slug.
    /// The reserved slug `"_primary"` resolves to `registry.primary_slug()`
    /// so callers do not need to know the actual project name.
    pub fn start(
        overlay_path: PathBuf,
        registry: Arc<ProjectRegistry>,
    ) -> std::result::Result<Self, crate::error::DaemonError> {
        let (tx, rx) = mpsc::channel::<()>(8);

        let tx_notify = tx.clone();
        let mut watcher = RecommendedWatcher::new(
            move |res: std::result::Result<Event, notify::Error>| {
                if res.is_ok() {
                    // Any event in the parent directory triggers a debounced reload.
                    // The reload task re-reads the file and skips if unchanged.
                    let _ = tx_notify.try_send(());
                }
            },
            Config::default(),
        )
        .map_err(|e| crate::error::DaemonError::Watcher(e.to_string()))?;

        if let Some(parent) = overlay_path.parent() {
            if parent.is_dir() {
                watcher
                    .watch(parent, RecursiveMode::NonRecursive)
                    .map_err(|e| crate::error::DaemonError::Watcher(e.to_string()))?;
                info!(path = %overlay_path.display(), "config watcher started");
            }
        }

        let task = tokio::spawn(config_reload_loop(rx, overlay_path, registry));

        Ok(Self {
            _watcher: watcher,
            _task: task,
        })
    }
}

async fn config_reload_loop(
    mut rx: mpsc::Receiver<()>,
    overlay_path: PathBuf,
    registry: Arc<ProjectRegistry>,
) {
    let debounce = Duration::from_millis(500);

    while rx.recv().await.is_some() {
        // Drain burst within the debounce window.
        tokio::time::sleep(debounce).await;
        while rx.try_recv().is_ok() {}

        if !overlay_path.exists() {
            continue;
        }

        let data = match std::fs::read_to_string(&overlay_path) {
            Ok(d) => d,
            Err(e) => {
                warn!(path = %overlay_path.display(), error = %e, "config watcher: failed to read overlay");
                continue;
            }
        };

        let overrides = match serde_json::from_str::<std::collections::HashMap<String, Vec<KeyEntry>>>(
            &data,
        ) {
            Ok(o) => o,
            Err(e) => {
                warn!(path = %overlay_path.display(), error = %e, "config watcher: overlay is invalid JSON — skipped");
                continue;
            }
        };

        let mut updated = 0usize;

        for (slug, keys) in overrides {
            let resolved = if slug == "_primary" {
                registry.primary_slug().to_string()
            } else {
                slug
            };
            if registry.update_keys(&resolved, keys).is_some() {
                updated += 1;
            } else {
                warn!(%resolved, "config watcher: no registered project with this slug — key update skipped");
            }
        }

        info!(updated, "keys-overlay hot-reloaded");
    }
}

// ── OpsWatcher (ADR-033 `config-watcher-applies-guard`) ─────────────────
//
// File watcher for `.ontoref/config.ncl`. On change, re-exports the file via
// nickel, parses the typed `DaemonNclConfig`, and calls
// `ProjectContext::apply_ops_reload` so the guard fires before the new ops
// value swaps in.
//
// Pending migrations are discovered via shell-out to the `ore` (or
// `ontoref`) dispatcher with `migrate pending --fmt json`. When the
// dispatcher is absent the watcher proceeds with an empty pending list
// (the guard then becomes advisory) — daemon startup is not gated on
// the Nu CLI being installed.

/// Watcher for `.ontoref/config.ncl` that applies guarded ops reloads.
pub struct OpsWatcher {
    _watcher: RecommendedWatcher,
    _task: tokio::task::JoinHandle<()>,
}

impl OpsWatcher {
    /// Start watching `config_path` for changes. Each detected change
    /// triggers a re-export + guarded apply against `ctx.apply_ops_reload`.
    pub fn start(
        config_path: PathBuf,
        ctx: Arc<ProjectContext>,
        import_path: Option<String>,
    ) -> std::result::Result<Self, crate::error::DaemonError> {
        let (tx, rx) = mpsc::channel::<()>(8);

        let tx_notify = tx.clone();
        let mut watcher = RecommendedWatcher::new(
            move |res: std::result::Result<Event, notify::Error>| {
                if res.is_ok() {
                    let _ = tx_notify.try_send(());
                }
            },
            Config::default(),
        )
        .map_err(|e| crate::error::DaemonError::Watcher(e.to_string()))?;

        if let Some(parent) = config_path.parent() {
            if parent.is_dir() {
                watcher
                    .watch(parent, RecursiveMode::NonRecursive)
                    .map_err(|e| crate::error::DaemonError::Watcher(e.to_string()))?;
                info!(path = %config_path.display(), "ops config watcher started");
            }
        }

        let task = tokio::spawn(ops_reload_loop(rx, config_path, ctx, import_path));

        Ok(Self {
            _watcher: watcher,
            _task: task,
        })
    }
}

/// Discover the pending-migration ids for a project by shelling out to
/// the `ore` (or `ontoref`) dispatcher with `migrate pending --fmt json`.
/// Returns an empty `Vec` when the dispatcher binary is absent, fails to
/// run, or the output cannot be parsed — the caller treats absence as
/// "no pending" rather than a hard error so daemon startup is not gated
/// on the Nu dispatcher being installed.
fn discover_pending_migrations(project_root: &Path) -> Vec<String> {
    // Try `ore` first, then fall back to `ontoref`.
    for bin in ["ore", "ontoref"] {
        let output = std::process::Command::new(bin)
            .args(["--actor", "agent", "migrate", "pending", "--fmt", "json"])
            .env("ONTOREF_PROJECT_ROOT", project_root)
            .env("ONTOREF_ACTOR", "agent")
            .env("ONTOREF_LOG_LEVEL", "none")
            .output();
        let Ok(out) = output else { continue };
        if !out.status.success() {
            continue;
        }
        let stdout = String::from_utf8_lossy(&out.stdout);
        let trimmed = stdout.trim();
        if !trimmed.starts_with('[') {
            continue;
        }
        // Each entry has shape { id: "0023", slug: "...", ... } — we only need ids.
        let Ok(arr) = serde_json::from_str::<Vec<serde_json::Value>>(trimmed) else {
            continue;
        };
        return arr
            .into_iter()
            .filter_map(|v| v.get("id").and_then(|i| i.as_str()).map(String::from))
            .collect();
    }
    Vec::new()
}

/// Run `nickel export --format json` against `config_path` with the given
/// import path. Returns the parsed JSON or `None` on any failure (with a
/// `warn!` log emitted by the caller's diagnostic context).
fn export_config_json(config_path: &Path, import_path: Option<&str>) -> Option<serde_json::Value> {
    let mut cmd = std::process::Command::new("nickel");
    cmd.args(["export", "--format", "json"]).arg(config_path);
    if let Some(ip) = import_path {
        cmd.env("NICKEL_IMPORT_PATH", ip);
    }
    let output = cmd.output().ok()?;
    if !output.status.success() {
        debug!(
            stderr = %String::from_utf8_lossy(&output.stderr),
            "nickel export failed for ops watcher reload"
        );
        return None;
    }
    serde_json::from_slice::<serde_json::Value>(&output.stdout).ok()
}

async fn ops_reload_loop(
    mut rx: mpsc::Receiver<()>,
    config_path: PathBuf,
    ctx: Arc<ProjectContext>,
    import_path: Option<String>,
) {
    let debounce = Duration::from_millis(500);

    while rx.recv().await.is_some() {
        tokio::time::sleep(debounce).await;
        while rx.try_recv().is_ok() {}

        if !config_path.exists() {
            continue;
        }

        let json = match export_config_json(&config_path, import_path.as_deref()) {
            Some(v) => v,
            None => {
                warn!(path = %config_path.display(), "ops watcher: nickel export failed — skipped");
                continue;
            }
        };

        let ncl: DaemonNclConfig = match serde_json::from_value(json) {
            Ok(v) => v,
            Err(e) => {
                warn!(error = %e, "ops watcher: config deserialisation failed — skipped");
                continue;
            }
        };

        // Discover pending migrations by shelling out to `ore migrate pending`.
        // Best-effort: returns empty when the dispatcher is absent or fails;
        // the guard then becomes advisory but still validates non-tier edits.
        let pending = discover_pending_migrations(&ctx.root);
        let action = ctx.apply_ops_reload(ncl.ops, &pending);
        match action {
            ReloadAction::Apply(applied) => {
                info!(
                    tier = ?applied.tier,
                    phase = ?applied.phase,
                    "ops reload applied"
                );
            }
            ReloadAction::Block { reason, ids } => {
                warn!(
                    reason,
                    pending = ?ids,
                    "ops reload blocked by guard — config edit not honoured"
                );
            }
        }
    }
}
