//! OxygenFlow runs, jobs and logs (hub spec §13.2).

use super::{now_ms, Pool};

use sqlx::FromRow;

#[derive(Clone, Debug, FromRow)]

pub struct Run {

    pub id: i64,

    pub repo_id: i64,

    pub bookmark: String,

    pub state: String,

    pub heads: String,

    pub pipeline_hash: String,

    pub status: String,

    pub error: String,

    pub actor: String,

    pub created_ms: i64,

    pub finished_ms: Option<i64>,

}

impl Run {

    pub fn head_list(&self) -> Vec<&str> {

        self.heads.split('+').filter(|h| !h.is_empty()).collect()

    }

    pub fn live(&self) -> bool {

        matches!(self.status.as_str(), "queued" | "running")

    }

}

#[derive(Clone, Debug, FromRow)]

pub struct Job {

    pub id: i64,

    pub run_id: i64,

    pub name: String,

    pub depth: i64,

    pub depends: String,

    pub spec: String,

    pub sandbox: String,

    pub runtime: String,

    pub status: String,

    pub runner: String,

    pub token_hash: String,

    pub started_ms: Option<i64>,

    pub finished_ms: Option<i64>,

    pub exit_code: Option<i64>,

    pub error: String,

    pub receipt: String,

    pub receipt_hash: String,

    pub log_count: i64,

    /// Comma-separated ids of other repositories the job token may read.

    pub also_read: String,

}

impl Job {

    pub fn depends_on(&self) -> Vec<String> {

        serde_json::from_str(&self.depends).unwrap_or_default()

    }

    pub fn done(&self) -> bool {

        matches!(self.status.as_str(), "passed" | "failed" | "skipped" | "error")

    }

}

#[derive(Clone, Debug, FromRow)]

pub struct LogLine {

    pub seq: i64,

    pub t: i64,

    pub stream: String,

    pub text: String,

}

/// A job as the trigger creates it.

pub struct NewJob {

    pub name: String,

    pub depth: i64,

    pub depends: Vec<String>,

    pub spec: String,

    pub sandbox: String,

    pub runtime: String,

}

pub struct NewRun<'a> {

    pub repo_id: i64,

    pub bookmark: &'a str,

    pub state: &'a str,

    pub heads: &'a str,

    pub pipeline_hash: &'a str,

    pub actor: &'a str,

    /// Problems when the pipeline is invalid: the run is recorded in error.

    pub error: Option<&'a str>,

}

pub async fn create_run(pool: &Pool, r: NewRun<'_>, jobs: &[NewJob]) -> sqlx::Result<i64> {

    let mut tx = pool.begin().await?;

    let now = now_ms();

    let status = if r.error.is_some() { "error" } else { "queued" };

    let id: i64 = sqlx::query_scalar(

        "INSERT INTO flow_runs (repo_id, bookmark, state, heads, pipeline_hash, status, error, actor, created_ms, finished_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",

    )

    .bind(r.repo_id)

    .bind(r.bookmark)

    .bind(r.state)

    .bind(r.heads)

    .bind(r.pipeline_hash)

    .bind(status)

    .bind(r.error.unwrap_or(""))

    .bind(r.actor)

    .bind(now)

    .bind(r.error.map(|_| now))

    .fetch_one(&mut *tx)

    .await?;

    for j in jobs {

        sqlx::query(

            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued')",

        )

        .bind(id)

        .bind(&j.name)

        .bind(j.depth)

        .bind(serde_json::to_string(&j.depends).unwrap_or_else(|_| "[]".into()))

        .bind(&j.spec)

        .bind(&j.sandbox)

        .bind(&j.runtime)

        .execute(&mut *tx)

        .await?;

    }

    tx.commit().await?;

    Ok(id)

}

pub async fn runs(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {

    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND id < ? ORDER BY id DESC LIMIT ?")

        .bind(repo_id)

        .bind(before.unwrap_or(i64::MAX))

        .bind(limit)

        .fetch_all(pool)

        .await

}

pub async fn run(pool: &Pool, repo_id: i64, id: i64) -> sqlx::Result<Option<Run>> {

    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND id = ?").bind(repo_id).bind(id).fetch_optional(pool).await

}

pub async fn run_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Run>> {

    sqlx::query_as("SELECT * FROM flow_runs WHERE id = ?").bind(id).fetch_optional(pool).await

}

pub async fn jobs(pool: &Pool, run_id: i64) -> sqlx::Result<Vec<Job>> {

    sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ? ORDER BY depth, name").bind(run_id).fetch_all(pool).await

}

pub async fn job(pool: &Pool, id: i64) -> sqlx::Result<Option<Job>> {

    sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(id).fetch_optional(pool).await

}

/// The running job a job token belongs to, while it may still be used.

pub async fn job_by_token(pool: &Pool, token_hash: &str, max_age_ms: i64) -> sqlx::Result<Option<Job>> {

    sqlx::query_as("SELECT * FROM flow_jobs WHERE token_hash = ? AND status = 'running' AND started_ms >= ?")

        .bind(token_hash)

        .bind(now_ms() - max_age_ms)

        .fetch_optional(pool)

        .await

}

/// Take the oldest queued job whose dependencies all passed and whose tier

/// the runner offers, and mark it running for `runner` with a token.

/// Let a job's token also read these repositories (its pinned subrepos).

pub async fn set_also_read(pool: &Pool, job_id: i64, repo_ids: &[i64]) -> sqlx::Result<()> {

    let list: Vec<String> = repo_ids.iter().map(|i| i.to_string()).collect();

    sqlx::query("UPDATE flow_jobs SET also_read = ? WHERE id = ?").bind(list.join(",")).bind(job_id).execute(pool).await?;

    Ok(())

}

impl Job {

    pub fn also_read_ids(&self) -> Vec<i64> {

        self.also_read.split(',').filter_map(|x| x.trim().parse().ok()).collect()

    }

}

pub async fn claim(pool: &Pool, runtimes: &[String], runner: &str, token_hash: &str) -> sqlx::Result<Option<Job>> {

    let mut tx = pool.begin().await?;

    let queued: Vec<Job> = sqlx::query_as(

        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id WHERE j.status = 'queued' AND r.status IN ('queued', 'running') ORDER BY j.run_id, j.depth, j.id LIMIT 200",

    )

    .fetch_all(&mut *tx)

    .await?;

    for j in queued {

        if !runtimes.iter().any(|r| r == &j.runtime) {

            continue;

        }

        let deps = j.depends_on();

        if !deps.is_empty() {

            let mut passed = 0i64;

            for d in &deps {

                let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = ? AND name = ?")

                    .bind(j.run_id)

                    .bind(d)

                    .fetch_optional(&mut *tx)

                    .await?;

                if st.as_deref() == Some("passed") {

                    passed += 1;

                }

            }

            if passed != deps.len() as i64 {

                continue;

            }

        }

        let now = now_ms();

        let done = sqlx::query(

            "UPDATE flow_jobs SET status = 'running', runner = ?, token_hash = ?, started_ms = ? WHERE id = ? AND status = 'queued'",

        )

        .bind(runner)

        .bind(token_hash)

        .bind(now)

        .bind(j.id)

        .execute(&mut *tx)

        .await?;

        if done.rows_affected() == 1 {

            sqlx::query("UPDATE flow_runs SET status = 'running' WHERE id = ? AND status = 'queued'")

                .bind(j.run_id)

                .execute(&mut *tx)

                .await?;

            tx.commit().await?;

            return job(pool, j.id).await;

        }

    }

    tx.commit().await?;

    Ok(None)

}

/// Append records; returns how many were kept (the rest exceed the cap).

pub async fn append_log(pool: &Pool, job_id: i64, records: &[flow_core::api::Record], cap: i64) -> sqlx::Result<usize> {

    let mut tx = pool.begin().await?;

    let count: i64 = sqlx::query_scalar("SELECT log_count FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;

    let room = (cap - count).max(0) as usize;

    let keep = &records[..records.len().min(room)];

    for (i, r) in keep.iter().enumerate() {

        sqlx::query("INSERT INTO flow_logs (job_id, seq, t, stream, text) VALUES (?, ?, ?, ?, ?)")

            .bind(job_id)

            .bind(count + i as i64 + 1)

            .bind(r.t as i64)

            .bind(&r.stream)

            .bind(&r.text)

            .execute(&mut *tx)

            .await?;

    }

    sqlx::query("UPDATE flow_jobs SET log_count = log_count + ? WHERE id = ?")

        .bind(keep.len() as i64)

        .bind(job_id)

        .execute(&mut *tx)

        .await?;

    tx.commit().await?;

    Ok(keep.len())

}

pub async fn log(pool: &Pool, job_id: i64, after: i64, limit: i64) -> sqlx::Result<Vec<LogLine>> {

    sqlx::query_as("SELECT seq, t, stream, text FROM flow_logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT ?")

        .bind(job_id)

        .bind(after)

        .bind(limit)

        .fetch_all(pool)

        .await

}

pub struct Outcome<'a> {

    pub status: &'a str,

    pub exit_code: Option<i64>,

    pub error: &'a str,

    pub receipt: &'a str,

    pub receipt_hash: &'a str,

}

/// Record a job's end, skip the jobs that can no longer run, and settle the

/// run. Returns the jobs whose status changed (for their checks).

pub async fn finish(pool: &Pool, job_id: i64, o: Outcome<'_>) -> sqlx::Result<Vec<Job>> {

    let mut tx = pool.begin().await?;

    let now = now_ms();

    let j: Job = sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;

    sqlx::query("UPDATE flow_jobs SET status = ?, exit_code = ?, error = ?, receipt = ?, receipt_hash = ?, finished_ms = ?, token_hash = '' WHERE id = ?")

        .bind(o.status)

        .bind(o.exit_code)

        .bind(o.error)

        .bind(o.receipt)

        .bind(o.receipt_hash)

        .bind(now)

        .bind(job_id)

        .execute(&mut *tx)

        .await?;

    let mut changed = vec![job_id];

    // Dependents of anything that did not pass are skipped, transitively.

    loop {

        let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;

        let blocked: Vec<i64> = all

            .iter()

            .filter(|x| x.status == "queued")

            .filter(|x| {

                x.depends_on()

                    .iter()

                    .any(|d| all.iter().any(|y| &y.name == d && matches!(y.status.as_str(), "failed" | "skipped" | "error")))

            })

            .map(|x| x.id)

            .collect();

        if blocked.is_empty() {

            break;

        }

        for id in blocked {

            sqlx::query("UPDATE flow_jobs SET status = 'skipped', finished_ms = ? WHERE id = ?")

                .bind(now)

                .bind(id)

                .execute(&mut *tx)

                .await?;

            changed.push(id);

        }

    }

    let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;

    if all.iter().all(Job::done) {

        let status = if all.iter().all(|x| x.status == "passed") { "passed" } else { "failed" };

        sqlx::query("UPDATE flow_runs SET status = ?, finished_ms = ? WHERE id = ?")

            .bind(status)

            .bind(now)

            .bind(j.run_id)

            .execute(&mut *tx)

            .await?;

    }

    tx.commit().await?;

    let mut out = Vec::new();

    for id in changed {

        if let Some(x) = job(pool, id).await? {

            out.push(x);

        }

    }

    Ok(out)

}