//! # History store (spec §1.3, §4)
//!
//! 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.

use std::path::Path;

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

/// A handle to the SQLite history database.
pub struct History {
    conn: Connection,
}

impl History {
    /// Open (creating if needed) the history database at `path` in WAL mode.
    pub fn open(path: &Path) -> rusqlite::Result<History> {
        let conn = Connection::open(path)?;
        // WAL: concurrent reads while the daemon writes; survives restarts.
        conn.pragma_update(None, "journal_mode", "WAL")?;
        Self::init(conn)
    }

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

    fn init(conn: Connection) -> 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)",
            [],
        )?;
        Ok(History { conn })
    }

    /// Append one signal to the history.
    pub fn record(&self, s: &Signal) -> rusqlite::Result<()> {
        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 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 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
    }
}