//! # History store
//!
//! A local SQLite (WAL) store of a rolling history of signals. History is what
//! lets a renderer read *recent aggregates* rather than only a live snapshot —
//! it is what lets the garden survive a restart and (later) the aging pet know
//! its true age.
//!
//! ## Same privacy constraints as the wire
//!
//! The store holds **aggregate scalars only**. Each row carries the schema
//! discriminants, the `f64` `value`, and the audited non-content `tag` — exactly
//! the fields of a [`Signal`], nothing wider. There is no content column. As a
//! belt-and-braces measure the audited wire frame is also stored as a BLOB and
//! is the source of truth on read, so reconstruction reuses the same validated
//! [`wire::decode`] the socket uses — a row can decode to nothing but a
//! well-formed, `f64`-only signal. The differential secret-typing test writes a
//! planted secret through the collector into this store and asserts it never
//! appears here in any encoding.
//!
//! ## Retention
//!
//! The store is rolling: rows older than the retention are deleted on open and
//! every [`PRUNE_EVERY`] records, so the file stays bounded for a daemon that
//! runs for months.

use std::path::Path;
use std::time::Duration;

use rusqlite::{params, Connection, OpenFlags};
use signal_schema::{wire, Signal};

use crate::now_millis;

/// Retention for [`History::open`].
pub const DEFAULT_RETENTION: Duration = Duration::from_secs(7 * 86_400);

/// [`History::record`] prunes after this many inserts.
pub const PRUNE_EVERY: u64 = 1000;

/// A handle to the SQLite history database.
pub struct History {
    conn: Connection,
    /// `None` for an in-memory store: nothing is ever pruned.
    retention_ms: Option<u64>,
    since_prune: u64,
}

impl History {
    /// Open (creating if needed) the history database at `path` in WAL mode
    /// with [`DEFAULT_RETENTION`].
    pub fn open(path: &Path) -> rusqlite::Result<History> {
        Self::open_with_retention(path, DEFAULT_RETENTION)
    }

    /// Open with an explicit retention. Rows older than it are pruned now and
    /// periodically as records arrive.
    pub fn open_with_retention(path: &Path, retention: Duration) -> rusqlite::Result<History> {
        let conn = Connection::open(path)?;
        // WAL: concurrent reads while the daemon writes; survives restarts.
        conn.pragma_update(None, "journal_mode", "WAL")?;
        let h = Self::init(conn, Some(retention.as_millis() as u64))?;
        h.drop_foreign_schema_rows()?;
        h.prune()?;
        Ok(h)
    }

    /// Open an existing database without writing to it.
    ///
    /// [`History::open`] prunes, creates and migrates. A reader that polls —
    /// the menu-bar pet asks every few seconds when activity last happened —
    /// must not mutate the daemon's store as a side effect of looking at it.
    /// Returns an error if the database does not exist yet.
    pub fn open_read_only(path: &Path) -> rusqlite::Result<History> {
        let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
        Ok(History {
            conn,
            retention_ms: None,
            since_prune: 0,
        })
    }

    /// When activity was last seen, as unix millis, or `None` if there is
    /// none on record.
    ///
    /// Activity is a keystroke rate or a commit-count above zero. Deliberately
    /// not `SessionSeconds`: a shell left open is not attention, and counting
    /// it would let a forgotten terminal stand in for a person indefinitely.
    /// Deliberately not the hardware signals either — a sleeping machine still
    /// reports a battery percentage.
    pub fn last_activity_ms(&self) -> rusqlite::Result<Option<u64>> {
        let ts: Option<i64> = self.conn.query_row(
            "SELECT MAX(ts) FROM signals WHERE name IN (?1, ?2) AND value > 0.0",
            params![
                signal_schema::SignalName::KeysPerMin.to_u8(),
                signal_schema::SignalName::CommitsToday.to_u8(),
            ],
            |row| row.get(0),
        )?;
        Ok(ts.map(|t| t as u64))
    }

    /// An ephemeral in-memory store (used by tests). No retention.
    pub fn open_in_memory() -> rusqlite::Result<History> {
        Self::init(Connection::open_in_memory()?, None)
    }

    fn init(conn: Connection, retention_ms: Option<u64>) -> rusqlite::Result<History> {
        // Columns for querying (name, ts, value, tag); `frame` is the audited
        // wire encoding, reused for faithful, validated reconstruction on read.
        conn.execute(
            "CREATE TABLE IF NOT EXISTS signals (
                id             INTEGER PRIMARY KEY AUTOINCREMENT,
                schema_version INTEGER NOT NULL,
                ts             INTEGER NOT NULL,
                source         INTEGER NOT NULL,
                name           INTEGER NOT NULL,
                value          REAL    NOT NULL,
                tag            TEXT,
                frame          BLOB    NOT NULL
            )",
            [],
        )?;
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_signals_name_ts ON signals(name, ts)",
            [],
        )?;
        // For the retention delete.
        conn.execute("CREATE INDEX IF NOT EXISTS idx_signals_ts ON signals(ts)", [])?;
        Ok(History {
            conn,
            retention_ms,
            since_prune: 0,
        })
    }

    /// Delete rows written by a different schema version. Returns the number
    /// deleted.
    ///
    /// `query` rebuilds each row from its stored `frame`, and `wire::decode`
    /// refuses a frame whose version is not [`signal_schema::SCHEMA_VERSION`].
    /// Rows from an older daemon would therefore linger, unreadable and
    /// silently skipped, until retention caught up with them. Dropping them on
    /// open keeps the store to rows it can actually return.
    pub fn drop_foreign_schema_rows(&self) -> rusqlite::Result<usize> {
        self.conn.execute(
            "DELETE FROM signals WHERE schema_version != ?1",
            params![signal_schema::SCHEMA_VERSION],
        )
    }

    /// Delete rows older than the retention. Returns the number deleted.
    pub fn prune(&self) -> rusqlite::Result<usize> {
        let Some(retention_ms) = self.retention_ms else {
            return Ok(0);
        };
        let cutoff = now_millis().saturating_sub(retention_ms) as i64;
        self.conn
            .execute("DELETE FROM signals WHERE ts < ?1", params![cutoff])
    }

    /// Append one signal to the history, pruning every [`PRUNE_EVERY`] calls.
    pub fn record(&mut self, s: &Signal) -> rusqlite::Result<()> {
        self.since_prune += 1;
        if self.since_prune >= PRUNE_EVERY {
            self.since_prune = 0;
            self.prune()?;
        }
        self.conn.execute(
            "INSERT INTO signals (schema_version, ts, source, name, value, tag, frame)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
            params![
                s.schema_version,
                s.ts as i64, // SQLite integers are i64; unix-millis fits
                s.source.to_u8(),
                s.name.to_u8(),
                s.value.0,
                s.tag.as_ref().map(|t| t.as_str()),
                wire::encode(s),
            ],
        )?;
        Ok(())
    }

    /// The most recent `limit` signals, newest first. This is the query that
    /// lets a renderer read recent history instead of only the live snapshot.
    pub fn recent(&self, limit: usize) -> rusqlite::Result<Vec<Signal>> {
        self.query("SELECT frame FROM signals ORDER BY id DESC LIMIT ?1", limit)
    }

    /// The most recent `limit` signals for one metric name, newest first.
    pub fn recent_named(
        &self,
        name: signal_schema::SignalName,
        limit: usize,
    ) -> rusqlite::Result<Vec<Signal>> {
        let mut stmt = self.conn.prepare(
            "SELECT frame FROM signals WHERE name = ?1 ORDER BY id DESC LIMIT ?2",
        )?;
        let rows = stmt.query_map(params![name.to_u8(), limit as i64], |row| {
            row.get::<_, Vec<u8>>(0)
        })?;
        Self::decode_rows(rows)
    }

    fn query(&self, sql: &str, limit: usize) -> rusqlite::Result<Vec<Signal>> {
        let mut stmt = self.conn.prepare(sql)?;
        let rows = stmt.query_map(params![limit as i64], |row| row.get::<_, Vec<u8>>(0))?;
        Self::decode_rows(rows)
    }

    fn decode_rows(
        rows: impl Iterator<Item = rusqlite::Result<Vec<u8>>>,
    ) -> rusqlite::Result<Vec<Signal>> {
        let mut out = Vec::new();
        for frame in rows {
            let frame = frame?;
            if let Some(sig) = wire::decode(&frame) {
                out.push(sig);
            }
        }
        Ok(out)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use signal_schema::{SignalName, Source, Tag, Value, SCHEMA_VERSION};

    fn sig(name: SignalName, value: f64, tag: Option<Tag>) -> Signal {
        Signal {
            schema_version: SCHEMA_VERSION,
            ts: 1_723_100_000_000,
            source: Source::Git,
            name,
            value: Value(value),
            tag,
        }
    }

    #[test]
    fn round_trip_write_then_query_recent() {
        let mut h = History::open_in_memory().unwrap();
        let repo = Tag::repo_path("/x/repo").unwrap();
        h.record(&sig(SignalName::CommitsWindow, 5.0, Some(repo.clone())))
            .unwrap();
        h.record(&sig(SignalName::CommitsToday, 2.0, Some(repo.clone())))
            .unwrap();
        h.record(&sig(SignalName::KeysPerMin, 88.0, None)).unwrap();

        let recent = h.recent(10).unwrap();
        assert_eq!(recent.len(), 3);
        // Newest first, and the aggregates survive the round trip exactly.
        assert_eq!(recent[0].name, SignalName::KeysPerMin);
        assert_eq!(recent[0].value, Value(88.0));
        assert_eq!(recent[2].name, SignalName::CommitsWindow);
        assert_eq!(recent[2].value, Value(5.0));
        assert_eq!(recent[2].tag.as_ref().unwrap().as_str(), "/x/repo");
    }

    #[test]
    fn recent_named_filters_by_metric() {
        let mut h = History::open_in_memory().unwrap();
        for v in [1.0, 2.0, 3.0] {
            h.record(&sig(SignalName::KeysPerMin, v, None)).unwrap();
        }
        h.record(&sig(SignalName::SessionSeconds, 42.0, None))
            .unwrap();

        let keys = h.recent_named(SignalName::KeysPerMin, 10).unwrap();
        assert_eq!(keys.len(), 3);
        assert!(keys.iter().all(|s| s.name == SignalName::KeysPerMin));
        assert_eq!(keys[0].value, Value(3.0)); // newest first
    }

    #[test]
    fn retention_prunes_old_rows_on_open_and_periodically() {
        let dir = std::env::temp_dir().join(format!("signald-history-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let db = dir.join("h.sqlite");
        let retention = Duration::from_secs(3600);
        let ancient = Signal {
            ts: 1,
            ..sig(SignalName::KeysPerMin, 1.0, None)
        };
        let fresh = Signal {
            ts: now_millis(),
            ..sig(SignalName::KeysPerMin, 2.0, None)
        };
        {
            let mut h = History::open_with_retention(&db, retention).unwrap();
            h.record(&ancient).unwrap();
            h.record(&fresh).unwrap();
            assert_eq!(h.recent(10).unwrap().len(), 2);
            assert_eq!(h.prune().unwrap(), 1);
            assert_eq!(h.recent(10).unwrap().len(), 1);
            // Left behind for the reopen to prune.
            h.record(&ancient).unwrap();
        }
        let h = History::open_with_retention(&db, retention).unwrap();
        let rows = h.recent(10).unwrap();
        assert_eq!(rows.len(), 1, "reopen prunes");
        assert_eq!(rows[0].value, Value(2.0));
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Activity is what the menu-bar pet lives on, so what counts is exact.
    #[test]
    fn last_activity_is_typing_and_commits_only() {
        let mut h = History::open_in_memory().unwrap();
        // A shell sitting open, and a machine reporting hardware while asleep.
        // Neither is a person showing up.
        for (name, value, ts) in [
            (SignalName::SessionSeconds, 9000.0, 500_u64),
            (SignalName::BatteryPct, 80.0, 600),
            (SignalName::CpuLoad, 0.4, 700),
        ] {
            h.record(&Signal { ts, ..sig(name, value, None) }).unwrap();
        }
        assert_eq!(h.last_activity_ms().unwrap(), None, "none of that is attention");

        h.record(&Signal { ts: 1_000, ..sig(SignalName::KeysPerMin, 30.0, None) })
            .unwrap();
        assert_eq!(h.last_activity_ms().unwrap(), Some(1_000));

        h.record(&Signal { ts: 2_000, ..sig(SignalName::CommitsToday, 3.0, None) })
            .unwrap();
        assert_eq!(h.last_activity_ms().unwrap(), Some(2_000), "the most recent wins");
    }

    /// A quiet day publishes zeroes rather than nothing, so a zero must not
    /// read as activity or the pet would never age.
    #[test]
    fn a_zero_is_not_activity() {
        let mut h = History::open_in_memory().unwrap();
        h.record(&Signal { ts: 1_000, ..sig(SignalName::KeysPerMin, 0.0, None) })
            .unwrap();
        h.record(&Signal { ts: 1_100, ..sig(SignalName::CommitsToday, 0.0, None) })
            .unwrap();
        assert_eq!(h.last_activity_ms().unwrap(), None);
    }

    /// The reason this exists: `open` prunes on the way in. A pet polling every
    /// five seconds must not be quietly deleting the daemon's history.
    #[test]
    fn read_only_open_does_not_prune() {
        let dir = std::env::temp_dir().join(format!("signald-ro-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let db = dir.join("history.db");

        {
            // Retention of an hour, and a row far older than that.
            let mut h =
                History::open_with_retention(&db, std::time::Duration::from_secs(3600)).unwrap();
            h.record(&Signal { ts: 1, ..sig(SignalName::KeysPerMin, 5.0, None) })
                .unwrap();
        }

        let ro = History::open_read_only(&db).unwrap();
        assert_eq!(ro.last_activity_ms().unwrap(), Some(1), "the old row is readable");
        drop(ro);

        // Still there: a read-only open left it alone. A normal open would have
        // pruned it on the way in.
        let ro = History::open_read_only(&db).unwrap();
        assert_eq!(
            ro.conn
                .query_row("SELECT COUNT(*) FROM signals", [], |r| r.get::<_, i64>(0))
                .unwrap(),
            1,
            "read-only open must not have deleted anything"
        );
        drop(ro);

        // And prove the contrast: a normal open does prune it.
        let h = History::open_with_retention(&db, std::time::Duration::from_secs(3600)).unwrap();
        assert_eq!(h.recent(10).unwrap().len(), 0, "a writing open prunes");
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn read_only_open_refuses_a_database_that_is_not_there() {
        let missing = std::env::temp_dir().join(format!("signald-absent-{}.db", std::process::id()));
        let _ = std::fs::remove_file(&missing);
        assert!(
            History::open_read_only(&missing).is_err(),
            "read-only must not conjure a database"
        );
    }

    /// A row written under an older schema cannot be rebuilt from its frame, so
    /// reopening drops it rather than leaving it to be skipped silently.
    #[test]
    fn reopen_drops_rows_from_a_foreign_schema_version() {
        let dir = std::env::temp_dir().join(format!("signald-schema-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let db = dir.join("history.db");

        let old_version = signal_schema::SCHEMA_VERSION - 1;
        {
            let h = History::open(&db).unwrap();
            // A row exactly as an older daemon left it: the version in the
            // column and in the stored frame both predate this build.
            let s = sig(SignalName::KeysPerMin, 1.0, None);
            let mut frame = wire::encode(&s);
            frame[4..6].copy_from_slice(&old_version.to_le_bytes());
            h.conn
                .execute(
                    "INSERT INTO signals (schema_version, ts, source, name, value, tag, frame)
                     VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6)",
                    params![old_version, s.ts as i64, s.source.to_u8(), s.name.to_u8(), s.value.0, frame],
                )
                .unwrap();
            assert_eq!(h.recent(10).unwrap().len(), 0, "unreadable while present");
        }

        let h = History::open(&db).unwrap();
        assert_eq!(
            h.conn
                .query_row("SELECT COUNT(*) FROM signals", [], |r| r.get::<_, i64>(0))
                .unwrap(),
            0,
            "the foreign-schema row is gone, not merely unreadable"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }
}