use anyhow::Result;
use std::collections::HashMap;
use std::path::{Path, PathBuf};

use agent_doc_document::watch_projection::file_watch_event_id;
use agent_doc_document::watch_projection::{
    document_node_events_payload, project_watch_node_events,
};
use agent_doc_document_realtime::watch_authority::{
    RawWatchEvent, WatchDelivery, WatchWriteProvenance, WatcherRegistry,
};
use agent_doc_markdown_ast::events::DocumentNodeEvent;
use agent_doc_state_backbone::{StateEvent, StateFact};

pub mod daemon;

pub use daemon::{
    WatchConfig, WatchDaemonEffects, cycle_freshly_in_flight, ensure_running, start, status, stop,
};

pub const PID_FILE: &str = ".agent-doc/watch.pid";

/// A controller-watch observation after write-provenance filtering and burst
/// coalescing.
pub struct WatchObservation {
    pub delivery: WatchDelivery,
    pub state_event: Option<StateEvent>,
}

/// Refresh the cached markdown snapshot for `path` and return node-keyed events
/// against the previous projection.
pub fn update_node_snapshot(
    path: &Path,
    snapshots: &mut HashMap<PathBuf, String>,
) -> Result<Vec<DocumentNodeEvent>> {
    let content = std::fs::read_to_string(path)?;
    let previous = snapshots.insert(path.to_path_buf(), content.clone());
    Ok(project_watch_node_events(previous.as_deref(), &content))
}

/// Emit watch node-event evidence through the caller's ops-log boundary.
pub fn log_node_events(
    path: &Path,
    events: &[DocumentNodeEvent],
    mut ops_logger: impl FnMut(&Path, &str),
) {
    if events.is_empty() {
        return;
    }
    let payload = document_node_events_payload(&path.display().to_string(), events);
    let message = format!("document_node_events {payload}");
    ops_logger(path, &message);
}

/// The process-wide controller-owned watcher registry singleton.
pub fn registry() -> &'static WatcherRegistry {
    static REGISTRY: std::sync::OnceLock<WatcherRegistry> = std::sync::OnceLock::new();
    REGISTRY.get_or_init(WatcherRegistry::new)
}

/// Remove a document from the process-wide watcher registry.
pub fn unregister_document(doc_id: &str) {
    registry().unregister(doc_id);
}

/// Observe a raw filesystem event through the controller-owned watch gate.
///
/// Self-write echoes and coalesced bursts are represented in `delivery` without
/// producing a state event. A real content change returns the state event that
/// callers should persist from their controller/session boundary.
pub fn observe_document_event(
    doc_id: &str,
    file: &str,
    raw: &RawWatchEvent,
    current_content: &str,
) -> WatchObservation {
    let (gate, _new) = registry().register(doc_id, file);
    let content_hash = agent_doc_hash::content_hash(current_content);
    let provenance = agent_doc_cycle_state_io::load_document_disk_write(Path::new(file))
        .ok()
        .flatten();
    let write_provenance = provenance
        .as_ref()
        .map(|prov| WatchWriteProvenance::new(prov.actor.as_str(), prov.content_hash.as_str()));
    let delivery = {
        let mut g = gate.lock();
        g.observe(raw, &content_hash, write_provenance)
    };
    let state_event = match delivery {
        WatchDelivery::Change { generation } => Some(StateEvent::new(
            file_watch_event_id(doc_id, generation, &content_hash),
            StateFact::FileWatchChangeObserved {
                document_hash: doc_id.to_string(),
                path: file.to_string(),
                watch_generation: generation,
                content_hash,
            },
        )),
        _ => None,
    };
    WatchObservation {
        delivery,
        state_event,
    }
}

pub fn pid_path(base_dir: &Path) -> PathBuf {
    base_dir.join(PID_FILE)
}

/// The directory the no-argument pid helpers resolve `.agent-doc/watch.pid`
/// against.
///
/// `#stalewatchpid`: these used to use the raw process working directory, so a
/// `watch --stop` / `--status` invoked from a subdirectory read and removed a
/// *different* path than the daemon wrote — the removal silently did nothing
/// (`let _ =`) and the real pid file survived as a stale record. Resolving the
/// project root makes every pid operation name one file no matter where the
/// operator stands. Falls back to the working directory when the caller is
/// outside an agent-doc project, which keeps the old behavior for that case.
fn watch_pid_base_dir() -> Option<PathBuf> {
    let cwd = std::env::current_dir().ok()?;
    Some(agent_doc_project_root_io::project_root_containing(&cwd).unwrap_or(cwd))
}

/// Check whether a PID is alive using the same `/proc` probe as the watcher.
pub fn pid_alive(pid: u32) -> bool {
    Path::new(&format!("/proc/{pid}")).exists()
}

/// Whether `pid` is a live agent-doc **watch daemon** — verified by process
/// identity, not mere liveness.
///
/// A bare `/proc/{pid}` existence check is fooled by a REUSED pid: when a watch
/// daemon dies and the OS later reassigns its pid to an unrelated process (e.g. a
/// kernel thread such as `[rcu_...]`, whose `/proc/{pid}/cmdline` is empty), the
/// stale `watch.pid` still "looks alive", so `ensure_running` believes a daemon is
/// running and refuses to start a real one — and `--status` reports a phantom
/// daemon. Verify the cmdline is actually `agent-doc … watch`, mirroring the
/// supervisor's `supervisor_pid_matches_doc` and the plugin-owner lease
/// pid-identity guard.
pub fn pid_is_agent_doc_watch(pid: u32) -> bool {
    let Ok(cmdline) = std::fs::read_to_string(format!("/proc/{pid}/cmdline")) else {
        return false;
    };
    // `cmdline` is NUL-separated argv; a kernel thread yields an empty string.
    let args = cmdline.replace('\0', " ");
    args.contains("agent-doc") && args.contains("watch")
}

pub fn read_pid() -> Option<u32> {
    read_pid_in(&watch_pid_base_dir()?)
}

/// Check whether the watch daemon recorded in the current directory is alive.
/// Identity-verified (`pid_is_agent_doc_watch`) so a stale pid reused by an
/// unrelated process does not read as a live daemon.
pub fn is_running() -> bool {
    read_pid().is_some_and(pid_is_agent_doc_watch)
}

pub fn read_pid_in(base_dir: &Path) -> Option<u32> {
    let content = std::fs::read_to_string(pid_path(base_dir)).ok()?;
    content.trim().parse().ok()
}

/// `#stalewatchpid`: delete a pid file that provably does not describe a live
/// watch daemon, returning the pruned pid.
///
/// A daemon that exits cleanly removes its own pid file, but `SIGKILL`, an OOM
/// kill, and a reboot cannot run cleanup — so the file must also be reclaimable
/// from the read side. Without this the record is immortal: the observed
/// haiven-dev file named a pid dead for three days. Identity-verified, never a
/// bare liveness check, so this cannot delete a live daemon's file after pid
/// reuse.
pub fn clear_stale_pid_in(base_dir: &Path) -> Result<Option<u32>> {
    let Some(pid) = read_pid_in(base_dir) else {
        return Ok(None);
    };
    if pid_is_agent_doc_watch(pid) {
        return Ok(None);
    }
    remove_pid_in(base_dir)?;
    Ok(Some(pid))
}

/// [`clear_stale_pid_in`] against the resolved project root.
pub fn clear_stale_pid() -> Result<Option<u32>> {
    match watch_pid_base_dir() {
        Some(base_dir) => clear_stale_pid_in(&base_dir),
        None => Ok(None),
    }
}

pub fn write_current_pid() -> Result<()> {
    let base_dir = watch_pid_base_dir()
        .ok_or_else(|| anyhow::anyhow!("could not resolve a base directory for watch.pid"))?;
    write_current_pid_in(&base_dir)
}

pub fn write_current_pid_in(base_dir: &Path) -> Result<()> {
    let pid_path = pid_path(base_dir);
    if let Some(parent) = pid_path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    std::fs::write(pid_path, std::process::id().to_string())?;
    Ok(())
}

/// Remove the watch pid file, resolved against the project root.
///
/// `#stalewatchpid`: this used to remove the bare relative `PID_FILE` and
/// discard the result. Two failure modes followed — a caller whose working
/// directory was not the project root removed nothing, and *every* failure was
/// invisible, so a pid file that could not be removed still read as a live
/// daemon forever. A missing file is success; anything else is reported.
pub fn remove_pid() -> Result<()> {
    match watch_pid_base_dir() {
        Some(base_dir) => remove_pid_in(&base_dir),
        None => Ok(()),
    }
}

pub fn remove_pid_in(base_dir: &Path) -> Result<()> {
    let path = pid_path(base_dir);
    match std::fs::remove_file(&path) {
        Ok(()) => Ok(()),
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(err) => {
            Err(anyhow::Error::from(err).context(format!("failed to remove {}", path.display())))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use agent_doc_document_realtime::watch_authority::RawWatchEvent;

    #[test]
    fn watch_liveness_verifies_identity_not_mere_pid_existence() {
        // The stale-pid bug: a pid reused by an unrelated live process (e.g. a
        // kernel thread with an empty cmdline) must NOT read as a live watch
        // daemon. pid 1 (init/systemd) is always alive but is not `agent-doc
        // watch`; a bare liveness check would wrongly call it running.
        assert!(
            !pid_is_agent_doc_watch(1),
            "pid 1 is live but is not an agent-doc watch daemon"
        );
        // A pid with no /proc entry (nonexistent) is not a daemon either.
        assert!(!pid_is_agent_doc_watch(u32::MAX - 1));
    }

    #[test]
    fn pid_file_roundtrip() {
        let dir = tempfile::TempDir::new().unwrap();
        std::fs::create_dir_all(dir.path().join(".agent-doc")).unwrap();

        write_current_pid_in(dir.path()).unwrap();
        let pid = read_pid_in(dir.path()).unwrap();
        assert_eq!(pid, std::process::id());

        remove_pid_in(dir.path()).unwrap();
        assert!(read_pid_in(dir.path()).is_none());
    }

    /// `#stalewatchpid` cause 4: a daemon killed by `SIGKILL` or lost to a
    /// reboot cannot run its own cleanup, so the read side must reclaim the
    /// record. Without this the file is immortal — the observed haiven-dev one
    /// named a pid dead for three days.
    #[test]
    fn clear_stale_pid_reclaims_a_dead_daemon_record() {
        let dir = tempfile::TempDir::new().unwrap();
        std::fs::create_dir_all(dir.path().join(".agent-doc")).unwrap();
        std::fs::write(pid_path(dir.path()), "4294967294").unwrap();

        let pruned = clear_stale_pid_in(dir.path()).unwrap();

        assert_eq!(pruned, Some(4_294_967_294));
        assert!(
            read_pid_in(dir.path()).is_none(),
            "a pid that is not a live watch daemon must not survive as a record"
        );
    }

    /// The prune is identity-verified, so it must not delete the record of a
    /// process that is merely alive under a reused pid — nor, symmetrically,
    /// keep one that is.
    #[test]
    fn clear_stale_pid_keeps_a_live_daemon_record() {
        let dir = tempfile::TempDir::new().unwrap();
        std::fs::create_dir_all(dir.path().join(".agent-doc")).unwrap();
        // pid 1 is always alive but is not `agent-doc watch`, which is exactly
        // the reused-pid shape. It must still be pruned...
        std::fs::write(pid_path(dir.path()), "1").unwrap();
        assert_eq!(clear_stale_pid_in(dir.path()).unwrap(), Some(1));

        // ...while a pid that IS a live agent-doc watch daemon is retained.
        // This test binary's own cmdline contains both tokens.
        let self_pid = std::process::id();
        if pid_is_agent_doc_watch(self_pid) {
            std::fs::write(pid_path(dir.path()), self_pid.to_string()).unwrap();
            assert_eq!(clear_stale_pid_in(dir.path()).unwrap(), None);
            assert_eq!(read_pid_in(dir.path()), Some(self_pid));
        }
    }

    /// `#stalewatchpid` cause 2: removal used the bare relative `PID_FILE` and
    /// discarded its result, so a caller standing anywhere but the project root
    /// removed nothing and reported success anyway.
    #[test]
    fn remove_pid_in_is_path_scoped_and_reports() {
        let dir = tempfile::TempDir::new().unwrap();
        let other = tempfile::TempDir::new().unwrap();
        std::fs::create_dir_all(dir.path().join(".agent-doc")).unwrap();
        std::fs::create_dir_all(other.path().join(".agent-doc")).unwrap();
        std::fs::write(pid_path(dir.path()), "12345").unwrap();

        // Removing against an unrelated root must not touch this file.
        remove_pid_in(other.path()).unwrap();
        assert_eq!(read_pid_in(dir.path()), Some(12345));

        remove_pid_in(dir.path()).unwrap();
        assert!(read_pid_in(dir.path()).is_none());

        // A already-absent file is success, not an error — cleanup is idempotent.
        remove_pid_in(dir.path()).unwrap();
    }

    #[test]
    fn pid_alive_self() {
        assert!(pid_alive(std::process::id()));
    }

    #[test]
    fn pid_alive_nonexistent() {
        assert!(!pid_alive(4_294_967_295));
    }

    #[test]
    fn observe_document_event_builds_state_event_for_real_change() {
        let dir = tempfile::TempDir::new().unwrap();
        let file = dir.path().join("doc.md");
        std::fs::write(&file, "one").unwrap();
        let file_str = file.to_string_lossy().to_string();
        let doc_id = "watch-io-event";
        let raw = RawWatchEvent::modify(&file);

        let first = observe_document_event(doc_id, &file_str, &raw, "one");
        assert_eq!(first.delivery, WatchDelivery::Change { generation: 1 });
        let event = first
            .state_event
            .expect("change should produce state event");
        assert!(event.event_id.contains(doc_id));

        let second = observe_document_event(doc_id, &file_str, &raw, "one");
        assert_eq!(second.delivery, WatchDelivery::Coalesced);
        assert!(
            second.state_event.is_none(),
            "coalesced events should not produce ledger events"
        );

        unregister_document(doc_id);
    }

    #[test]
    fn update_node_snapshot_emits_node_keyed_events_after_seed() {
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("session.md");
        std::fs::write(
            &path,
            "<!-- agent:queue -->\n- do [#alpha]\n<!-- /agent:queue -->\n",
        )
        .unwrap();
        let mut snapshots = HashMap::new();

        assert!(
            update_node_snapshot(&path, &mut snapshots)
                .unwrap()
                .is_empty()
        );

        std::fs::write(
            &path,
            "<!-- agent:queue -->\n- do [#alpha]\n- do [#beta]\n<!-- /agent:queue -->\n",
        )
        .unwrap();
        let events = update_node_snapshot(&path, &mut snapshots).unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(
            events[0].kind,
            agent_doc_markdown_ast::events::DocumentNodeEventKind::Insert
        );
        assert!(events[0].node_key.contains(":beta:"));
        assert!(
            events[0]
                .previous_node_key
                .as_deref()
                .is_some_and(|key| key.contains(":alpha:"))
        );
    }

    #[test]
    fn log_node_events_emits_document_node_events_payload() {
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("session.md");
        let before =
            "<!-- agent:backlog -->\n1. [ ] [#task] old wording\n<!-- /agent:backlog -->\n";
        let after = "<!-- agent:backlog -->\n1. [ ] [#task] new wording\n<!-- /agent:backlog -->\n";
        std::fs::write(&path, after).unwrap();
        let events = agent_doc_markdown_ast::events::diff_node_events(before, after);
        let mut logged = Vec::new();

        log_node_events(&path, &events, |file, message| {
            logged.push((file.to_path_buf(), message.to_string()));
        });

        assert_eq!(logged.len(), 1);
        assert_eq!(logged[0].0, path);
        let message = &logged[0].1;
        assert!(message.contains("document_node_events"), "{message}");
        assert!(
            message.contains("\"event\":\"document_node_events\""),
            "{message}"
        );
        assert!(message.contains("\"component\":\"backlog\""), "{message}");
        assert!(
            message.contains("\"node_key\":\"backlog:0:task:0\""),
            "{message}"
        );
        assert!(message.contains("\"op\":\"replace\""), "{message}");
    }
}
