Approval gates: a job waits for a person with the role 417184c73f4e

-- Approval gates (oxygenflow#57): a job whose pipeline asks for approval

-- waits ('waiting') until someone with the role approves it ('approved');

-- '' for a job that needs none. Who and when are kept on the job.

ALTER TABLE flow_jobs ADD COLUMN approval TEXT NOT NULL DEFAULT '';

ALTER TABLE flow_jobs ADD COLUMN approved_by TEXT NOT NULL DEFAULT '';

ALTER TABLE flow_jobs ADD COLUMN approved_ms INTEGER;

        .route("/repos/{owner}/{name}/flow/runs/{number}/jobs/{job}/approve", post(approve_flow_job))

        .route("/repos/{owner}/{name}/flow/runs/{number}/jobs/{job}/reject", post(reject_flow_job))

/// `POST /repos/{o}/{n}/flow/runs/{number}/jobs/{job}/approve` and

/// `.../reject` (oxygenflow#57): the job's approval rules decide who may.

async fn approve_flow_job(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path((owner, name, number, job)): Path<(String, String, i64, String)>,

) -> Result<Json<serde_json::Value>, ApiError> {

    decide_flow_job(&app, &headers, &owner, &name, number, &job, true).await

}

async fn reject_flow_job(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path((owner, name, number, job)): Path<(String, String, i64, String)>,

) -> Result<Json<serde_json::Value>, ApiError> {

    decide_flow_job(&app, &headers, &owner, &name, number, &job, false).await

}

async fn decide_flow_job(

    app: &Shared,

    headers: &HeaderMap,

    owner: &str,

    name: &str,

    number: i64,

    job: &str,

    approve: bool,

) -> Result<Json<serde_json::Value>, ApiError> {

    let c = caller(app, headers).await?;

    let me = c.require("repo:write")?.clone();

    let (_, row, _) = reach(app, &c, owner, name, Access::Read).await?;

    let run = db::flow::run(&app.db, row.id, number).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;

    let said = crate::flow::decide(app, &row, &run, job, &me, approve).await.map_err(|m| ApiError(StatusCode::CONFLICT, m))?;

    Ok(Json(serde_json::json!({ "number": number, "job": job, "decision": if approve { "approved" } else { "rejected" }, "said": said })))

}

    /// `waiting` for a person, `approved`, or empty (oxygenflow#57).

    pub approval: String,

    /// Who approved or rejected it, and when.

    pub approved_by: String,

    pub approved_ms: Option<i64>,

            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status, need_disk_mib, approval) \

             VALUES (?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?)",

        .bind(if needs_approval(&j.spec) { "waiting" } else { "" })

/// Whether a compiled job asks for approval (oxygenflow#57).

fn needs_approval(spec: &str) -> bool {

    serde_json::from_str::<serde_json::Value>(spec).is_ok_and(|v| !v["approval"].is_null())

}

         WHERE j.status IN ('queued', 'running') AND j.approval != 'waiting' AND r.status IN ('queued', 'running') \

         WHERE j.status = 'queued' AND j.approval != 'waiting' AND r.status IN ('queued', 'running') \

         WHERE j.status = 'queued' AND j.approval != 'waiting' AND r.status IN ('queued', 'running') ORDER BY j.run_id, j.depth, j.id LIMIT 500",

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

             peak_memory_mib = NULL, approval = CASE WHEN approval = '' THEN '' ELSE 'waiting' END, approved_by = '',

             approved_ms = NULL WHERE id = ?",

    .await

}

/// Approve job `id` if it is still waiting for a person (oxygenflow#57);

/// whether it was.

pub async fn approve_job(pool: &Pool, id: i64, who: &str) -> sqlx::Result<bool> {

    let r = sqlx::query(

        "UPDATE flow_jobs SET approval = 'approved', approved_by = ?, approved_ms = ? WHERE id = ? AND status = 'queued' AND approval = 'waiting'",

    )

    .bind(who)

    .bind(now_ms())

    .bind(id)

    .execute(pool)

    .await?;

    Ok(r.rows_affected() > 0)

}

/// Note who rejected job `id` (it is then finished as cancelled).

pub async fn note_rejection(pool: &Pool, id: i64, who: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_jobs SET approved_by = ?, approved_ms = ? WHERE id = ?")

        .bind(who)

        .bind(now_ms())

        .bind(id)

        .execute(pool)

        .await?;

    Ok(())

}

/// Jobs waiting for a person, in runs still going.

pub async fn waiting_for_approval(pool: &Pool) -> sqlx::Result<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 j.approval = 'waiting' AND r.status IN ('queued', 'running')",

    )

    .fetch_all(pool)

/// Why job `job` of `run` cannot be decided on by `who` (oxygenflow#57),

/// or its approval: it must be waiting, the jobs it depends on done and

/// passed, `who` must have the role it names on the repository, and with

/// others-only must not be the person whose push started the run.

pub async fn may_decide(

    app: &Shared,

    row: &db::repos::RepoRow,

    run: &fdb::Run,

    job: &fdb::Job,

    who: &db::accounts::Account,

) -> Result<flow_core::Approval, String> {

    let spec: flow_core::Job = serde_json::from_str(&job.spec).map_err(|e| format!("stored job: {e}"))?;

    let Some(approval) = spec.approval else { return Err(format!("{} does not wait for approval", job.name)) };

    if job.status != "queued" || job.approval != "waiting" {

        return Err(format!("{} is not waiting for approval", job.name));

    }

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

    if let Some(d) = job.depends_on().iter().find(|d| !jobs.iter().any(|j| &j.name == *d && j.status == "passed")) {

        return Err(format!("{} waits for {d} first", job.name));

    }

    let need = crate::auth::roles::Role::parse(&approval.by).unwrap_or(crate::auth::roles::Role::Maintain);

    let role = crate::auth::roles::role_for(&app.db, Some(who), row).await.map_err(|e| e.to_string())?;

    if role.is_none_or(|r| r < need) {

        return Err(format!("approving {} takes the {} role on {}/{}", job.name, approval.by, row.owner, row.name));

    }

    if approval.others_only && who.username == run.actor {

        return Err(format!("{} must be approved by someone other than {}, whose push started the run", job.name, run.actor));

    }

    Ok(approval)

}

/// Approve or reject job `name` of `run` as `who` (oxygenflow#57): the

/// decision goes on the job, its log and the audit log; a rejected job ends

/// as cancelled and the jobs after it are skipped. What happened, or why not.

pub async fn decide(

    app: &Shared,

    row: &db::repos::RepoRow,

    run: &fdb::Run,

    name: &str,

    who: &db::accounts::Account,

    approve: bool,

) -> Result<String, String> {

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

    let Some(job) = jobs.into_iter().find(|j| j.name == name) else { return Err(format!("run #{} has no job {name}", run.number)) };

    may_decide(app, row, run, &job, who).await?;

    let target = format!("{}/{}#{} {name}", row.owner, row.name, run.number);

    let said = if approve {

        if !fdb::approve_job(&app.db, job.id, &who.username).await.map_err(|e| e.to_string())? {

            return Err(format!("{name} was decided on meanwhile"));

        }

        db::accounts::audit(&app.db, Some(who), "flow.approved", &target, "").await.ok();

        format!("approved by {}", who.username)

    } else {

        fdb::note_rejection(&app.db, job.id, &who.username).await.map_err(|e| e.to_string())?;

        let why = format!("rejected by {}", who.username);

        let changed =

            fdb::finish(&app.db, job.id, fdb::Outcome { status: "cancelled", exit_code: None, error: &why, receipt: "", receipt_hash: "" })

                .await

                .map_err(|e| e.to_string())?;

        report_changed(app, run.id, changed).await.ok();

        db::accounts::audit(&app.db, Some(who), "flow.rejected", &target, "").await.ok();

        why

    };

    let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: said.clone() };

    fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();

    Ok(format!("{name}: {said}"))

}

/// End the approvals that waited too long (oxygenflow#57): from when the

/// jobs a waiting job depends on finished (or its run began).

pub async fn expire_approvals(app: &Shared, now: i64) -> sqlx::Result<usize> {

    let mut ended = 0;

    for job in fdb::waiting_for_approval(&app.db).await? {

        let Ok(spec) = serde_json::from_str::<flow_core::Job>(&job.spec) else { continue };

        let Some(approval) = spec.approval else { continue };

        let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? else { continue };

        let jobs = fdb::jobs(&app.db, run.id).await?;

        let wanted = job.depends_on();

        let deps: Vec<&fdb::Job> = jobs.iter().filter(|j| wanted.contains(&j.name)).collect();

        if deps.iter().any(|d| d.status != "passed") {

            continue;

        }

        let since = deps.iter().filter_map(|d| d.finished_ms).max().unwrap_or(run.created_ms);

        if now - since < approval.expires_s as i64 * 1000 {

            continue;

        }

        let hours = approval.expires_s as f64 / 3600.0;

        let why =

            format!("no approval within {}", if hours >= 1.0 { format!("{hours:.0}h") } else { format!("{}m", approval.expires_s / 60) });

        let changed =

            fdb::finish(&app.db, job.id, fdb::Outcome { status: "cancelled", exit_code: None, error: &why, receipt: "", receipt_hash: "" })

                .await?;

        report_changed(app, run.id, changed).await?;

        ended += 1;

    }

    Ok(ended)

}

        }

        match expire_approvals(&app, db::now_ms()).await {

            Ok(n) if n > 0 => tracing::info!("flow: {n} approvals expired"),

            Ok(_) => {}

            Err(e) => tracing::warn!("flow: expiring approvals: {e}"),

    /// Jobs that wait, or waited, for a person (oxygenflow#57).

    approvals: Vec<ApprovalView>,

/// A job's approval as the run page shows it.

pub struct ApprovalView {

    pub job: String,

    pub by: String,

    pub others_only: bool,

    /// `waiting`, `approved`, `rejected` or `expired`.

    pub state: &'static str,

    pub who: String,

    pub when_ms: i64,

    /// Whether the viewer may approve or reject it now, else why not.

    pub may: Result<(), String>,

}

impl ApprovalView {

    pub fn may_decide(&self) -> bool {

        self.may.is_ok()

    }

    pub fn why_not(&self) -> String {

        self.may.clone().err().unwrap_or_default()

    }

}

    let mut approvals = Vec::new();

    for j in jobs.iter().filter(|j| !j.approval.is_empty()) {

        let Some(a) = serde_json::from_str::<flow_core::Job>(&j.spec).ok().and_then(|s| s.approval) else { continue };

        let state = match (j.approval.as_str(), j.status.as_str()) {

            ("waiting", "queued") => "waiting",

            ("approved", _) => "approved",

            (_, "cancelled") if !j.approved_by.is_empty() => "rejected",

            (_, "cancelled") if j.error.starts_with("no approval within") => "expired",

            _ => continue,

        };

        let may = match (state, viewer.account()) {

            ("waiting", Some(me)) => crate::flow::may_decide(&app, &row, &run, j, me).await.map(|_| ()),

            ("waiting", None) => Err("sign in to approve".into()),

            _ => Err(String::new()),

        };

        approvals.push(ApprovalView {

            job: j.name.clone(),

            by: a.by,

            others_only: a.others_only,

            state,

            who: j.approved_by.clone(),

            when_ms: j.approved_ms.unwrap_or(0),

            may,

        });

    }

        approvals,

}

#[derive(Deserialize)]

pub struct DecideForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    /// `approve` or `reject`.

    action: String,

}

/// `POST /{owner}/{name}/flow/{id}/jobs/{job}/decide`: approve or reject

/// a job that waits for a person (oxygenflow#57).

pub async fn decide(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, number, job)): Path<(String, String, i64, String)>,

    Form(f): Form<DecideForm>,

) -> Result<Response, Error> {

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

    let me = require(&viewer, &shell)?;

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

    csrf(&viewer, &f.csrf_token, &shell)?;

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

        .await

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

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

    let approve = match f.action.as_str() {

        "approve" => true,

        "reject" => false,

        _ => return Err(Error::bad(shell, "approve or reject")),

    };

    match crate::flow::decide(&app, &row, &run, &job, &me.account, approve).await {

        Ok(_) => Ok(Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response()),

        Err(why) => Err(Error::bad(shell, why)),

    }

        .route("/{owner}/{name}/flow/{id}/jobs/{job}/decide", post(flow::decide))

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

async fn a_job_waits_for_approval_and_is_approved_rejected_or_expires() {

    let h = hub().await;

    let andy = db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();

    let bob = db::accounts::create_account(&h.app.db, "bob", "[email protected]", "x", "salt", false).await.unwrap();

    let carol = db::accounts::create_account(&h.app.db, "carol", "[email protected]", "x", "salt", true).await.unwrap();

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

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

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

    let c = client(&format!("{}/andy/gated", h.base), work.path());

    write(

        work.path(),

        ".flow/pipeline.kdl",

        &format!(

            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"plan\" {{\n    sandbox \"alpine\"\n    exec \"plan\"\n}}\njob \"apply\" {{\n    sandbox \"alpine\"\n    depends-on \"plan\"\n    approval {{\n        by \"maintain\"\n        others-only #true\n    }}\n    exec \"apply\"\n}}\n"

        ),

    );

    let mut n = 0;

    let mut push = || {

        n += 1;

        write(work.path(), "n.txt", &format!("{n}\n"));

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

        c.commit(&format!("Push {n}"), &[]).unwrap();

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

    };

    // Runs `plan` to a pass; returns the run, its actor set to andy.

    let plan_passes = |number: i64| {

        let app = h.app.clone();

        let row = row.clone();

        async move {

            let mut run = None;

            for _ in 0..100 {

                if let Some(r) = db::flow::run(&app.db, row.id, number).await.unwrap() {

                    run = Some(r);

                    break;

                }

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

            }

            let run = run.expect("the run");

            sqlx::query("UPDATE flow_runs SET actor = 'andy' WHERE id = ?").bind(run.id).execute(&app.db).await.unwrap();

            let j = claim(&app).await.expect("plan is claimed");

            assert_eq!(j["job"]["name"], "plan");

            let (id, ph) = (j["job_id"].as_i64().unwrap(), j["pipeline_hash"].as_str().unwrap().to_string());

            finish(&app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": receipt("plan", &ph, 0) })).await;

            db::flow::run(&app.db, row.id, number).await.unwrap().unwrap()

        }

    };

    // Run 1: apply waits; only someone with the role, other than the pusher, approves it.

    push();

    let run = runs_of(&h.app, "andy", "gated").await[0].clone();

    sqlx::query("UPDATE flow_runs SET actor = 'andy' WHERE id = ?").bind(run.id).execute(&h.app.db).await.unwrap();

    let early = oxygenhub::flow::decide(&h.app, &row, &run, "apply", &carol, true).await.unwrap_err();

    assert!(early.contains("waits for plan first"), "{early}");

    let run = plan_passes(run.number).await;

    assert!(claim(&h.app).await.is_none(), "apply waits for a person, so no runner gets it");

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

    assert!(page.contains("waits for approval by someone with the <em>maintain</em> role, other than andy"), "the run page says so");

    let own = oxygenhub::flow::decide(&h.app, &row, &run, "apply", &andy, true).await.unwrap_err();

    assert!(own.contains("someone other than andy"), "{own}");

    let no_role = oxygenhub::flow::decide(&h.app, &row, &run, "apply", &bob, true).await.unwrap_err();

    assert!(no_role.contains("takes the maintain role"), "{no_role}");

    assert_eq!(oxygenhub::flow::decide(&h.app, &row, &run, "apply", &carol, true).await.unwrap(), "apply: approved by carol");

    let j = claim(&h.app).await.expect("approved, it runs");

    assert_eq!(j["job"]["name"], "apply");

    let (id, ph) = (j["job_id"].as_i64().unwrap(), j["pipeline_hash"].as_str().unwrap().to_string());

    finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": receipt("apply", &ph, 0) })).await;

    let audit = db::accounts::audit_rows(&h.app.db, 50).await.unwrap();

    assert!(audit.iter().any(|r| r.action == "flow.approved" && r.actor == "carol" && r.target.ends_with("#1 apply")), "audited");

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

    assert!(page.contains("approved by carol"), "the run says who");

    // Run 2: rejected through the API's rules: the job ends cancelled, the run ends.

    push();

    let run = plan_passes(2).await;

    assert_eq!(oxygenhub::flow::decide(&h.app, &row, &run, "apply", &carol, false).await.unwrap(), "apply: rejected by carol");

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

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

    assert_eq!((apply.status.as_str(), apply.error.as_str()), ("cancelled", "rejected by carol"));

    assert!(claim(&h.app).await.is_none());

    // Run 3: nobody decides in time.

    push();

    let run = plan_passes(3).await;

    assert_eq!(oxygenhub::flow::expire_approvals(&h.app, db::now_ms() + 3_600_000).await.unwrap(), 0, "not yet");

    assert_eq!(oxygenhub::flow::expire_approvals(&h.app, db::now_ms() + 25 * 3_600_000).await.unwrap(), 1);

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

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

    assert_eq!((apply.status.as_str(), apply.error.as_str()), ("cancelled", "no approval within 24h"));

}

{% if !approvals.is_empty() %}<section class="panel approvals" aria-label="Approvals">

  <h2>Approvals</h2>

  <ul class="plain">{% for a in approvals %}<li class="approval st-{{ a.state }}"><strong>{{ a.job }}</strong>

    {% if a.state == "waiting" %}waits for approval by someone with the <em>{{ a.by }}</em> role{% if a.others_only %}, other than {{ run.actor }}{% endif %}.

      {% if a.may_decide() %}<form class="inline" method="post" action="{{ repo.url }}/flow/{{ run.number }}/jobs/{{ a.job }}/decide"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit" name="action" value="approve" class="small">APPROVE</button> <button type="submit" name="action" value="reject" class="small danger-btn">REJECT</button></form>

      {% else %}<span class="small mute">{{ a.why_not() }}</span>{% endif %}

    {% else if a.state == "approved" %}approved by {{ a.who }} <time title="{{ a.when_ms|when }}">{{ a.when_ms|ago }}</time>.

    {% else if a.state == "rejected" %}rejected by {{ a.who }} <time title="{{ a.when_ms|when }}">{{ a.when_ms|ago }}</time>.

    {% else %}not approved in time.{% endif %}</li>{% endfor %}</ul>

</section>{% endif %}