OxygenFlow runs: pushes start pipeline runs, a runner API with job tokens, jobs as checks, and a Flow tab with the job graph and live logs 42fa95540468

-- OxygenFlow runs, jobs and their logs (hub spec §13.2).

CREATE TABLE flow_runs (

    id            INTEGER PRIMARY KEY,

    repo_id       INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,

    bookmark      TEXT NOT NULL,

    -- The bookmark's state hash (checks are keyed by it) and its heads, joined with '+'.

    state         TEXT NOT NULL,

    heads         TEXT NOT NULL,

    pipeline_hash TEXT NOT NULL DEFAULT '',

    status        TEXT NOT NULL CHECK (status IN ('queued', 'running', 'passed', 'failed', 'error')),

    -- Pipeline problems, one per line, for a run in error.

    error         TEXT NOT NULL DEFAULT '',

    actor         TEXT NOT NULL DEFAULT '',

    created_ms    INTEGER NOT NULL,

    finished_ms   INTEGER

);

CREATE INDEX flow_runs_by_repo ON flow_runs (repo_id, id DESC);

CREATE TABLE flow_jobs (

    id           INTEGER PRIMARY KEY,

    run_id       INTEGER NOT NULL REFERENCES flow_runs(id) ON DELETE CASCADE,

    name         TEXT NOT NULL,

    depth        INTEGER NOT NULL,

    -- JSON: the job's dependencies, compiled form and sandbox (flow-core types).

    depends      TEXT NOT NULL,

    spec         TEXT NOT NULL,

    sandbox      TEXT NOT NULL,

    runtime      TEXT NOT NULL,

    status       TEXT NOT NULL CHECK (status IN ('queued', 'running', 'passed', 'failed', 'skipped', 'error')),

    runner       TEXT NOT NULL DEFAULT '',

    token_hash   TEXT NOT NULL DEFAULT '',

    started_ms   INTEGER,

    finished_ms  INTEGER,

    exit_code    INTEGER,

    error        TEXT NOT NULL DEFAULT '',

    receipt      TEXT NOT NULL DEFAULT '',

    receipt_hash TEXT NOT NULL DEFAULT '',

    log_count    INTEGER NOT NULL DEFAULT 0,

    UNIQUE (run_id, name)

);

CREATE INDEX flow_jobs_queued ON flow_jobs (status, id);

CREATE INDEX flow_jobs_token ON flow_jobs (token_hash) WHERE token_hash <> '';

CREATE TABLE flow_logs (

    job_id INTEGER NOT NULL REFERENCES flow_jobs(id) ON DELETE CASCADE,

    seq    INTEGER NOT NULL,

    t      INTEGER NOT NULL,

    stream TEXT NOT NULL,

    text   TEXT NOT NULL,

    PRIMARY KEY (job_id, seq)

);

//! 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,

}

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.

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)

}

//! OxygenFlow on the hub (hub spec §13.2): runs started by pushes, the

//! runner API, job tokens and the checks that mirror jobs. Pipelines are

//! parsed by `flow-core`, the same code `flow check` runs.

use crate::api::ApiError;

use crate::db::{self, flow as fdb};

use crate::repos::RepoName;

use crate::Shared;

use axum::extract::{Path, State};

use axum::http::{HeaderMap, StatusCode};

use axum::response::{IntoResponse, Response};

use axum::Json;

use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, MAX_RECORDS, MAX_RECORD_TEXT};

use flow_core::{Pipeline, PIPELINE_PATH};

use oxy_core::hash::ChangeHash;

/// Prefix of job tokens, so the wire protocol can tell them from personal tokens.

pub const JOB_TOKEN_PREFIX: &str = "flowjob_";

/// A job token is refused this long after its job started, whatever its state.

pub const JOB_TOKEN_MAX_AGE_MS: i64 = 2 * 3600 * 1000;

/// A bookmark moved by a push: start a run when the repository is allowed

/// and the new state has a pipeline.

pub async fn on_push(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, state: &str, heads: &[ChangeHash], actor: &str) {

    if bookmark.starts_with("tags/") || heads.is_empty() || !app.config.flow.allows(&row.owner, &row.name) {

        return;

    }

    let Some(repo) = RepoName::new(&row.owner, &row.name) else { return };

    let h = heads.to_vec();

    let text = app

        .repos

        .read(&repo, move |r, cache| {

            let loaded = cache.load(r, &h)?;

            Ok(match loaded.snapshot.files.get(PIPELINE_PATH).map(|f| &f.content) {

                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),

                _ => None,

            })

        })

        .await;

    let text = match text {

        Ok(Some(t)) => t,

        Ok(None) => return,

        Err(e) => {

            tracing::warn!("flow: reading the pipeline of {repo}: {e}");

            return;

        }

    };

    let mut hex: Vec<String> = heads.iter().map(|h| h.to_hex()).collect();

    hex.sort();

    let heads_s = hex.join("+");

    let base = db::flow::NewRun { repo_id: row.id, bookmark, state, heads: &heads_s, pipeline_hash: "", actor, error: None };

    let result = match Pipeline::parse(&text) {

        Err(problems) => {

            let msg = problems.to_string();

            db::flow::create_run(&app.db, db::flow::NewRun { error: Some(&msg), ..base }, &[]).await

        }

        Ok(p) => {

            let depths = p.depths();

            let jobs: Vec<fdb::NewJob> = p

                .order()

                .into_iter()

                .map(|j| {

                    let sb = &p.sandboxes[&j.sandbox];

                    fdb::NewJob {

                        name: j.name.clone(),

                        depth: depths[j.name.as_str()] as i64,

                        depends: j.depends_on.clone(),

                        spec: serde_json::to_string(j).unwrap_or_default(),

                        sandbox: serde_json::to_string(sb).unwrap_or_default(),

                        runtime: sb.runtime.as_str().into(),

                    }

                })

                .collect();

            let hash = hex::encode(p.hash());

            let id = db::flow::create_run(&app.db, db::flow::NewRun { pipeline_hash: &hash, ..base }, &jobs).await;

            if let Ok(id) = id {

                for j in &jobs {

                    report_check(app, row, state, id, &j.name, "queued", "waiting for a runner").await;

                }

            }

            id

        }

    };

    if let Err(e) = result {

        tracing::warn!("flow: recording a run of {repo}: {e}");

    }

}

async fn report_check(app: &Shared, row: &db::repos::RepoRow, state: &str, run_id: i64, job: &str, status: &str, summary: &str) {

    let url = format!("{}/{}/{}/flow/{run_id}", app.config.base_url(), row.owner, row.name);

    let name = format!("flow/{job}");

    let c = db::hooks::upsert_check(

        &app.db,

        row.id,

        db::hooks::CheckReport { state, name: &name, status, summary, url: &url, reporter: None, seal: "" },

    )

    .await;

    if let Ok(c) = c {

        let api = crate::api::check_of(&c);

        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });

        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;

    }

}

/// The check status a job's status shows as.

fn check_status(job_status: &str) -> &'static str {

    match job_status {

        "queued" => "queued",

        "running" => "running",

        "passed" => "passed",

        "skipped" => "skipped",

        _ => "failed",

    }

}

fn runner_auth(app: &Shared, headers: &HeaderMap) -> Result<(), ApiError> {

    let Some(t) = oxy_server::bearer_token(headers) else {

        return Err(ApiError(StatusCode::UNAUTHORIZED, "a runner token is required".into()));

    };

    if app.config.flow.runner_tokens.iter().any(|x| crate::auth::mac_equal(x, t)) {

        Ok(())

    } else {

        Err(ApiError(StatusCode::UNAUTHORIZED, "not a runner token".into()))

    }

}

/// `POST /api/v1/flow/claim`.

pub async fn claim(State(app): State<Shared>, headers: HeaderMap, Json(req): Json<ClaimRequest>) -> Result<Response, ApiError> {

    runner_auth(&app, &headers)?;

    let runner: String = req.runner.chars().filter(|c| !c.is_control()).take(64).collect();

    let secret = format!("{JOB_TOKEN_PREFIX}{}", crate::auth::random_hex(32));

    let Some(job) = fdb::claim(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret)).await? else {

        return Ok(StatusCode::NO_CONTENT.into_response());

    };

    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;

    let row =

        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;

    report_check(&app, &row, &run.state, run.id, &job.name, "running", &format!("running on {runner}")).await;

    let bad = |e: serde_json::Error| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}"));

    let claim = Claim {

        job_id: job.id,

        run_id: run.id,

        repo: format!("{}/{}", row.owner, row.name),

        clone_url: format!("{}/{}/{}", app.config.base_url(), row.owner, row.name),

        token: secret,

        heads: run.head_list().into_iter().map(String::from).collect(),

        pipeline_hash: run.pipeline_hash.clone(),

        job: serde_json::from_str(&job.spec).map_err(bad)?,

        sandbox: serde_json::from_str(&job.sandbox).map_err(bad)?,

        timeout_s: app.config.flow.job_timeout_minutes.max(1) * 60,

    };

    Ok(Json(claim).into_response())

}

async fn running_job(app: &Shared, id: i64) -> Result<fdb::Job, ApiError> {

    match fdb::job(&app.db, id).await? {

        Some(j) if j.status == "running" => Ok(j),

        Some(_) => Err(ApiError(StatusCode::CONFLICT, "this job is not running".into())),

        None => Err(ApiError(StatusCode::NOT_FOUND, "no such job".into())),

    }

}

/// `POST /api/v1/flow/jobs/{id}/log`.

pub async fn log(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(id): Path<i64>,

    Json(mut batch): Json<LogBatch>,

) -> Result<StatusCode, ApiError> {

    runner_auth(&app, &headers)?;

    let job = running_job(&app, id).await?;

    for r in &mut batch.records {

        if r.text.len() > MAX_RECORD_TEXT {

            let mut cut = MAX_RECORD_TEXT;

            while !r.text.is_char_boundary(cut) {

                cut -= 1;

            }

            r.text.truncate(cut);

        }

        if !matches!(r.stream.as_str(), "out" | "err" | "console" | "flow") {

            r.stream = "console".into();

        }

    }

    let kept = fdb::append_log(&app.db, job.id, &batch.records, MAX_RECORDS as i64).await?;

    if kept < batch.records.len() && job.log_count < MAX_RECORDS as i64 {

        tracing::info!("flow: job {} reached the log limit", job.id);

    }

    Ok(StatusCode::NO_CONTENT)

}

/// `POST /api/v1/flow/jobs/{id}/finish`.

pub async fn finish(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(id): Path<i64>,

    Json(f): Json<Finish>,

) -> Result<StatusCode, ApiError> {

    runner_auth(&app, &headers)?;

    let job = running_job(&app, id).await?;

    let status = match (f.exit_code, &f.error) {

        (Some(0), None) => "passed",

        (Some(_), _) => "failed",

        (None, _) => "error",

    };

    let (receipt, receipt_hash) = match &f.receipt {

        Some(r) if r.job_name == job.name && Some(r.exit_code as i32) == f.exit_code => {

            (serde_json::to_string(r).unwrap_or_default(), hex::encode(r.hash()))

        }

        Some(_) => return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt does not match the job".into())),

        None => (String::new(), String::new()),

    };

    let error = f.error.clone().unwrap_or_default();

    let changed = fdb::finish(

        &app.db,

        job.id,

        fdb::Outcome { status, exit_code: f.exit_code.map(i64::from), error: &error, receipt: &receipt, receipt_hash: &receipt_hash },

    )

    .await?;

    if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {

        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {

            for j in changed {

                let summary = match j.status.as_str() {

                    "passed" => format!("passed in {}", duration(&j)),

                    "failed" => format!("exit {} after {}", j.exit_code.unwrap_or(-1), duration(&j)),

                    "skipped" => "skipped: a dependency did not pass".into(),

                    _ => format!("could not run: {}", j.error),

                };

                report_check(&app, &row, &run.state, run.id, &j.name, check_status(&j.status), &summary).await;

            }

        }

    }

    Ok(StatusCode::NO_CONTENT)

}

/// A job's wall time, as the pages show it.

pub fn duration(j: &fdb::Job) -> String {

    match (j.started_ms, j.finished_ms) {

        (Some(a), Some(b)) => human_ms(b - a),

        (Some(a), None) => human_ms(db::now_ms() - a),

        _ => "–".into(),

    }

}

pub fn human_ms(ms: i64) -> String {

    let ms = ms.max(0);

    if ms < 1000 {

        format!("{ms} ms")

    } else if ms < 60_000 {

        format!("{:.1} s", ms as f64 / 1000.0)

    } else {

        format!("{} min {} s", ms / 60_000, (ms % 60_000) / 1000)

    }

}

//! The Flow tab (hub spec §13.2): runs of a repository, and a run's job

//! graph with live status and logs.

use super::{filters, lookup, Error, Page, RepoShell, Shell};

use crate::auth::Viewer;

use crate::db::{self, flow as fdb};

use crate::Shared;

use askama::Template;

use axum::extract::{Path, Query, State};

use axum::response::{IntoResponse, Response};

use axum::Json;

use serde::Deserialize;

pub struct RunRow {

    pub r: fdb::Run,

    pub duration: String,

    pub jobs: Vec<(String, String)>,

}

#[derive(Template)]

#[template(path = "flow_runs.html")]

pub struct RunsPage {

    shell: Shell,

    repo: RepoShell,

    runs: Vec<RunRow>,

    enabled: bool,

    older: Option<i64>,

}

#[derive(Deserialize)]

pub struct RunsQuery {

    before: Option<i64>,

}

const PER_PAGE: i64 = 30;

pub async fn runs(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Query(q): Query<RunsQuery>,

) -> Result<Page<RunsPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);

    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let list = fdb::runs(&app.db, row.id, PER_PAGE + 1, q.before).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let older = (list.len() as i64 > PER_PAGE).then(|| list[PER_PAGE as usize - 1].id);

    let mut runs = Vec::new();

    for r in list.into_iter().take(PER_PAGE as usize) {

        let jobs = fdb::jobs(&app.db, r.id).await.unwrap_or_default();

        let duration = match r.finished_ms {

            Some(f) => crate::flow::human_ms(f - r.created_ms),

            None if r.live() => crate::flow::human_ms(db::now_ms() - r.created_ms),

            None => "–".into(),

        };

        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r });

    }

    let enabled = app.config.flow.allows(&row.owner, &row.name);

    Ok(Page(RunsPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), runs, enabled, older }))

}

/// A job placed in the graph.

pub struct Node {

    pub j: fdb::Job,

    pub x: i64,

    pub y: i64,

    pub duration: String,

}

pub struct Edge {

    pub d: String,

    pub status: String,

}

pub struct Graph {

    pub nodes: Vec<Node>,

    pub edges: Vec<Edge>,

    pub width: i64,

    pub height: i64,

}

pub const NODE_W: i64 = 200;

pub const NODE_H: i64 = 48;

const COL_GAP: i64 = 72;

const ROW_GAP: i64 = 18;

const PAD: i64 = 12;

/// Columns by dependency depth, jobs in each column in name order.

pub fn layout(jobs: Vec<fdb::Job>) -> Graph {

    let mut by_depth: std::collections::BTreeMap<i64, Vec<fdb::Job>> = Default::default();

    for j in jobs {

        by_depth.entry(j.depth).or_default().push(j);

    }

    let rows = by_depth.values().map(Vec::len).max().unwrap_or(0) as i64;

    let height = PAD * 2 + rows * NODE_H + (rows - 1).max(0) * ROW_GAP;

    let mut nodes = Vec::new();

    for (col, (_, mut js)) in by_depth.into_iter().enumerate() {

        js.sort_by(|a, b| a.name.cmp(&b.name));

        let n = js.len() as i64;

        let column_h = n * NODE_H + (n - 1).max(0) * ROW_GAP;

        let top = (height - column_h) / 2;

        for (i, j) in js.into_iter().enumerate() {

            let duration = crate::flow::duration(&j);

            nodes.push(Node { x: PAD + col as i64 * (NODE_W + COL_GAP), y: top + i as i64 * (NODE_H + ROW_GAP), duration, j });

        }

    }

    let mut edges = Vec::new();

    for n in &nodes {

        for d in n.j.depends_on() {

            if let Some(from) = nodes.iter().find(|m| m.j.name == d) {

                let (x1, y1) = (from.x + NODE_W, from.y + NODE_H / 2);

                let (x2, y2) = (n.x, n.y + NODE_H / 2);

                let mid = (x1 + x2) / 2;

                edges.push(Edge { d: format!("M{x1},{y1} C{mid},{y1} {mid},{y2} {x2},{y2}"), status: from.j.status.clone() });

            }

        }

    }

    let cols = nodes.iter().map(|n| n.x).max().map(|x| x + NODE_W + PAD).unwrap_or(PAD * 2);

    Graph { nodes, edges, width: cols, height: height.max(NODE_H + PAD * 2) }

}

pub fn glyph(status: &str) -> &'static str {

    match status {

        "passed" => "✓",

        "failed" => "✕",

        "running" => "●",

        "skipped" => "⤼",

        "error" => "!",

        _ => "○",

    }

}

pub struct JobLog {

    pub j: fdb::Job,

    pub lines: Vec<fdb::LogLine>,

    pub duration: String,

    /// The last line shown; the live view asks for what follows.

    pub last_seq: i64,

}

#[derive(Template)]

#[template(path = "flow_run.html")]

pub struct RunPage {

    shell: Shell,

    repo: RepoShell,

    run: fdb::Run,

    graph: Graph,

    logs: Vec<JobLog>,

    duration: String,

}

/// Lines of a job's log a page shows before "the rest is loaded as it comes".

const PAGE_LINES: i64 = 5000;

pub async fn run(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, id)): Path<(String, String, i64)>,

) -> Result<Page<RunPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer);

    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let run = fdb::run(&app.db, row.id, id)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let mut logs = Vec::new();

    for j in &jobs {

        let lines = fdb::log(&app.db, j.id, 0, PAGE_LINES).await.unwrap_or_default();

        let last_seq = lines.last().map(|l| l.seq).unwrap_or(0);

        logs.push(JobLog { duration: crate::flow::duration(j), j: j.clone(), lines, last_seq });

    }

    let duration = match run.finished_ms {

        Some(f) => crate::flow::human_ms(f - run.created_ms),

        None => crate::flow::human_ms(db::now_ms() - run.created_ms),

    };

    let graph = layout(jobs);

    Ok(Page(RunPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), run, graph, logs, duration }))

}

#[derive(Deserialize)]

pub struct LiveQuery {

    /// `job:seq,job:seq`: the last log line each open job already shows.

    after: Option<String>,

}

/// `GET /{owner}/{name}/flow/{id}/live`: statuses and new log lines, for the

/// run page while it is live.

pub async fn live(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, id)): Path<(String, String, i64)>,

    Query(q): Query<LiveQuery>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, "", &viewer);

    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let run = fdb::run(&app.db, row.id, id)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let after: std::collections::HashMap<i64, i64> = q

        .after

        .unwrap_or_default()

        .split(',')

        .filter_map(|p| p.split_once(':'))

        .filter_map(|(a, b)| Some((a.parse().ok()?, b.parse().ok()?)))

        .collect();

    let mut out = Vec::new();

    for j in &jobs {

        let lines = match after.get(&j.id) {

            Some(seq) => fdb::log(&app.db, j.id, *seq, 1000).await.unwrap_or_default(),

            None => vec![],

        };

        out.push(serde_json::json!({

            "id": j.id,

            "name": j.name,

            "status": j.status,

            "glyph": glyph(&j.status),

            "duration": crate::flow::duration(j),

            "exit_code": j.exit_code,

            "lines": lines.iter().map(|l| serde_json::json!({ "seq": l.seq, "t": l.t, "stream": l.stream, "text": l.text })).collect::<Vec<_>>(),

        }));

    }

    Ok(Json(serde_json::json!({ "status": run.status, "live": run.live(), "jobs": out })).into_response())

}

//! OxygenFlow on the hub (hub spec §13.2), end to end: a real `oxy` push

//! starts a run, a runner claims jobs through the API in dependency order,

//! job tokens read only their repository, results become checks and pages.

use axum::body::Body;

use axum::http::{header, Request, StatusCode};

use http_body_util::BodyExt;

use oxy_core::protocol::{open_remote, RemoteOptions};

use oxy_core::repo::Repo;

use oxygenhub::config::Config;

use oxygenhub::repos::RepoName;

use oxygenhub::{db, Shared};

use std::path::Path;

use tower::ServiceExt;

const ROOTFS: &str = "36fc4e3238044e424219a2b9002bf61bbe1805fec14d52cf37e17fae77c9ff16";

fn hermetic() {

    static ONCE: std::sync::Once = std::sync::Once::new();

    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();

    ONCE.call_once(|| {

        let d = tempfile::tempdir().unwrap();

        std::env::set_var("OXY_HOME", d.path());

        DIR.set(d).ok();

    });

}

fn write(root: &Path, rel: &str, content: &str) {

    let p = root.join(rel);

    std::fs::create_dir_all(p.parent().unwrap()).unwrap();

    std::fs::write(p, content).unwrap();

}

struct Hub {

    app: Shared,

    base: String,

    _dir: tempfile::TempDir,

}

async fn hub() -> Hub {

    hermetic();

    let dir = tempfile::tempdir().unwrap();

    let mut config = Config::load(dir.path()).unwrap();

    config.auth.push_tokens = vec!["site".into()];

    config.flow.runner_tokens = vec!["runner-secret".into()];

    config.flow.repos = vec!["andy/*".into()];

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

    let port = listener.local_addr().unwrap().port();

    config.server.base_url = format!("http://127.0.0.1:{port}");

    let app = oxygenhub::build(config).await.unwrap();

    let router = oxygenhub::router(app.clone());

    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });

    Hub { app, base: format!("http://127.0.0.1:{port}"), _dir: dir }

}

async fn call(app: &Shared, method: &str, path: &str, bearer: Option<&str>, json: Option<String>) -> (StatusCode, String) {

    let mut req = Request::builder().method(method).uri(path);

    if let Some(b) = bearer {

        req = req.header(header::AUTHORIZATION, format!("Bearer {b}"));

    }

    let body = match json {

        Some(j) => {

            req = req.header(header::CONTENT_TYPE, "application/json");

            Body::from(j)

        }

        None => Body::empty(),

    };

    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();

    let status = res.status();

    let bytes = res.into_body().collect().await.unwrap().to_bytes();

    (status, String::from_utf8_lossy(&bytes).into_owned())

}

async fn claim(app: &Shared) -> Option<serde_json::Value> {

    let (st, body) =

        call(app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(r#"{"runner":"test-1","runtimes":["microvm"]}"#.into())).await;

    match st {

        StatusCode::NO_CONTENT => None,

        StatusCode::OK => Some(serde_json::from_str(&body).unwrap()),

        other => panic!("claim: {other} {body}"),

    }

}

fn receipt(job: &str, pipeline_hash: &str, exit: u32) -> serde_json::Value {

    serde_json::json!({

        "job_name": job, "pipeline_hash": pipeline_hash, "scope_tree_hash": "11".repeat(32),

        "inputs": [], "environment": "22".repeat(32), "outputs": [], "exit_code": exit, "duration_ms": 640

    })

}

async fn finish(app: &Shared, id: i64, body: serde_json::Value) -> StatusCode {

    call(app, "POST", &format!("/api/v1/flow/jobs/{id}/finish"), Some("runner-secret"), Some(body.to_string())).await.0

}

async fn create(app: &Shared, owner: &str, name: &str) {

    app.repos.create(&RepoName::parse(&format!("{owner}/{name}")).unwrap(), None).unwrap();

    db::repos::create(&app.db, owner, name, "", hub_api::Visibility::Public).await.unwrap();

}

fn client(url: &str, work: &Path) -> Repo {

    let mut c = Repo::init(work).unwrap();

    c.config.set("remote.origin.url", url).unwrap();

    c.config.set("remote.origin.token", "site").unwrap();

    c.save_config().unwrap();

    c

}

fn pipeline(extra: &str) -> String {

    format!(

        "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\njob \"test\" {{\n    sandbox \"alpine\"\n    depends-on \"build\"\n    exec \"sh\" \"-eu\" \"test.sh\"\n}}\njob \"lint\" {{\n    sandbox \"alpine\"\n    exec \"true\"\n}}\n{extra}"

    )

}

async fn runs_of(app: &Shared, owner: &str, name: &str) -> Vec<db::flow::Run> {

    let row = db::repos::get(&app.db, owner, name).await.unwrap().unwrap();

    for _ in 0..100 {

        let r = db::flow::runs(&app.db, row.id, 10, None).await.unwrap();

        if !r.is_empty() {

            return r;

        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    }

    vec![]

}

#[tokio::test(flavor = "multi_thread")]

async fn a_push_runs_the_pipeline_through_a_runner() {

    let h = hub().await;

    create(&h.app, "andy", "demo").await;

    create(&h.app, "andy", "secret").await;

    create(&h.app, "other", "plain").await;

    let url = format!("{}/andy/demo", h.base);

    let work = tempfile::tempdir().unwrap();

    let mut c = client(&url, work.path());

    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));

    write(work.path(), "test.sh", "echo ok\n");

    c.snapshot("test").unwrap();

    c.commit("Add a pipeline", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The push started a run with three queued jobs, and checks for them.

    let runs = runs_of(&h.app, "andy", "demo").await;

    assert_eq!(runs.len(), 1);

    let run = &runs[0];

    assert_eq!((run.bookmark.as_str(), run.status.as_str(), run.actor.as_str()), ("main", "queued", "site token"));

    assert_eq!(run.pipeline_hash.len(), 64);

    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();

    assert_eq!(jobs.iter().map(|j| (j.name.as_str(), j.depth)).collect::<Vec<_>>(), [("build", 0), ("lint", 0), ("test", 1)]);

    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();

    assert_eq!(

        checks.iter().map(|c| (c.name.as_str(), c.status.as_str())).collect::<Vec<_>>(),

        [("flow/build", "queued"), ("flow/lint", "queued"), ("flow/test", "queued")]

    );

    // Runners need a runner token.

    assert_eq!(

        call(&h.app, "POST", "/api/v1/flow/claim", Some("site"), Some(r#"{"runner":"x","runtimes":["microvm"]}"#.into())).await.0,

        StatusCode::UNAUTHORIZED

    );

    assert_eq!(

        call(&h.app, "POST", "/api/v1/flow/claim", None, Some(r#"{"runner":"x","runtimes":["microvm"]}"#.into())).await.0,

        StatusCode::UNAUTHORIZED

    );

    // A runner without the tier gets nothing.

    let (st, _) =

        call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(r#"{"runner":"x","runtimes":["wasm"]}"#.into())).await;

    assert_eq!(st, StatusCode::NO_CONTENT);

    // Claims come in dependency order; `test` waits for `build`.

    let a = claim(&h.app).await.unwrap();

    let b = claim(&h.app).await.unwrap();

    assert_eq!((a["job"]["name"].as_str(), b["job"]["name"].as_str()), (Some("build"), Some("lint")));

    assert!(claim(&h.app).await.is_none(), "test waits for build");

    assert_eq!(a["repo"], "andy/demo");

    assert_eq!(a["clone_url"], url.as_str());

    assert_eq!(a["sandbox"]["rootfs"], ROOTFS);

    assert_eq!(a["job"]["exec"], serde_json::json!(["make"]));

    assert_eq!(a["timeout_s"], 3600);

    let heads: Vec<String> = a["heads"].as_array().unwrap().iter().map(|v| v.as_str().unwrap().to_string()).collect();

    assert_eq!(heads.join("+"), run.heads);

    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();

    assert_eq!(checks.iter().find(|c| c.name == "flow/build").unwrap().status, "running");

    // The job token reads this repository over the wire, and nothing else.

    let token = a["token"].as_str().unwrap().to_string();

    assert!(token.starts_with("flowjob_"));

    let opts = RemoteOptions { token: Some(token.clone()), ca: None, repo_id: None };

    open_remote(&url, &opts).unwrap().refs().unwrap();

    assert!(open_remote(&format!("{}/andy/secret", h.base), &opts).unwrap().refs().is_err(), "another repository");

    let bad = RemoteOptions { token: Some("flowjob_00".into()), ca: None, repo_id: None };

    assert!(open_remote(&url, &bad).unwrap().refs().is_err(), "an unknown job token");

    // … and cannot push.

    c.config.set("remote.origin.token", &token).unwrap();

    c.save_config().unwrap();

    write(work.path(), "more.txt", "x\n");

    c.snapshot("test").unwrap();

    c.commit("More", &[]).unwrap();

    let pushed = c.push("origin", false);

    assert!(pushed.is_err() || pushed.unwrap().iter().all(|r| !r.ok), "a job token cannot push");

    c.config.set("remote.origin.token", "site").unwrap();

    c.save_config().unwrap();

    // Logs, then results with receipts.

    let a_id = a["job_id"].as_i64().unwrap();

    let b_id = b["job_id"].as_i64().unwrap();

    let log = serde_json::json!({ "records": [

        { "t": 5, "stream": "flow", "text": "booting" },

        { "t": 120, "stream": "out", "text": "cc -o demo demo.c" },

        { "t": 130, "stream": "weird", "text": "x".repeat(20_000) },

    ]});

    assert_eq!(

        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{a_id}/log"), Some("runner-secret"), Some(log.to_string())).await.0,

        StatusCode::NO_CONTENT

    );

    let lines = db::flow::log(&h.app.db, a_id, 0, 100).await.unwrap();

    assert_eq!(lines.len(), 3);

    assert_eq!((lines[2].stream.as_str(), lines[2].text.len()), ("console", 16 * 1024), "streams are known ones, text is capped");

    let ph = a["pipeline_hash"].as_str().unwrap();

    assert_eq!(

        finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("lint", ph, 0) })).await,

        StatusCode::BAD_REQUEST,

        "a receipt for another job"

    );

    assert_eq!(

        finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("build", ph, 0) })).await,

        StatusCode::NO_CONTENT

    );

    assert_eq!(finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700 })).await, StatusCode::CONFLICT, "only once");

    assert!(open_remote(&url, &opts).unwrap().refs().is_err(), "the token ends with its job");

    assert_eq!(

        finish(&h.app, b_id, serde_json::json!({ "exit_code": 0, "duration_ms": 300, "receipt": receipt("lint", ph, 0) })).await,

        StatusCode::NO_CONTENT

    );

    let t = claim(&h.app).await.unwrap();

    assert_eq!(t["job"]["name"], "test");

    let t_id = t["job_id"].as_i64().unwrap();

    assert_eq!(

        finish(&h.app, t_id, serde_json::json!({ "exit_code": 3, "duration_ms": 500, "receipt": receipt("test", ph, 3) })).await,

        StatusCode::NO_CONTENT

    );

    let run = db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap();

    assert_eq!(run.status, "failed");

    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();

    let build = jobs.iter().find(|j| j.name == "build").unwrap();

    assert_eq!((build.status.as_str(), build.exit_code, build.receipt_hash.len()), ("passed", Some(0), 64));

    assert_eq!(jobs.iter().find(|j| j.name == "test").unwrap().status, "failed");

    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();

    assert_eq!(

        checks.iter().map(|c| (c.name.as_str(), c.status.as_str())).collect::<Vec<_>>(),

        [("flow/build", "passed"), ("flow/lint", "passed"), ("flow/test", "failed")]

    );

    assert!(checks[2].summary.contains("exit 3") && checks[2].url.ends_with(&format!("/andy/demo/flow/{}", run.id)));

    // Pages: the list, the run with its graph and logs, and the live view.

    let (st, list) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;

    assert_eq!(st, StatusCode::OK);

    assert!(list.contains(&format!("/andy/demo/flow/{}", run.id)) && list.contains("st-failed"), "{list}");

    let (st, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.id), None, None).await;

    assert_eq!(st, StatusCode::OK);

    assert!(page.contains("<svg class=\"flow-graph\""), "graph");

    assert_eq!(page.matches("data-job-node=").count(), 3);

    assert_eq!(page.matches("<path class=\"edge").count(), 1, "one dependency edge");

    assert!(page.contains("cc -o demo demo.c"), "log shown");

    assert!(!page.contains("data-flow-live"), "a finished run does not poll");

    let (st, live) = call(&h.app, "GET", &format!("/andy/demo/flow/{}/live?after={a_id}:1", run.id), None, None).await;

    assert_eq!(st, StatusCode::OK);

    let live: serde_json::Value = serde_json::from_str(&live).unwrap();

    assert_eq!(live["status"], "failed");

    let build_live = live["jobs"].as_array().unwrap().iter().find(|j| j["name"] == "build").unwrap().clone();

    assert_eq!(build_live["lines"].as_array().unwrap().len(), 2, "lines after seq 1");

    assert_eq!(call(&h.app, "GET", "/andy/demo/flow/999", None, None).await.0, StatusCode::NOT_FOUND);

    // A failed dependency skips what follows; an unrun job is an error.

    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));

    write(work.path(), "test.sh", "echo changed\n");

    c.snapshot("test").unwrap();

    c.commit("Change the test", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    let mut second = None;

    for _ in 0..100 {

        let r = runs_of(&h.app, "andy", "demo").await;

        if r.len() == 2 {

            second = Some(r[0].clone());

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    }

    let second = second.expect("a second run");

    let x = claim(&h.app).await.unwrap();

    let y = claim(&h.app).await.unwrap();

    let (build_id, lint_id) = if x["job"]["name"] == "build" {

        (x["job_id"].as_i64().unwrap(), y["job_id"].as_i64().unwrap())

    } else {

        (y["job_id"].as_i64().unwrap(), x["job_id"].as_i64().unwrap())

    };

    assert_eq!(

        finish(&h.app, build_id, serde_json::json!({ "exit_code": null, "duration_ms": 10, "error": "rootfs is not on this runner" }))

            .await,

        StatusCode::NO_CONTENT

    );

    assert_eq!(finish(&h.app, lint_id, serde_json::json!({ "exit_code": 0, "duration_ms": 10 })).await, StatusCode::NO_CONTENT);

    assert!(claim(&h.app).await.is_none(), "test was skipped");

    let jobs = db::flow::jobs(&h.app.db, second.id).await.unwrap();

    assert_eq!(

        jobs.iter().map(|j| (j.name.as_str(), j.status.as_str())).collect::<Vec<_>>(),

        [("build", "error"), ("lint", "passed"), ("test", "skipped")]

    );

    assert_eq!(db::flow::run_by_id(&h.app.db, second.id).await.unwrap().unwrap().status, "failed");

    // An invalid pipeline is a run in error with the line.

    write(work.path(), ".flow/pipeline.kdl", &pipeline("job \"broken\" {\n    sandbox \"missing\"\n    exec \"x\"\n}\n"));

    c.snapshot("test").unwrap();

    c.commit("Break the pipeline", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    let mut third = None;

    for _ in 0..100 {

        let r = runs_of(&h.app, "andy", "demo").await;

        if r.len() == 3 {

            third = Some(r[0].clone());

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    }

    let third = third.expect("a third run");

    assert_eq!(third.status, "error");

    assert!(third.error.contains("line 18") && third.error.contains("\"missing\""), "{}", third.error);

    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", third.id), None, None).await;

    assert!(page.contains("The pipeline is not valid"));

    // A repository outside [flow] repos starts nothing.

    let work2 = tempfile::tempdir().unwrap();

    let c2 = client(&format!("{}/other/plain", h.base), work2.path());

    write(work2.path(), ".flow/pipeline.kdl", &pipeline(""));

    c2.snapshot("test").unwrap();

    c2.commit("Pipeline", &[]).unwrap();

    assert!(c2.push("origin", false).unwrap().iter().all(|r| r.ok));

    tokio::time::sleep(std::time::Duration::from_millis(500)).await;

    let row = db::repos::get(&h.app.db, "other", "plain").await.unwrap().unwrap();

    assert!(db::flow::runs(&h.app.db, row.id, 10, None).await.unwrap().is_empty());

    let (_, page) = call(&h.app, "GET", "/other/plain/flow", None, None).await;

    assert!(page.contains("not enabled for this repository"));

}

{% extends "base.html" %}

{% import "repo_tabs.html" as m %}

{% block title %}Run #{{ run.id }} · Flow · {{ repo.owner }}/{{ repo.name }}{% endblock %}

{% block bodyattrs %} data-repo="{{ repo.url }}"{% if run.live() %} data-flow-live="{{ repo.url }}/flow/{{ run.id }}/live"{% endif %}{% endblock %}

{% block tabs %}{% call m::tabs(repo, "flow") %}{% endblock %}

{% block mainclass %} class="wide"{% endblock %}

{% block content %}

<div class="head">

  <h1><span class="flow-glyph st-{{ run.status }}" data-run-glyph>{{ crate::web::flow::glyph(run.status) }}</span> Run #{{ run.id }} <small data-run-status>{{ run.status }}</small></h1>

  <p class="small mute"><span class="lbl lbl-bm">{{ run.bookmark }}</span> at <a class="px" href="{{ repo.url }}/tree/{{ run.heads }}" title="{{ run.state }}">{{ run.state|short }}</a> · pushed by {{ run.actor }} <time title="{{ run.created_ms|when_i }}">{{ run.created_ms|ago_i }}</time> · {{ duration }}{% if !run.pipeline_hash.is_empty() %} · pipeline <code title="{{ run.pipeline_hash }}">{{ run.pipeline_hash|short }}</code>{% endif %} · <a href="{{ repo.url }}/file/{{ run.heads }}/.flow/pipeline.kdl">.flow/pipeline.kdl</a> · <a href="{{ repo.url }}/flow">all runs</a></p>

</div>

{% if run.status == "error" %}

<section class="panel"><div class="strip bad">The pipeline is not valid, so nothing ran.</div><pre class="snippet">{{ run.error }}</pre><p class="small mute">Run <code>flow check</code> locally to see the same problems.</p></section>

{% else %}

<section class="panel flow-graph-box">

  <svg class="flow-graph" viewBox="0 0 {{ graph.width }} {{ graph.height }}" width="{{ graph.width }}" height="{{ graph.height }}" role="img" aria-label="Job graph of run {{ run.id }}">

    {% for e in graph.edges %}<path class="edge st-{{ e.status }}" d="{{ e.d }}"/>{% endfor %}

    {% for n in graph.nodes %}

    <a href="#job-{{ n.j.id }}" data-job-node="{{ n.j.id }}" class="node st-{{ n.j.status }}">

      <rect x="{{ n.x }}" y="{{ n.y }}" width="{{ crate::web::flow::NODE_W }}" height="{{ crate::web::flow::NODE_H }}" rx="6"/>

      <text class="g" x="{{ n.x + 18 }}" y="{{ n.y + 30 }}" text-anchor="middle" data-glyph>{{ crate::web::flow::glyph(n.j.status) }}</text>

      <text class="name" x="{{ n.x + 34 }}" y="{{ n.y + 21 }}">{{ n.j.name }}</text>

      <text class="meta" x="{{ n.x + 34 }}" y="{{ n.y + 37 }}" data-meta>{{ n.j.status }} · {{ n.duration }}</text>

    </a>

    {% endfor %}

  </svg>

</section>

{% for l in logs %}

<details class="panel flow-job st-{{ l.j.status }}" id="job-{{ l.j.id }}" data-job="{{ l.j.id }}"{% if l.j.status == "failed" || l.j.status == "error" || l.j.status == "running" %} open{% endif %}>

  <summary><span class="flow-glyph st-{{ l.j.status }}" data-glyph>{{ crate::web::flow::glyph(l.j.status) }}</span> <strong>{{ l.j.name }}</strong> <span class="small mute" data-meta>{{ l.j.status }}{% if let Some(c) = l.j.exit_code %} · exit {{ c }}{% endif %} · {{ l.duration }}{% if !l.j.runner.is_empty() %} · on {{ l.j.runner }}{% endif %}</span>{% if !l.j.receipt_hash.is_empty() %} <span class="small mute">· receipt <code title="{{ l.j.receipt_hash }}">{{ l.j.receipt_hash|short }}</code></span>{% endif %}</summary>

  {% if !l.j.error.is_empty() %}<div class="strip bad">{{ l.j.error }}</div>{% endif %}

  <pre class="flow-log" data-log data-seq="{{ l.last_seq }}">{% for line in l.lines %}<span class="ln s-{{ line.stream }}"><span class="t">{{ line.t / 1000 }}.{{ "{:01}"|format(line.t % 1000 / 100) }}s</span>{{ line.text }}</span>

{% endfor %}</pre>

  {% if l.lines.is_empty() && l.j.status == "queued" %}<p class="small mute pad">Waiting for a runner{% if !l.j.depends_on().is_empty() %} and for {{ l.j.depends_on().join(", ") }}{% endif %}.</p>{% endif %}

  {% if l.lines.is_empty() && l.j.status == "skipped" %}<p class="small mute pad">Skipped: a job it depends on did not pass.</p>{% endif %}

</details>

{% endfor %}

{% endif %}

{% endblock %}

{% extends "base.html" %}

{% import "repo_tabs.html" as m %}

{% block title %}Flow · {{ repo.owner }}/{{ repo.name }}{% endblock %}

{% block bodyattrs %} data-repo="{{ repo.url }}"{% endblock %}

{% block tabs %}{% call m::tabs(repo, "flow") %}{% endblock %}

{% block content %}

<div class="head"><h1>Flow <small>pipeline runs</small></h1></div>

{% if runs.is_empty() %}

<section class="panel">

  {% if enabled %}

  <p>No runs yet. Add <code>.flow/pipeline.kdl</code> to a bookmark and push: every push that moves a bookmark runs it in OxygenFlow microVMs.</p>

  {% else %}

  <p>OxygenFlow is not enabled for this repository. A site administrator lists it under <code>[flow] repos</code> in <code>hub.toml</code>.</p>

  {% endif %}

<pre class="snippet">schema-version 1

sandbox "alpine" {

    runtime "microvm"

    rootfs "cas://blob/…"   // flow rootfs import prints this

    memory-limit "512MiB"

}

job "test" {

    sandbox "alpine"

    exec "sh" "-eu" "test.sh"

}</pre>

</section>

{% else %}

<table class="list flow-runs">

  <thead><tr><th></th><th>Run</th><th>Jobs</th><th>Bookmark</th><th>State</th><th>By</th><th>When</th><th>Took</th></tr></thead>

  <tbody>

  {% for row in runs %}

  <tr class="st-{{ row.r.status }}">

    <td class="st"><span class="flow-glyph st-{{ row.r.status }}" title="{{ row.r.status }}">{{ crate::web::flow::glyph(row.r.status) }}</span></td>

    <td><a href="{{ repo.url }}/flow/{{ row.r.id }}"><strong>#{{ row.r.id }}</strong></a> <span class="small mute">{{ row.r.status }}</span></td>

    <td class="flow-dots">{% for (n, s) in row.jobs %}<span class="flow-glyph st-{{ s }}" title="{{ n }}: {{ s }}">{{ crate::web::flow::glyph(s) }}</span>{% endfor %}{% if row.r.status == "error" %}<span class="small bad">pipeline invalid</span>{% endif %}</td>

    <td><span class="lbl lbl-bm">{{ row.r.bookmark }}</span></td>

    <td><a class="px" href="{{ repo.url }}/tree/{{ row.r.heads }}" title="{{ row.r.state }}">{{ row.r.state|short }}</a></td>

    <td>{{ row.r.actor }}</td>