//! 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");

    }

}