//! Reading a producer's own observation store.
//!
//! This is the one place magicmirror renders something other than the
//! `ops-status` contract, and the break is deliberate and kept narrow (infra
//! `7bafb5dd`). A store is opened **read-only**, queried for exactly the series
//! the config names, and never written to. Nothing here learns what a series
//! means: the config says that, because the data cannot.
//!
//! ## Why the config carries the meaning
//!
//! witchbroom's store is `(series, labels, value, at)` with an `f64`, a shape
//! its own module header defends: a new metric is a new string rather than a
//! migration, and the accepted cost is that the schema cannot say what a number
//! is or what unit it is in. So a viewer has two options. It can render every
//! series it finds, unlabelled, which is a generic table browser; or the
//! operator can name the series worth watching and say what each one is. The
//! second was ruled. A series nobody named is not shown, and that silence is
//! the choice, not an oversight.
//!
//! ## Why it is polled rather than read on draw
//!
//! `render` is a pure function of the model, the theme and the clock, and it is
//! what makes the whole surface testable without a daemon or a database. A
//! query inside it would end that. So a store is read on its own slow interval
//! into the model, exactly as a source is polled, and the tab draws what was
//! last read.

use std::path::Path;

use chrono::{DateTime, TimeZone, Utc};
use rusqlite::{Connection, OpenFlags};
use tokio::sync::mpsc;

use crate::config::{Config, Store};

/// One store's read, tagged with which store it came from.
pub(crate) struct Update {
    pub index: usize,
    pub at: DateTime<Utc>,
    pub result: Result<Vec<Reading>, String>,
}

/// The newest observation of one series under one label set.
///
/// `labels` is the store's own canonical JSON text, carried through verbatim.
/// magicmirror does not parse it: the labels are a producer's dimensions and
/// what they mean is the producer's business, so they are shown as written
/// rather than interpreted into columns this crate would have to invent.
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct Reading {
    pub series: String,
    pub labels: String,
    pub value: f64,
    pub at: DateTime<Utc>,
}

/// Spawn one reading task per configured store. Returns the receiving end.
pub(crate) fn spawn_all(stores: &[Store]) -> mpsc::Receiver<Update> {
    let (tx, rx) = mpsc::channel(stores.len().max(1) * 4);
    for (index, store) in stores.iter().enumerate() {
        tokio::spawn(read_forever(index, store.clone(), tx.clone()));
    }
    rx
}

async fn read_forever(index: usize, store: Store, tx: mpsc::Sender<Update>) {
    let mut ticker = tokio::time::interval(Config::store_interval(&store));
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        ticker.tick().await;
        // The file is opened per read rather than held. A batch producer
        // replaces its database wholesale often enough (a restore, a prune, a
        // fresh checkout) that a connection held across hours can end up
        // reading a file nobody writes to any more, which looks exactly like a
        // producer that stopped.
        let store = store.clone();
        let result = tokio::task::spawn_blocking(move || read(&store))
            .await
            .unwrap_or_else(|e| Err(format!("read task: {e}")));
        if tx
            .send(Update {
                index,
                at: Utc::now(),
                result,
            })
            .await
            .is_err()
        {
            return; // UI is gone
        }
    }
}

/// Read every configured series out of one store.
///
/// One connection, one query per series, in the order the config names them.
/// The order matters: it is the operator's own ranking of what is worth looking
/// at, and re-sorting it here would replace a decision with an alphabet.
fn read(store: &Store) -> Result<Vec<Reading>, String> {
    let conn = open(&store.path)?;
    let mut readings = Vec::new();
    for series in &store.series {
        readings.extend(latest_per_labels(&conn, &series.name).map_err(|e| short(&e))?);
    }
    Ok(readings)
}

/// Open a store read-only.
///
/// `SQLITE_OPEN_READ_ONLY` without `CREATE`, so a path with a typo in it fails
/// as "no such file" rather than quietly creating an empty database that then
/// reads as a producer which has never recorded anything. That distinction is
/// the whole difference between a broken config and a quiet night.
fn open(path: &Path) -> Result<Connection, String> {
    Connection::open_with_flags(
        path,
        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI,
    )
    .map_err(|e| short(&e))
}

/// The newest value of one series, once per distinct label set.
///
/// A bare `value`/`at` beside `MAX(at)` is SQLite's documented bare-column
/// behaviour: with a single aggregate the row the aggregate came from is the
/// row the bare columns are taken from. It is exactly the "latest per group"
/// this needs and costs no window function.
fn latest_per_labels(conn: &Connection, series: &str) -> rusqlite::Result<Vec<Reading>> {
    let mut stmt = conn.prepare(
        "SELECT labels, value, MAX(at)
           FROM observations
          WHERE series = ?1
          GROUP BY labels
          ORDER BY labels",
    )?;
    let rows = stmt.query_map([series], |row| {
        Ok(Reading {
            series: series.to_string(),
            labels: row.get(0)?,
            value: row.get(1)?,
            // Unix seconds, per the producer's schema. A timestamp outside what
            // chrono can represent is a corrupt row, not a reason to lose the
            // rest of the series, so it reads as the epoch and shows as very
            // old rather than taking the read down.
            at: Utc
                .timestamp_opt(row.get::<_, i64>(2)?, 0)
                .single()
                .unwrap_or_else(|| Utc.timestamp_nanos(0)),
        })
    })?;
    rows.collect()
}

/// A one-line reason, because the store tab has one column for it.
fn short(error: &dyn std::error::Error) -> String {
    let text = error.to_string();
    match text.split_once('\n') {
        Some((first, _)) => first.to_string(),
        None => text,
    }
}

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

    /// A store file with the producer's schema and the given observations.
    fn seeded(rows: &[(&str, &str, f64, i64)]) -> (tempfile::TempDir, std::path::PathBuf) {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("observations.db");
        let conn = Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE observations (
                 id     INTEGER PRIMARY KEY,
                 series TEXT    NOT NULL,
                 labels TEXT    NOT NULL,
                 value  REAL    NOT NULL,
                 at     INTEGER NOT NULL
             );",
        )
        .unwrap();
        for (series, labels, value, at) in rows {
            conn.execute(
                "INSERT INTO observations (series, labels, value, at) VALUES (?1, ?2, ?3, ?4)",
                rusqlite::params![series, labels, value, at],
            )
            .unwrap();
        }
        (dir, path)
    }

    fn store(path: &Path, series: &[&str]) -> Store {
        Store {
            name: "witchbroom".into(),
            path: path.to_path_buf(),
            poll_secs: 60,
            series: series
                .iter()
                .map(|s| Series {
                    name: (*s).to_string(),
                    label: format!("label for {s}"),
                    unit: Some("edges".into()),
                })
                .collect(),
        }
    }

    #[test]
    fn a_series_reads_its_newest_value_per_label_set() {
        let (_dir, path) = seeded(&[
            ("soak.coverage_edges", r#"{"repo":"a"}"#, 100.0, 1_000),
            ("soak.coverage_edges", r#"{"repo":"a"}"#, 200.0, 2_000),
            ("soak.coverage_edges", r#"{"repo":"b"}"#, 50.0, 1_500),
        ]);
        let readings = read(&store(&path, &["soak.coverage_edges"])).unwrap();
        assert_eq!(readings.len(), 2, "one row per label set, not per sample");
        assert_eq!(readings[0].labels, r#"{"repo":"a"}"#);
        assert!(
            (readings[0].value - 200.0).abs() < f64::EPSILON,
            "the newest, not the first"
        );
        assert_eq!(readings[0].at.timestamp(), 2_000);
        assert!((readings[1].value - 50.0).abs() < f64::EPSILON);
    }

    #[test]
    fn a_series_nobody_named_is_not_read() {
        // The ruling's accepted cost, asserted: the store holds it and the
        // config does not name it, so it does not exist as far as this goes.
        let (_dir, path) = seeded(&[
            ("soak.coverage_edges", "{}", 1.0, 10),
            ("cache.size_bytes", "{}", 999.0, 10),
        ]);
        let readings = read(&store(&path, &["soak.coverage_edges"])).unwrap();
        assert_eq!(readings.len(), 1);
        assert_eq!(readings[0].series, "soak.coverage_edges");
    }

    #[test]
    fn a_named_series_with_no_observations_reads_as_nothing_not_an_error() {
        let (_dir, path) = seeded(&[("cache.size_bytes", "{}", 1.0, 10)]);
        let readings = read(&store(&path, &["soak.coverage_edges"])).unwrap();
        assert!(readings.is_empty());
    }

    #[test]
    fn a_missing_store_is_a_short_reason_rather_than_a_panic() {
        let dir = tempfile::tempdir().unwrap();
        let err = read(&store(&dir.path().join("nope.db"), &["x"])).unwrap_err();
        assert!(!err.is_empty());
        assert!(!err.contains('\n'), "one line for one column: {err}");
    }

    #[test]
    fn reading_never_creates_the_file_it_could_not_find() {
        // A typo in a path must stay a visible error. Creating an empty
        // database here would turn it into a producer that reads as having
        // recorded nothing, which is a very different night.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("nope.db");
        assert!(read(&store(&path, &["x"])).is_err());
        assert!(!path.exists(), "a read must not create a store");
    }

    #[test]
    fn a_store_missing_the_producers_table_says_so_rather_than_reading_empty() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("wrong.db");
        Connection::open(&path)
            .unwrap()
            .execute_batch("CREATE TABLE something_else (x INTEGER);")
            .unwrap();
        let err = read(&store(&path, &["x"])).unwrap_err();
        assert!(err.contains("observations"), "{err}");
    }

    #[test]
    fn the_series_come_back_in_the_order_the_config_names_them() {
        // The operator's ranking, not an alphabet.
        let (_dir, path) = seeded(&[("zzz", "{}", 1.0, 10), ("aaa", "{}", 2.0, 10)]);
        let readings = read(&store(&path, &["zzz", "aaa"])).unwrap();
        let names: Vec<&str> = readings.iter().map(|r| r.series.as_str()).collect();
        assert_eq!(names, vec!["zzz", "aaa"]);
    }
}