//! # 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`).
//! v0.4 adds the **hardware path**: the daemon spawns the sibling
//! `macos-collector` and ingests its wire frames ([`collectors::hardware`]).

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> <session_id>`,
    /// numbers only — to a spool file on each `precmd`. Every shell appends to
    /// the same spool; the session id (the shell's pid) keeps them apart.
    ///
    /// [`Collector`](terminal::Collector) consumes the spool each tick (renames
    /// it aside, parses it, deletes it, so it never regrows) and keeps the last
    /// two flushes of each session in memory. It derives
    /// [`SignalName::KeysPerMin`](signal_schema::SignalName::KeysPerMin) as the
    /// sum of each active session's rate and
    /// [`SignalName::SessionSeconds`](signal_schema::SignalName::SessionSeconds)
    /// as the longest active session. A line that is not four 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::collections::BTreeMap;
        use std::path::{Path, PathBuf};

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

        use super::super::now_millis;

        /// A session with no flush inside this window is dropped from the
        /// aggregates (its shell is idle or gone).
        pub const ACTIVE_WINDOW_MS: u64 = 5 * 60_000;

        /// One aggregate flush parsed from the spool: a timestamp, counts, and
        /// the writing shell's session id. 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,
            pub session: u64,
        }

        /// Parse one spool line. Returns `None` for anything that is not exactly
        /// four 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()?;
            let session = 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,
                session,
            })
        }

        /// The previous and latest flush of one session.
        #[derive(Debug, Clone, Copy)]
        struct Session {
            prev: Option<Flush>,
            last: Flush,
        }

        impl Session {
            /// Keys/min over the latest flush interval; the raw count when only
            /// one flush (or a zero interval) is available.
            fn keys_per_min(&self) -> f64 {
                if let Some(prev) = self.prev {
                    let dt_ms = self.last.ts_ms.saturating_sub(prev.ts_ms);
                    if dt_ms > 0 {
                        return self.last.keys as f64 * 60_000.0 / dt_ms as f64;
                    }
                }
                self.last.keys as f64
            }
        }

        /// Per-session state carried across ticks, since the spool is consumed.
        #[derive(Debug, Default)]
        pub struct Collector {
            sessions: BTreeMap<u64, Session>,
        }

        impl Collector {
            pub fn new() -> Collector {
                Collector::default()
            }

            /// What the daemon does each tick: consume the spool, then derive
            /// the current aggregates. Empty if no session is active.
            pub fn collect(&mut self, spool: &Path) -> Vec<Signal> {
                self.ingest(spool);
                self.signals(now_millis())
            }

            /// Consume the spool: rename it aside (the hook opens it with `>>`
            /// per write, so later appends land in a fresh file), parse, delete.
            /// A leftover `.reading` file from an interrupted tick is read
            /// first. Returns the number of records absorbed.
            pub fn ingest(&mut self, spool: &Path) -> usize {
                let reading = reading_path(spool);
                if !reading.exists() && std::fs::rename(spool, &reading).is_err() {
                    return 0; // no spool yet
                }
                let text = std::fs::read_to_string(&reading).unwrap_or_default();
                let _ = std::fs::remove_file(&reading);
                self.absorb(text.lines().filter_map(parse_flush))
            }

            /// Fold flushes into per-session state (pure, unit-testable).
            pub fn absorb(&mut self, flushes: impl IntoIterator<Item = Flush>) -> usize {
                let mut n = 0;
                for f in flushes {
                    n += 1;
                    self.sessions
                        .entry(f.session)
                        .and_modify(|s| {
                            s.prev = Some(s.last);
                            s.last = f;
                        })
                        .or_insert(Session { prev: None, last: f });
                }
                n
            }

            /// The aggregates at `now_ms`, dropping sessions idle longer than
            /// [`ACTIVE_WINDOW_MS`]. Empty if no session is active.
            pub fn signals(&mut self, now_ms: u64) -> Vec<Signal> {
                self.sessions
                    .retain(|_, s| now_ms.saturating_sub(s.last.ts_ms) <= ACTIVE_WINDOW_MS);
                if self.sessions.is_empty() {
                    return Vec::new();
                }
                let keys_per_min: f64 = self.sessions.values().map(Session::keys_per_min).sum();
                let session_s = self.sessions.values().map(|s| s.last.session_s).max().unwrap_or(0);
                let mk = |name: SignalName, value: f64| Signal {
                    schema_version: SCHEMA_VERSION,
                    ts: now_ms,
                    source: Source::Terminal,
                    name,
                    value: Value(value),
                    tag: None, // terminal aggregates are never tagged (spec §1.2)
                };
                vec![
                    mk(SignalName::KeysPerMin, keys_per_min),
                    mk(SignalName::SessionSeconds, session_s as f64),
                ]
            }
        }

        fn reading_path(spool: &Path) -> PathBuf {
            let mut p = spool.as_os_str().to_os_string();
            p.push(".reading");
            PathBuf::from(p)
        }

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

            fn flush(ts_ms: u64, keys: u64, session_s: u64, session: u64) -> Flush {
                Flush {
                    ts_ms,
                    keys,
                    session_s,
                    session,
                }
            }

            #[test]
            fn parse_requires_exactly_four_integers() {
                assert_eq!(parse_flush("1000 12 30 4242"), Some(flush(1000, 12, 30, 4242)));
                assert_eq!(parse_flush("1000 12 30"), None);
                assert_eq!(parse_flush("1000 12 30 4242 extra"), None);
                assert_eq!(parse_flush("1000 twelve 30 4242"), None);
            }

            #[test]
            fn rates_are_per_session_then_summed() {
                let mut c = Collector::new();
                // Shell 1: 60 keys over 30 s = 120/min. Shell 2: 10 keys over
                // 60 s = 10/min. Interleaved in the spool as they would be.
                c.absorb([
                    flush(0, 0, 0, 1),
                    flush(0, 0, 0, 2),
                    flush(60_000, 10, 60, 2),
                    flush(30_000, 60, 30, 1),
                ]);
                let sigs = c.signals(60_000);
                let kpm = sigs.iter().find(|s| s.name == SignalName::KeysPerMin).unwrap();
                assert_eq!(kpm.value, Value(130.0));
                let ss = sigs.iter().find(|s| s.name == SignalName::SessionSeconds).unwrap();
                assert_eq!(ss.value, Value(60.0), "longest active session");
            }

            #[test]
            fn idle_sessions_age_out() {
                let mut c = Collector::new();
                c.absorb([flush(0, 5, 10, 1), flush(1_000, 5, 10, 1)]);
                assert_eq!(c.signals(1_000).len(), 2);
                assert!(c.signals(1_000 + ACTIVE_WINDOW_MS + 1).is_empty());
            }

            #[test]
            fn ingest_consumes_the_spool() {
                let dir = std::env::temp_dir().join(format!("signald-spool-{}", std::process::id()));
                std::fs::create_dir_all(&dir).unwrap();
                let spool = dir.join("terminal.spool");
                std::fs::write(&spool, "1000 3 1 7\nnot a record\n2000 4 2 7\n").unwrap();

                let mut c = Collector::new();
                assert_eq!(c.ingest(&spool), 2);
                assert!(!spool.exists(), "spool is consumed, not left to grow");
                assert!(!reading_path(&spool).exists());
                assert_eq!(c.ingest(&spool), 0, "nothing until the hook appends again");

                // State survives consumption: the rate uses both flushes.
                let sigs = c.signals(2_000);
                let kpm = sigs.iter().find(|s| s.name == SignalName::KeysPerMin).unwrap();
                assert_eq!(kpm.value, Value(240.0)); // 4 keys over 1 s
                let _ = std::fs::remove_dir_all(&dir);
            }
        }
    }

    /// System + hardware collector: **IOKit-only, no powermetrics, no root**
    /// (spec §1.4). Native macOS.
    ///
    /// The reads live out of process in the sibling Swift package
    /// `macos-collector/` (SwiftPM, not in this cargo workspace), which writes
    /// `signal-schema` wire frames to stdout. This module spawns it and publishes
    /// every frame it emits, so hardware signals reach the hub, the history
    /// store, and subscribers by the same path as every other collector. Frames
    /// are decoded by the same `wire::read_frame` the socket uses, so one that
    /// fails the schema's structural checks (version, tag rule) is rejected at
    /// this boundary.
    pub mod hardware {
        use std::io::{self, Read};
        use std::path::{Path, PathBuf};
        use std::process::{Child, Command, Stdio};
        use std::thread;

        use signal_schema::wire;

        use crate::hub::Hub;

        /// The collector binary's name, looked up on `PATH` when no explicit
        /// path is configured.
        pub const PROGRAM: &str = "macos-collector";

        /// Read frames from `reader` until EOF, publishing each into `hub`.
        /// Returns the number of frames published. A malformed frame ends the
        /// stream with an error.
        pub fn ingest(reader: &mut impl Read, hub: &Hub) -> io::Result<usize> {
            let mut n = 0;
            while let Some(sig) = wire::read_frame(reader)? {
                hub.publish(sig);
                n += 1;
            }
            Ok(n)
        }

        /// Spawn `program` streaming frames every `interval_ms` and ingest its
        /// stdout on a background thread. Returns once the child is running;
        /// the thread logs when the child's stream ends. The child's stdout is
        /// a pipe, so it exits on its next write after the daemon goes away.
        pub fn spawn(program: &Path, interval_ms: u64, hub: Hub) -> io::Result<Child> {
            let mut child = Command::new(program)
                .arg("--interval-ms")
                .arg(interval_ms.to_string())
                .stdin(Stdio::null())
                .stdout(Stdio::piped())
                .stderr(Stdio::inherit())
                .spawn()?;
            let mut stdout = child.stdout.take().expect("stdout is piped");
            thread::spawn(move || match ingest(&mut stdout, &hub) {
                Ok(n) => eprintln!("signald: hardware collector exited after {n} frame(s)"),
                Err(e) => eprintln!("signald: hardware collector stream error: {e}"),
            });
            Ok(child)
        }

        /// Find [`PROGRAM`] on `PATH`.
        pub fn find_on_path() -> Option<PathBuf> {
            let path = std::env::var_os("PATH")?;
            std::env::split_paths(&path)
                .map(|dir| dir.join(PROGRAM))
                .find(|p| p.is_file())
        }
    }
}

/// 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)
}