Flow runs, jobs and logs in the API; checks by heads 90326c9601f9

/// An OxygenFlow run (hub spec §13.2).

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]

pub struct FlowRun {

    /// The run's number within its repository.

    pub number: i64,

    pub bookmark: String,

    /// The state's hash, as checks are keyed.

    pub state: String,

    /// Its heads, full change hashes.

    pub heads: Vec<String>,

    /// `queued`, `running`, `passed`, `failed` or `error` (an invalid pipeline).

    pub status: String,

    /// `push`, `manual` or `rerun:N`.

    pub trigger: String,

    pub actor: String,

    pub pipeline_hash: String,

    /// An invalid pipeline's problems.

    pub error: String,

    pub created_ms: i64,

    pub finished_ms: Option<i64>,

    pub url: String,

    /// Filled by the single-run endpoint.

    #[serde(default, skip_serializing_if = "Vec::is_empty")]

    pub jobs: Vec<FlowJob>,

}

/// A job of a run.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]

pub struct FlowJob {

    pub name: String,

    /// `queued`, `running`, `passed`, `failed`, `skipped` or `error`.

    pub status: String,

    pub depends_on: Vec<String>,

    pub runner: String,

    pub exit_code: Option<i64>,

    pub error: String,

    pub started_ms: Option<i64>,

    pub finished_ms: Option<i64>,

    /// Log lines stored so far.

    pub log_lines: i64,

}

/// A line of a job's log.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]

pub struct FlowLogLine {

    /// Increasing within the job; pass the last one as `after` for more.

    pub seq: i64,

    /// Milliseconds since the job started.

    pub t: i64,

    /// `stdout`, `stderr` or `flow` (the runner's own notes).

    pub stream: String,

    pub text: String,

}

/// Any other path under `/api/v1`: a problem document, not the site's page.

async fn no_route() -> ApiError {

    ApiError(StatusCode::NOT_FOUND, "no such API route".into())

}

        .fallback(no_route)

        .route("/repos/{owner}/{name}/flow/runs", get(flow_runs).post(start_flow_run))

        .route("/repos/{owner}/{name}/flow/runs/{number}", get(flow_run))

        .route("/repos/{owner}/{name}/flow/runs/{number}/jobs/{job}/log", get(flow_job_log))

fn flow_run_of(app: &Shared, row: &db::repos::RepoRow, r: &db::flow::Run, jobs: Vec<db::flow::Job>) -> FlowRun {

    FlowRun {

        number: r.number,

        bookmark: r.bookmark.clone(),

        state: r.state.clone(),

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

        status: r.status.clone(),

        trigger: r.trigger.clone(),

        actor: r.actor.clone(),

        pipeline_hash: r.pipeline_hash.clone(),

        error: r.error.clone(),

        created_ms: r.created_ms,

        finished_ms: r.finished_ms,

        url: format!("{}/{}/{}/flow/{}", app.config.base_url(), row.owner, row.name, r.number),

        jobs: jobs

            .into_iter()

            .map(|j| FlowJob {

                depends_on: j.depends_on(),

                name: j.name,

                status: j.status,

                runner: j.runner,

                exit_code: j.exit_code,

                error: j.error,

                started_ms: j.started_ms,

                finished_ms: j.finished_ms,

                log_lines: j.log_count,

            })

            .collect(),

    }

}

struct RunsPage {

    /// Runs numbered below this one (the next page).

    before: Option<i64>,

    limit: Option<i64>,

}

/// `GET /repos/{o}/{n}/flow/runs`: newest first, `?before=N&limit=` (max 100).

async fn flow_runs(

    State(app): State<Shared>,

    headers: HeaderMap,

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

    Query(q): Query<RunsPage>,

) -> Result<Json<Vec<FlowRun>>, ApiError> {

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

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

    let limit = q.limit.unwrap_or(30).clamp(1, 100);

    let runs = db::flow::runs_by_number(&app.db, row.id, limit, q.before).await?;

    Ok(Json(runs.iter().map(|r| flow_run_of(&app, &row, r, vec![])).collect()))

}

/// `GET /repos/{o}/{n}/flow/runs/{number}`: a run with its jobs.

async fn flow_run(

    State(app): State<Shared>,

    headers: HeaderMap,

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

) -> Result<Json<FlowRun>, ApiError> {

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

    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 jobs = db::flow::jobs(&app.db, run.id).await?;

    Ok(Json(flow_run_of(&app, &row, &run, jobs)))

}

#[derive(Deserialize)]

struct LogPage {

    /// Lines after this `seq`.

    #[serde(default)]

    after: i64,

    limit: Option<i64>,

}

/// `GET /repos/{o}/{n}/flow/runs/{number}/jobs/{job}/log`: `?after=seq&limit=` (max 5000).

async fn flow_job_log(

    State(app): State<Shared>,

    headers: HeaderMap,

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

    Query(q): Query<LogPage>,

) -> Result<Json<Vec<FlowLogLine>>, ApiError> {

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

    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 j = db::flow::jobs(&app.db, run.id)

        .await?

        .into_iter()

        .find(|j| j.name == job)

        .ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such job".into()))?;

    let lines = db::flow::log(&app.db, j.id, q.after, q.limit.unwrap_or(1000).clamp(1, 5000)).await?;

    Ok(Json(lines.into_iter().map(|l| FlowLogLine { seq: l.seq, t: l.t, stream: l.stream, text: l.text }).collect()))

}

#[derive(Deserialize)]

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

    let state = state_key(&app, &repo, &state).await?;

}

/// The state hash checks are keyed by, from what a client has: a state

/// hash, its heads (full change hashes joined with `+`, or the one head of

/// a single-head state, which is what `oxy log` shows), or a bookmark.

async fn state_key(app: &Shared, repo: &RepoName, s: &str) -> Result<String, ApiError> {

    use oxy_core::hash::{hash_object, ChangeHash, ObjectType};

    let hexes: Vec<&str> = s.split('+').collect();

    let all_hashes = hexes.iter().all(|h| h.len() == 64 && h.bytes().all(|b| b.is_ascii_hexdigit()));

    if all_hashes {

        let row = db::repos::get(&app.db, &repo.owner, &repo.name).await?.ok_or_else(|| not_found("no such repository"))?;

        if hexes.len() == 1 && !db::hooks::checks_for_state(&app.db, row.id, s).await?.is_empty() {

            return Ok(s.to_string());

        }

        let heads: Vec<ChangeHash> = hexes.iter().filter_map(|h| ChangeHash::from_hex(h)).collect();

        return Ok(hash_object(ObjectType::State, &oxy_core::objects::StateObject::new(heads).encode()).to_hex());

    }

    let b = s.to_string();

    let found = app.repos.read(repo, move |r, _| Ok(r.view()?.bookmarks.get(&b).copied())).await?;

    found.map(|st| st.to_hex()).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{s:?} is not a state hash, heads or a bookmark")))

/// A repository's runs, newest first, numbered below `before`.

pub async fn runs_by_number(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 number < ? ORDER BY number DESC LIMIT ?")

        .bind(repo_id)

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

        .bind(limit)

        .fetch_all(pool)

        .await

}

/// The checks on a state once there are `n` of them (they follow the run).

async fn checks_when(app: &Shared, repo_id: i64, state: &str, n: usize) -> Vec<db::hooks::Check> {

    for _ in 0..100 {

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

        if c.len() >= n {

            return c;

        }

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

    }

    db::hooks::checks_for_state(&app.db, repo_id, state).await.unwrap()

}

    // The checks are reported just after the run is recorded.

    let checks = checks_when(&h.app, run.repo_id, &run.state, 3).await;

    let checks = checks_when(&h.app, row.id, &plain[0].state, 3).await;

}

/// The Flow API (hub spec §13.2): runs, a run with its jobs, a job's log,

/// checks by the heads a client has, and JSON for unknown API routes.

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

async fn runs_jobs_and_logs_are_in_the_api() {

    let h = hub().await;

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

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

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

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

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

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

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

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

    let run = runs_of(&h.app, "andy", "demo").await.remove(0);

    // A job runs and logs.

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

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

    let recs =

        serde_json::json!({ "records": [{ "t": 1, "stream": "stdout", "text": "one" }, { "t": 3, "stream": "stderr", "text": "three" }] });

    let (st, _) = call(&h.app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(recs.to_string())).await;

    assert!(st.is_success(), "{st}");

    let (st, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs", None, None).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let runs: Vec<hub_api::FlowRun> = serde_json::from_str(&body).unwrap();

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

    assert_eq!((runs[0].number, runs[0].state.as_str(), runs[0].trigger.as_str()), (1, run.state.as_str(), "push"));

    assert!(runs[0].url.ends_with("/andy/demo/flow/1") && runs[0].jobs.is_empty());

    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", None, None).await;

    let one: hub_api::FlowRun = serde_json::from_str(&body).unwrap();

    let names: Vec<(&str, &str)> = one.jobs.iter().map(|j| (j.name.as_str(), j.status.as_str())).collect();

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

    let running = one.jobs.iter().find(|j| j.status == "running").unwrap();

    assert_eq!(running.name, a["job"]["name"].as_str().unwrap());

    assert!(one.jobs.iter().any(|j| j.name == "test" && j.depends_on == ["build"]));

    let job = running.name.clone();

    let (st, body) = call(&h.app, "GET", &format!("/api/v1/repos/andy/demo/flow/runs/1/jobs/{job}/log"), None, None).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let lines: Vec<hub_api::FlowLogLine> = serde_json::from_str(&body).unwrap();

    assert!(lines.iter().any(|l| l.text == "three"), "{body}");

    let last = lines.last().unwrap().seq;

    let (_, body) = call(&h.app, "GET", &format!("/api/v1/repos/andy/demo/flow/runs/1/jobs/{job}/log?after={last}"), None, None).await;

    assert_eq!(body, "[]", "nothing after the last line");

    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/9", None, None).await.0, StatusCode::NOT_FOUND);

    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1/jobs/nope/log", None, None).await.0, StatusCode::NOT_FOUND);

    // Checks by the state hash, by the head `oxy log` shows, and by bookmark.

    let by = |key: String| {

        let app = h.app.clone();

        async move {

            let (st, body) = call(&app, "GET", &format!("/api/v1/repos/andy/demo/checks/{key}"), None, None).await;

            assert_eq!(st, StatusCode::OK, "{key}: {body}");

            serde_json::from_str::<Vec<hub_api::Check>>(&body).unwrap().len()

        }

    };

    checks_when(&h.app, run.repo_id, &run.state, 3).await;

    assert_eq!(by(run.state.clone()).await, 3);

    assert_eq!(by(head.to_hex()).await, 3, "a single head names its state");

    assert_eq!(by("main".into()).await, 3);

    let (st, _) = call(&h.app, "GET", "/api/v1/repos/andy/demo/checks/no-such-bookmark", None, None).await;

    assert_eq!(st, StatusCode::NOT_FOUND);

    // An unknown API route answers with a problem document.

    let (st, body) = call(&h.app, "GET", "/api/v1/users/andy/repos", None, None).await;

    assert_eq!(st, StatusCode::NOT_FOUND);

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

    assert_eq!(v["status"], 404, "{body}");

JSON under `/api/v1`, bearer tokens, `application/json`, RFC 7807 problem responses (an unknown route too), cursor pagination (`?after=`), ETags on cacheable resources, and rate limits per token. Resources mirror the pages: `/repos/{owner}/{name}`, `/bookmarks`, `/changes/{hash}`, `/tree/{ref}/{path}`, `/file/{ref}/{path}`, `/blame`, `/history`, `/compare`, `/conflicts`, `/proposals`, `/proposals/{n}/comments|reviews|checks|files|land`, `/proposals/{n}/content/{base|head}/{path}` (a changed file's bytes in the target's state or the projected state, for editor diff views), `/issues` (creating an issue can set its labels, assignees and milestone; `PATCH /issues/{n}` edits the title and body for the author or triage, and replaces labels, assignees or the milestone for triage, recording each on the timeline; unknown names are refused before anything changes), `/labels` (`POST` creates or changes one, `DELETE /labels/{name}` removes it; triage), `/milestones`, `/releases`, `/tags`, `/trust`, `/pushes`, `/hooks`, `/checks`, `/user`, `/notifications` (the caller's inbox and unread count; `/notifications/{id}/read` and `/notifications/read` mark read; scope `user:read`), `/users/{name}`, `/orgs/{name}`, `/checks/{state}` (the checks on a state, named by its hash, by its heads joined with `+`, by the one head of a single-head state, or by a bookmark), `/checks/required` (`GET`, and `PUT` of the whole list by a repository admin: the checks a proposal needs, passed, to land), `/projects/{owner}` (list, create) and `/projects/{owner}/{n}` (the board as the caller may see it, `PATCH` for title, description, privacy and closing, `/cards` to add a reference or note to a column named by id or name, `/cards/{id}` to move or remove one; `repo:write` and the page's rights to change), `/flow/runs` (`GET`: runs newest first, `?before=N&limit=`; `POST {"bookmark"}`, repo:write: run a bookmark's current state, the default bookmark when none is named), `/flow/runs/{n}` (a run with its jobs), `/flow/runs/{n}/jobs/{job}/log` (`?after=seq&limit=`) and `/flow/runs/{n}/rerun` (repo:write: a finished run again), both answering `201 {"number", "url"}` or `409` with why nothing started, `/search`. Shapes reuse `oxy --json` shapes wherever one exists (change, status, blame, diff) so tools written against the CLI work against the hub. `crates/hub-api` holds the types with `serde` derives and is what the runner and the VS Code extension will link.