//! Insights (oxygenhub#257): OxygenFlow's standard numbers over time, for
//! the site, an owner or a repository: how long jobs wait for a runner, how
//! many run and how they end, how long they take, how deep the queue is and
//! how busy the runners are. All from `flow_jobs` (migration 0080's queue
//! and ready times) and the runners' rows, so nothing new is collected;
//! power users point Grafana at `/metrics` for anything more.

use crate::db::{self, Pool};
use sqlx::FromRow;

/// The span a page shows.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Range {
    Hour,
    Day,
    Week,
}

impl Range {
    pub fn parse(s: &str) -> Range {
        match s {
            "1h" => Range::Hour,
            "7d" => Range::Week,
            _ => Range::Day,
        }
    }

    pub fn key(self) -> &'static str {
        match self {
            Range::Hour => "1h",
            Range::Day => "24h",
            Range::Week => "7d",
        }
    }

    pub fn ms(self) -> i64 {
        match self {
            Range::Hour => 3_600_000,
            Range::Day => 86_400_000,
            Range::Week => 7 * 86_400_000,
        }
    }

    /// How many buckets the charts have.
    pub fn buckets(self) -> usize {
        match self {
            Range::Hour => 60,
            Range::Day => 48,
            Range::Week => 56,
        }
    }

    pub fn bucket_ms(self) -> i64 {
        self.ms() / self.buckets() as i64
    }
}

/// Which jobs count.
#[derive(Clone, Debug, Default)]
pub struct Scope {
    /// An owner's repositories.
    pub owner: Option<String>,
    /// One repository.
    pub repo_id: Option<i64>,
    /// One runner's jobs.
    pub runner: Option<String>,
}

#[derive(Clone, Debug, FromRow)]
struct JobRow {
    status: String,
    runner: String,
    owner: String,
    repo: String,
    queued_ms: Option<i64>,
    ready_ms: Option<i64>,
    started_ms: Option<i64>,
    finished_ms: Option<i64>,
}

impl JobRow {
    /// Seconds from ready to started, when it started.
    fn pickup(&self) -> Option<f64> {
        Some((self.started_ms? - self.ready_ms.or(self.queued_ms)?).max(0) as f64 / 1000.0)
    }

    fn duration(&self) -> Option<f64> {
        Some((self.finished_ms? - self.started_ms?).max(0) as f64 / 1000.0)
    }

    /// When it left the queue: started, or ended without starting.
    fn left_queue(&self) -> Option<i64> {
        self.started_ms.or(self.finished_ms)
    }

    fn ended(&self) -> bool {
        !matches!(self.status.as_str(), "queued" | "running")
    }
}

/// One chart's line or stack: a value per bucket.
#[derive(Clone, Debug)]
pub struct Series {
    pub name: &'static str,
    /// The CSS class that colours it (`s-ok`, `s-bad`, …).
    pub class: &'static str,
    pub values: Vec<Option<f64>>,
}

/// One runner's row.
#[derive(Clone, Debug)]
pub struct RunnerRow {
    pub name: String,
    pub status: &'static str,
    pub running: i64,
    pub max_jobs: i64,
    pub jobs: i64,
    pub failed: i64,
    pub pickup_median: Option<f64>,
    pub duration_median: Option<f64>,
    pub last_seen_ms: Option<i64>,
}

/// One repository's row.
#[derive(Clone, Debug)]
pub struct RepoRow {
    pub slug: String,
    pub jobs: i64,
    pub failed: i64,
    pub pickup_median: Option<f64>,
    pub pickup_p95: Option<f64>,
    pub queued: i64,
}

/// Everything a page draws.
#[derive(Clone, Debug)]
pub struct Insights {
    pub range: Range,
    /// The first bucket's start; buckets follow every `range.bucket_ms()`.
    pub start_ms: i64,
    pub now_ms: i64,
    /// Seconds: median and 95th percentile per bucket, by the job's start.
    pub pickup: Vec<Series>,
    /// Jobs that ended per bucket, by result.
    pub results: Vec<Series>,
    /// Seconds: median and 95th percentile per bucket, by the job's end.
    pub durations: Vec<Series>,
    /// Jobs in the queue and running, at each bucket's end.
    pub load: Vec<Series>,
    /// Running jobs over the online runners' slots, percent, per bucket.
    pub utilisation: Vec<Series>,
    pub ended: i64,
    pub passed: i64,
    pub failed: i64,
    pub pickup_median: Option<f64>,
    pub pickup_p95: Option<f64>,
    pub duration_median: Option<f64>,
    pub queued_now: i64,
    pub running_now: i64,
    pub runners_online: i64,
    pub runners_total: i64,
    pub slots_busy: i64,
    pub slots_total: i64,
    pub runners: Vec<RunnerRow>,
    pub repos: Vec<RepoRow>,
}

/// The `p`-th percentile (0..=1) of `v`, nearest rank.
pub fn percentile(v: &mut [f64], p: f64) -> Option<f64> {
    if v.is_empty() {
        return None;
    }
    v.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
    let rank = ((p * v.len() as f64).ceil() as usize).clamp(1, v.len());
    Some(v[rank - 1])
}

async fn rows(pool: &Pool, scope: &Scope, since: i64) -> sqlx::Result<Vec<JobRow>> {
    sqlx::query_as(
        "SELECT j.status, j.runner, p.owner, p.name AS repo, COALESCE(j.queued_ms, r.created_ms) AS queued_ms, j.ready_ms,
                j.started_ms, j.finished_ms
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id
         WHERE (j.finished_ms >= $1 OR j.status IN ('queued', 'running'))
           AND ($2 = '' OR p.owner = $3) AND ($4 IS NULL OR p.id = $5) AND ($6 = '' OR j.runner = $7)",
    )
    .bind(since)
    .bind(scope.owner.as_deref().unwrap_or(""))
    .bind(scope.owner.as_deref().unwrap_or(""))
    .bind(scope.repo_id)
    .bind(scope.repo_id)
    .bind(scope.runner.as_deref().unwrap_or(""))
    .bind(scope.runner.as_deref().unwrap_or(""))
    .fetch_all(pool)
    .await
}

/// Gather a scope's numbers over `range`, ending now.
pub async fn gather(pool: &Pool, scope: &Scope, range: Range) -> sqlx::Result<Insights> {
    let now = db::now_ms();
    let start = now - range.ms();
    let jobs = rows(pool, scope, start).await?;
    let runners = db::runners::list(pool).await?;
    let n = range.buckets();
    let width = range.bucket_ms();
    let bucket = |t: i64| -> Option<usize> { (t >= start && t <= now).then(|| (((t - start) / width) as usize).min(n - 1)) };

    let mut pickups: Vec<Vec<f64>> = vec![Vec::new(); n];
    let mut durations: Vec<Vec<f64>> = vec![Vec::new(); n];
    let (mut passed, mut failed, mut other) = (vec![0f64; n], vec![0f64; n], vec![0f64; n]);
    let (mut all_pickups, mut all_durations) = (Vec::new(), Vec::new());
    for j in &jobs {
        if let (Some(p), Some(b)) = (j.pickup(), j.started_ms.and_then(bucket)) {
            pickups[b].push(p);
            all_pickups.push(p);
        }
        if let (Some(f), true) = (j.finished_ms, j.ended()) {
            if let Some(b) = bucket(f) {
                match j.status.as_str() {
                    "passed" => passed[b] += 1.0,
                    "failed" => failed[b] += 1.0,
                    _ => other[b] += 1.0,
                }
                if let Some(d) = j.duration() {
                    durations[b].push(d);
                    all_durations.push(d);
                }
            }
        }
    }
    // The queue and the running jobs at each bucket's end.
    let (mut queued_at, mut running_at) = (vec![0f64; n], vec![0f64; n]);
    for (i, (q, r)) in queued_at.iter_mut().zip(running_at.iter_mut()).enumerate() {
        let t = (start + (i as i64 + 1) * width).min(now);
        for j in &jobs {
            let entered = j.queued_ms.unwrap_or(i64::MAX);
            if entered <= t && j.left_queue().is_none_or(|l| l > t) && !(j.ended() && j.finished_ms.is_none()) {
                *q += 1.0;
            }
            if j.started_ms.is_some_and(|s| s <= t) && j.finished_ms.is_none_or(|f| f > t) && !(j.ended() && j.finished_ms.is_none()) {
                *r += 1.0;
            }
        }
    }
    // Slots: the runners serving the scope that are online now.
    let serving: Vec<&db::runners::Runner> =
        runners.iter().filter(|r| scope.runner.as_deref().is_none_or(|n| n == r.name)).filter(|r| r.state != "removed").collect();
    let slots_total: i64 = serving.iter().filter(|r| r.online()).map(|r| r.max_jobs.max(1)).sum();
    let utilisation: Vec<Option<f64>> =
        running_at.iter().map(|r| (slots_total > 0).then(|| (r / slots_total as f64 * 100.0).min(100.0))).collect();

    let quantiles = |sets: &mut Vec<Vec<f64>>, p: f64| -> Vec<Option<f64>> { sets.iter_mut().map(|v| percentile(v, p)).collect() };
    let pickup = vec![
        Series { name: "median", class: "s-main", values: quantiles(&mut pickups, 0.5) },
        Series { name: "95th percentile", class: "s-warn", values: quantiles(&mut pickups, 0.95) },
    ];
    let durations_series = vec![
        Series { name: "median", class: "s-main", values: quantiles(&mut durations, 0.5) },
        Series { name: "95th percentile", class: "s-warn", values: quantiles(&mut durations, 0.95) },
    ];
    let some = |v: Vec<f64>| v.into_iter().map(Some).collect::<Vec<_>>();
    let ended_total = |v: &[f64]| v.iter().sum::<f64>() as i64;
    let (p_total, f_total, o_total) = (ended_total(&passed), ended_total(&failed), ended_total(&other));

    // Runners: what each did in the range.
    let mut runner_rows: Vec<RunnerRow> = serving
        .iter()
        .filter(|r| scope.owner.is_none() && scope.repo_id.is_none() || jobs.iter().any(|j| j.runner == r.name))
        .map(|r| {
            let mine: Vec<&JobRow> = jobs.iter().filter(|j| j.runner == r.name).collect();
            let mut p: Vec<f64> = mine.iter().filter(|j| j.started_ms.is_some_and(|s| s >= start)).filter_map(|j| j.pickup()).collect();
            let mut d: Vec<f64> = mine.iter().filter(|j| j.ended()).filter_map(|j| j.duration()).collect();
            RunnerRow {
                name: r.name.clone(),
                status: r.status(),
                running: mine.iter().filter(|j| j.status == "running").count() as i64,
                max_jobs: r.max_jobs.max(1),
                jobs: mine.iter().filter(|j| j.ended()).count() as i64,
                failed: mine.iter().filter(|j| j.status == "failed").count() as i64,
                pickup_median: percentile(&mut p, 0.5),
                duration_median: percentile(&mut d, 0.5),
                last_seen_ms: r.last_seen_ms,
            }
        })
        .collect();
    runner_rows.sort_by(|a, b| b.jobs.cmp(&a.jobs).then_with(|| a.name.cmp(&b.name)));

    // Repositories, busiest first (a repository's own page has just itself).
    let mut by_repo: std::collections::BTreeMap<String, (i64, i64, Vec<f64>, i64)> = Default::default();
    for j in &jobs {
        let e = by_repo.entry(format!("{}/{}", j.owner, j.repo)).or_default();
        if j.ended() && j.finished_ms.is_some_and(|f| f >= start) {
            e.0 += 1;
            if j.status == "failed" {
                e.1 += 1;
            }
        }
        if let Some(p) = j.pickup().filter(|_| j.started_ms.is_some_and(|s| s >= start)) {
            e.2.push(p);
        }
        if j.status == "queued" {
            e.3 += 1;
        }
    }
    let mut repos: Vec<RepoRow> = by_repo
        .into_iter()
        .map(|(slug, (jobs, failed, mut p, queued))| RepoRow {
            slug,
            jobs,
            failed,
            pickup_median: percentile(&mut p, 0.5),
            pickup_p95: percentile(&mut p, 0.95),
            queued,
        })
        .collect();
    repos.sort_by(|a, b| (b.jobs + b.queued).cmp(&(a.jobs + a.queued)).then_with(|| a.slug.cmp(&b.slug)));
    repos.truncate(15);

    Ok(Insights {
        range,
        start_ms: start,
        now_ms: now,
        pickup,
        results: vec![
            Series { name: "passed", class: "s-ok", values: some(passed) },
            Series { name: "failed", class: "s-bad", values: some(failed) },
            Series { name: "other (error, cancelled, skipped)", class: "s-mute", values: some(other) },
        ],
        durations: durations_series,
        load: vec![
            Series { name: "queued", class: "s-warn", values: some(queued_at) },
            Series { name: "running", class: "s-main", values: some(running_at) },
        ],
        utilisation: vec![Series { name: "busy slots", class: "s-main", values: utilisation }],
        ended: p_total + f_total + o_total,
        passed: p_total,
        failed: f_total,
        pickup_median: percentile(&mut all_pickups, 0.5),
        pickup_p95: percentile(&mut all_pickups, 0.95),
        duration_median: percentile(&mut all_durations, 0.5),
        queued_now: jobs.iter().filter(|j| j.status == "queued").count() as i64,
        running_now: jobs.iter().filter(|j| j.status == "running").count() as i64,
        runners_online: serving.iter().filter(|r| r.online()).count() as i64,
        runners_total: serving.len() as i64,
        slots_busy: jobs.iter().filter(|j| j.status == "running").count() as i64,
        slots_total,
        runners: runner_rows,
        repos,
    })
}

/// A span of seconds, short: `42 s`, `3.5 min`, `1.2 h`.
pub fn seconds(s: f64) -> String {
    if s < 60.0 {
        format!("{s:.0} s")
    } else if s < 3600.0 {
        let m = s / 60.0;
        if m < 10.0 {
            format!("{m:.1} min")
        } else {
            format!("{m:.0} min")
        }
    } else {
        format!("{:.1} h", s / 3600.0)
    }
}

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

    #[test]
    fn percentiles_are_nearest_rank() {
        let mut v = vec![5.0, 1.0, 3.0, 2.0, 4.0];
        assert_eq!(percentile(&mut v, 0.5), Some(3.0));
        assert_eq!(percentile(&mut v, 0.95), Some(5.0));
        assert_eq!(percentile(&mut [], 0.5), None);
        assert_eq!(seconds(42.0), "42 s");
        assert_eq!(seconds(210.0), "3.5 min");
        assert_eq!(seconds(4320.0), "1.2 h");
    }
}