//! # signald (library)
//!
//! The collector + transport internals of the ambient-companions daemon, split
//! into a library so the collectors are unit-testable and the thin `signald`
//! binary (`src/main.rs`) just wires arguments to the runtime.
//!
//! v0.2 implements the **git path** and the **terminal path** end to end, adds a
//! **SQLite (WAL) history store** ([`history`]), and turns the socket into a
//! **live pub/sub bus with a last-value cache** ([`hub`], [`publish`]). The
//! sensitive terminal collector is aggregate-only and ships behind the now-active
//! differential secret-typing test (see `signal-schema/tests/privacy_invariant.rs`).
//! The system/hardware collector remains a stub (spec phase 3).

use std::time::{SystemTime, UNIX_EPOCH};

pub mod history;
pub mod hub;

/// Collectors: each reduces its domain to schema scalars (spec §1.1, §1.4).
pub mod collectors {
    /// Git collector: shell out to `git` for counts and ages. Counts and
    /// branch/age scalars only — never diff content. This is the cleanest
    /// signal in the suite and the first collector wired (spec §2.2, §3).
    pub mod git {
        use std::path::Path;
        use std::process::Command;

        use signal_schema::{Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};

        use super::super::now_millis;

        /// Rolling window (days) for [`SignalName::CommitsWindow`].
        pub const WINDOW_DAYS: u64 = 7;

        /// Derive the aggregate git signals for one repo. Every value is a
        /// scalar `f64`; the only string on the wire is the repo path carried
        /// in the audited `tag` (spec §1.2). Returns an empty vec if `repo` is
        /// not a git repo.
        pub fn collect(repo: &Path) -> Vec<Signal> {
            if !is_git_repo(repo) {
                return Vec::new();
            }
            let ts = now_millis();
            let now_secs = ts / 1000;
            let tag = Tag::repo_path(&repo.to_string_lossy());

            let mk = |name: SignalName, value: f64| Signal {
                schema_version: SCHEMA_VERSION,
                ts,
                source: Source::Git,
                name,
                value: Value(value),
                tag: tag.clone(),
            };

            vec![
                mk(
                    SignalName::CommitsWindow,
                    commits_since(repo, &format!("{WINDOW_DAYS} days ago")),
                ),
                mk(SignalName::CommitsToday, commits_since(repo, "midnight")),
                mk(SignalName::BranchCount, branch_count(repo)),
                mk(
                    SignalName::DaysSinceLastCommit,
                    days_since_last_commit(repo, now_secs),
                ),
            ]
        }

        fn is_git_repo(repo: &Path) -> bool {
            run_git(repo, &["rev-parse", "--is-inside-work-tree"])
                .map(|s| s == "true")
                .unwrap_or(false)
        }

        /// Count commits reachable from `HEAD` newer than `since` (a git
        /// approxidate, e.g. "midnight" or "7 days ago").
        fn commits_since(repo: &Path, since: &str) -> f64 {
            run_git(
                repo,
                &["rev-list", "--count", &format!("--since={since}"), "HEAD"],
            )
            .and_then(|s| s.parse::<f64>().ok())
            .unwrap_or(0.0)
        }

        fn branch_count(repo: &Path) -> f64 {
            run_git(repo, &["for-each-ref", "--format=%(refname)", "refs/heads/"])
                .map(|s| if s.is_empty() { 0 } else { s.lines().count() })
                .unwrap_or(0) as f64
        }

        /// Whole days between the most recent `HEAD` commit and now. `0.0` when
        /// the repo has no commits yet.
        fn days_since_last_commit(repo: &Path, now_secs: u64) -> f64 {
            match run_git(repo, &["log", "-1", "--format=%ct", "HEAD"])
                .and_then(|s| s.parse::<u64>().ok())
            {
                Some(last) => (now_secs.saturating_sub(last) / 86_400) as f64,
                None => 0.0,
            }
        }

        /// Run `git -C <repo> <args>` and return trimmed stdout on success.
        fn run_git(repo: &Path, args: &[&str]) -> Option<String> {
            let out = Command::new("git")
                .arg("-C")
                .arg(repo)
                .args(args)
                .output()
                .ok()?;
            if !out.status.success() {
                return None;
            }
            Some(String::from_utf8_lossy(&out.stdout).trim().to_string())
        }
    }

    /// Terminal collector (the sensitive one): **no input tap, no PTY sniff.**
    ///
    /// The shell side (`shell-hooks/signald-hooks.zsh`) counts keystrokes with a
    /// `zle` widget that increments a *number* and discards the key, and appends
    /// aggregate count records — `<epoch_ms> <keys> <session_seconds>`, numbers
    /// only — to a spool file on each `precmd`. This collector reads that spool
    /// and derives [`SignalName::KeysPerMin`] and [`SignalName::SessionSeconds`].
    /// It parses numbers; a line that is not three integers is dropped, so the
    /// spool can carry nothing but counts (the privacy contract, enforced by the
    /// differential secret-typing test).
    pub mod terminal {
        use std::path::Path;

        use signal_schema::{Signal, SignalName, Source, Value, SCHEMA_VERSION};

        use super::super::now_millis;

        /// One aggregate flush parsed from the spool: a timestamp and counts.
        /// Every field is a number — there is nowhere to put content.
        #[derive(Debug, Clone, Copy, PartialEq)]
        pub struct Flush {
            pub ts_ms: u64,
            pub keys: u64,
            pub session_s: u64,
        }

        /// Parse one spool line. Returns `None` for anything that is not exactly
        /// three integers, so non-count lines can never survive into a signal.
        pub fn parse_flush(line: &str) -> Option<Flush> {
            let mut it = line.split_whitespace();
            let ts_ms = it.next()?.parse().ok()?;
            let keys = it.next()?.parse().ok()?;
            let session_s = it.next()?.parse().ok()?;
            if it.next().is_some() {
                return None; // extra fields => not a well-formed count record
            }
            Some(Flush {
                ts_ms,
                keys,
                session_s,
            })
        }

        /// Read the spool and derive the current aggregate terminal signals.
        /// Returns an empty vec if the spool is missing or holds no count lines.
        pub fn collect(spool: &Path) -> Vec<Signal> {
            let text = match std::fs::read_to_string(spool) {
                Ok(t) => t,
                Err(_) => return Vec::new(),
            };
            let flushes: Vec<Flush> = text.lines().filter_map(parse_flush).collect();
            signals_from_flushes(&flushes)
        }

        /// Pure mapping from parsed flushes to schema signals (unit-testable).
        pub fn signals_from_flushes(flushes: &[Flush]) -> Vec<Signal> {
            let Some(last) = flushes.last() else {
                return Vec::new();
            };
            let ts = now_millis();
            let mk = |name: SignalName, value: f64| Signal {
                schema_version: SCHEMA_VERSION,
                ts,
                source: Source::Terminal,
                name,
                value: Value(value),
                tag: None, // terminal aggregates are never tagged (spec §1.2)
            };
            vec![
                mk(SignalName::KeysPerMin, keys_per_min(flushes)),
                mk(SignalName::SessionSeconds, last.session_s as f64),
            ]
        }

        /// Keys/min from the most recent flush interval; falls back to the raw
        /// per-flush count when only one flush (or a zero interval) is available.
        fn keys_per_min(f: &[Flush]) -> f64 {
            if f.len() >= 2 {
                let a = &f[f.len() - 2];
                let b = &f[f.len() - 1];
                let dt_ms = b.ts_ms.saturating_sub(a.ts_ms);
                if dt_ms > 0 {
                    return b.keys as f64 * 60_000.0 / dt_ms as f64;
                }
            }
            f.last().map(|x| x.keys as f64).unwrap_or(0.0)
        }
    }

    use signal_schema::Signal;

    /// System + hardware collector: **IOKit-only, no powermetrics, no root**
    /// (spec §1.4). Native macOS.
    ///
    /// v0.3: the real collector ships out-of-process as the sibling Swift package
    /// `macos-collector/` (SwiftPM, not in this cargo workspace), which reads
    /// aggregate hardware scalars via IOKit and emits `signal-schema` wire frames
    /// this daemon can parse (see `signal-schema/tests/hardware_wire.rs`). This
    /// in-process hook stays a stub: signald's live frame-ingest handshake for
    /// those frames is the next increment.
    pub fn system_hw_tick() -> Vec<Signal> {
        todo!("v0.3: hardware signals come from the sibling macos-collector Swift package")
    }
}

/// The publish side (spec §1.3): a live pub/sub fan-out with a last-value cache.
///
/// On connect a subscriber is handed the current value of every cached `name`
/// immediately (so a renderer paints correct state at once), then streams
/// updates as they change. Each subscriber runs on its own thread with its own
/// unbounded channel, so a slow subscriber never blocks the daemon (spec §1.3:
/// "a stuck wallpaper process must not stall the audio thread's feed").
pub mod publish {
    use std::io;
    use std::io::Write;
    use std::os::unix::net::{UnixListener, UnixStream};
    use std::path::Path;
    use std::thread;

    use signal_schema::wire;

    use crate::hub::Hub;

    /// Bind `socket_path` and serve the live stream from `hub` to each
    /// subscriber. Blocks, accepting connections until the listener errors.
    pub fn serve(socket_path: &Path, hub: Hub) -> io::Result<()> {
        // Clear any stale socket file from a previous run.
        let _ = std::fs::remove_file(socket_path);
        if let Some(parent) = socket_path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let listener = UnixListener::bind(socket_path)?;
        eprintln!("signald: listening on {}", socket_path.display());

        for stream in listener.incoming() {
            match stream {
                Ok(stream) => {
                    let hub = hub.clone();
                    // One subscriber per thread; a misbehaving one cannot take
                    // the daemon or the other subscribers down.
                    thread::spawn(move || {
                        if let Err(e) = serve_subscriber(stream, hub) {
                            eprintln!("signald: subscriber dropped: {e}");
                        }
                    });
                }
                Err(e) => eprintln!("signald: accept error: {e}"),
            }
        }
        Ok(())
    }

    /// Replay the last-value cache to a freshly connected subscriber, then
    /// stream live updates until it disconnects.
    fn serve_subscriber(mut stream: UnixStream, hub: Hub) -> io::Result<()> {
        let (snapshot, rx) = hub.subscribe();
        for sig in snapshot {
            wire::write_frame(&mut stream, &sig)?;
        }
        stream.flush()?;
        // `rx` yields every signal published after we subscribed. A write error
        // means the subscriber went away; returning drops `rx`, and the hub
        // prunes the dead sender on its next publish.
        for sig in rx.iter() {
            wire::write_frame(&mut stream, &sig)?;
            stream.flush()?;
        }
        Ok(())
    }
}

/// Unix millis, best-effort (spec: monotonic-corrected later).
pub fn now_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}