Flow state stamps: a state-addressed, verifiable pipeline result for READMEs, and a raw route so Markdown images work 736f6813543d

/// The newest run of exactly this state (heads joined with `+`, sorted).

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

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

        .bind(repo_id)

        .bind(heads)

        .fetch_optional(pool)

        .await

}

/// A file's name and bytes, and nothing else (for raw responses).

pub fn raw(repo: &Repo, cache: &StateCache, target: &RefTarget, path: &str) -> Result<(String, Vec<u8>)> {

    let loaded = cache.load(repo, &target.heads)?;

    let path = path.trim_matches('/');

    let Some(f) = loaded.snapshot.files.get(path) else { return Err(RepoError::NotFound) };

    let bytes = match (&f.kind, &f.content) {

        (NodeKind::Dir, _) | (_, FileContent::Dir) => return Err(RepoError::NotFound),

        (_, FileContent::Text { bytes, .. }) => bytes.clone(),

        (_, FileContent::Opaque { blobs }) => match blobs.first() {

            Some(b) => repo.read_blob(b)?,

            None => Vec::new(),

        },

    };

    Ok((path.rsplit('/').next().unwrap_or(path).to_string(), bytes))

}

// ------------------------------------------------------------ state stamps

/// In rendered Markdown, the image path that is the viewed state's stamp.

pub const STAMP_IMAGE: &str = ".flow/stamp";

/// In rendered Markdown, the link path to the viewed state's run.

pub const RUN_LINK: &str = ".flow/run";

/// Resolve a ref to its state address, and whether the ref can move.

async fn state_of(app: &Shared, owner: &str, name: &str, r: &str) -> Result<(String, bool), crate::repos::RepoError> {

    let repo = crate::repos::RepoName::new(owner, name).ok_or(crate::repos::RepoError::NotFound)?;

    let spec = r.to_string();

    app.repos

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

            let t = crate::repos::view::resolve(repo, cache, &spec)?;

            let moves = matches!(t.kind, crate::repos::view::RefKind::Bookmark | crate::repos::view::RefKind::Tag);

            Ok((t.permalink().unwrap_or_default(), moves))

        })

        .await

}

/// `GET /{owner}/{name}/flow/stamp/{ref}` (hub spec §13.2).

pub async fn stamp(

    State(app): State<Shared>,

    viewer: Viewer,

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

) -> Result<Response, Error> {

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

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

    let (heads, moves) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;

    if moves && !heads.is_empty() {

        let mut res = axum::response::Redirect::to(&format!("/{owner}/{name}/flow/stamp/{heads}")).into_response();

        res.headers_mut().insert(axum::http::header::CACHE_CONTROL, axum::http::HeaderValue::from_static("no-cache"));

        return Ok(res);

    }

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

    let jobs = match &run {

        Some(r) => fdb::jobs(&app.db, r.id).await.map_err(|e| Error::from_db(shell.clone(), e))?,

        None => vec![],

    };

    let svg = stamp_svg(&heads, run.as_ref(), &jobs);

    let cache = if run.as_ref().is_some_and(|r| !r.live()) { "public, max-age=3600" } else { "public, max-age=60" };

    Ok((

        [

            (axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"),

            (axum::http::header::CACHE_CONTROL, cache),

            (axum::http::header::CONTENT_SECURITY_POLICY, "default-src 'none'; style-src 'unsafe-inline'"),

        ],

        svg,

    )

        .into_response())

}

/// `GET /{owner}/{name}/flow/at/{ref}`: the run of the ref's state, or the Flow tab.

pub async fn at(

    State(app): State<Shared>,

    viewer: Viewer,

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

) -> Result<Response, Error> {

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

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

    let (heads, _) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;

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

    let to = match run {

        Some(r) => format!("/{owner}/{name}/flow/{}", r.id),

        None => format!("/{owner}/{name}/flow"),

    };

    Ok(axum::response::Redirect::to(&to).into_response())

}

fn xml(s: &str) -> String {

    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;").replace('"', "&quot;")

}

/// The stamp: the mark, one glyph per job, the status, the state and the

/// date; the same facts machine-readable in `<metadata>`. Fixed colours,

/// because an image does not follow the page's theme.

pub fn stamp_svg(heads: &str, run: Option<&fdb::Run>, jobs: &[fdb::Job]) -> String {

    const CH: f64 = 6.6; // IBM Plex Mono at 11px

    let colour = |s: &str| match s {

        "passed" => "#5fe3a1",

        "failed" | "error" => "#ff5e7a",

        "running" => "#00e0f0",

        _ => "#8592ad",

    };

    let short: String = heads.chars().take(12).collect();

    let (status, date) = match run {

        Some(r) => (r.status.as_str(), crate::render::time::absolute(r.created_ms as u64).chars().take(10).collect::<String>()),

        None => ("no run", String::new()),

    };

    let mut ordered: Vec<&fdb::Job> = jobs.iter().collect();

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

    let label = "OXYGENFLOW";

    let label_w = 14.0 + label.len() as f64 * 6.2 + 8.0;

    let glyphs_w = if ordered.is_empty() { 0.0 } else { ordered.len() as f64 * 10.0 + 6.0 };

    let tail = if date.is_empty() { format!("{status} · {short}") } else { format!("{status} · {short} · {date}") };

    let right_w = 8.0 + glyphs_w + tail.chars().count() as f64 * CH + 8.0;

    let w = (label_w + right_w).ceil();

    let mut glyphs = String::new();

    for (i, j) in ordered.iter().enumerate() {

        let x = label_w + 8.0 + i as f64 * 10.0 + 4.0;

        glyphs.push_str(&format!(

            r#"<text x="{x:.1}" y="15" fill="{}" text-anchor="middle">{}</text>"#,

            colour(&j.status),

            glyph(&j.status)

        ));

    }

    let meta = serde_json::json!({

        "heads": heads.split('+').filter(|h| !h.is_empty()).collect::<Vec<_>>(),

        "run": run.map(|r| r.id),

        "status": status,

        "jobs": ordered.iter().map(|j| serde_json::json!({ "name": j.name, "status": j.status, "receipt": j.receipt_hash })).collect::<Vec<_>>(),

    });

    let title = match run {

        Some(r) => format!("OxygenFlow run #{} {status} on state {short}", r.id),

        None => format!("OxygenFlow: no run for state {short}"),

    };

    format!(

        r##"<svg xmlns="http://www.w3.org/2000/svg" width="{w}" height="22" viewBox="0 0 {w} 22" role="img" aria-label="{t}"><title>{t}</title><metadata>{m}</metadata><rect width="{w}" height="22" rx="4" fill="#0e1526"/><rect width="{lw:.1}" height="22" rx="4" fill="#0b3f9e"/><rect x="{lx:.1}" width="4" height="22" fill="#0b3f9e"/><circle cx="9" cy="11" r="3.2" fill="none" stroke="#00f0ff" stroke-width="1.6"/><text x="16" y="15" fill="#ffffff" font-family="Barlow, Verdana, sans-serif" font-size="10.5" font-weight="700" letter-spacing=".6">{label}</text><g font-family="IBM Plex Mono, ui-monospace, monospace" font-size="11">{glyphs}<text x="{tx:.1}" y="15" fill="{sc}">{tail}</text></g><rect x=".5" y=".5" width="{wi}" height="21" rx="3.5" fill="none" stroke="#1f2b47"/></svg>"##,

        t = xml(&title),

        m = xml(&meta.to_string()),

        lw = label_w,

        lx = label_w - 4.0,

        tx = label_w + 8.0 + glyphs_w,

        sc = colour(status),

        tail = xml(&tail),

        wi = w - 1.0,

    )

}

    render::markdown::render(&linked, &format!("{repo_url}/file/main/"), &format!("{repo_url}/raw/main/"))

        .route("/{owner}/{name}/raw/{r}/{*path}", get(repo::raw_file))

        .route("/{owner}/{name}/flow/stamp/{r}", get(flow::stamp))

        .route("/{owner}/{name}/flow/at/{r}", get(flow::at))

            render::markdown::render(&md.text, &format!("{}/file/{r}/{dir}", rs.url), &format!("{}/raw/{r}/{dir}", rs.url))

        render::markdown::render(&md.text, &format!("{}/file/{r}/{dir}", rs.url), &format!("{}/raw/{r}/{dir}", rs.url))

    if path == super::flow::RUN_LINK && !has_file(&app, &repo, &r, &path).await {

        return Ok(axum::response::Redirect::to(&format!("/{owner}/{name}/flow/at/{r}")).into_response());

    }

                let (base, raw) = (format!("{}/file/{r}/{dir}", rs.url), format!("{}/raw/{r}/{dir}", rs.url));

                (vec![], highlight::language(&f.path), Some(render::markdown::render(text, &base, &raw)))

}

/// Whether `path` is a file in the state `r` names.

async fn has_file(app: &Shared, repo: &crate::repos::RepoName, r: &str, path: &str) -> bool {

    let (spec, p) = (r.to_string(), path.to_string());

    app.repos

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

            let target = view::resolve(repo, cache, &spec)?;

            Ok(cache.load(repo, &target.heads)?.snapshot.files.get(&p).is_some_and(|f| f.kind != oxy_core::objects::NodeKind::Dir))

        })

        .await

        .unwrap_or(false)

}

/// `GET /{owner}/{name}/raw/{ref}/{path}`: a file's bytes (hub spec §6.3),

/// where relative images in rendered Markdown point. `.flow/stamp` is the

/// state stamp of `{ref}` unless a real file is there (§13.2).

pub async fn raw_file(

    State(app): State<Shared>,

    viewer: Viewer,

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

) -> Result<Response, Error> {

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

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

    if path == super::flow::STAMP_IMAGE && !has_file(&app, &repo, &r, &path).await {

        return Ok(axum::response::Redirect::to(&format!("/{owner}/{name}/flow/stamp/{r}")).into_response());

    }

    let (spec, p) = (r.clone(), path.clone());

    let (file_name, bytes) = app

        .repos

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

            let target = view::resolve(repo, cache, &spec)?;

            view::raw(repo, cache, &target, &p)

        })

        .await

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

    Ok(super::raw(bytes, &file_name, true))

/// A GET's status, `Location` and `Content-Type`.

async fn head(app: &Shared, path: &str) -> (StatusCode, String, String) {

    let res = oxygenhub::router(app.clone()).oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();

    let h = |k: header::HeaderName| res.headers().get(k).and_then(|v| v.to_str().ok()).unwrap_or("").to_string();

    (res.status(), h(header::LOCATION), h(header::CONTENT_TYPE))

}

    // State stamps (hub spec §13.2) and images in READMEs.

    write(work.path(), "README.md", "# Demo\n\n[![flow](.flow/stamp)](.flow/run)\n\n![logo](img/logo.png)\n");

    std::fs::create_dir_all(work.path().join("img")).unwrap();

    std::fs::write(work.path().join("img/logo.png"), b"\x89PNG\r\n\x1a\n\0\0\0\rIHDR\0\0\0\x01\0\0\0\x01\x08\x06\0\0\0\x1f\x15\xc4\x89")

        .unwrap();

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

    c.commit("A README with a stamp", &[]).unwrap();

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

    let mut fourth = None;

    for _ in 0..100 {

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

        if r.len() == 4 {

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

            break;

        }

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

    }

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

    let (_, overview) = call(&h.app, "GET", "/andy/demo", None, None).await;

    assert!(overview.contains("src=\"/andy/demo/raw/main/.flow/stamp\""), "{overview}");

    assert!(overview.contains("href=\"/andy/demo/file/main/.flow/run\""));

    assert!(overview.contains("src=\"/andy/demo/raw/main/img/logo.png\""), "images use the raw route");

    let (st, _, ct) = head(&h.app, "/andy/demo/raw/main/img/logo.png").await;

    assert_eq!((st, ct.as_str()), (StatusCode::OK, "image/png"));

    let (st, loc, _) = head(&h.app, "/andy/demo/raw/main/.flow/stamp").await;

    assert!(st.is_redirection() && loc == "/andy/demo/flow/stamp/main", "{st} {loc}");

    let (st, loc, _) = head(&h.app, "/andy/demo/flow/stamp/main").await;

    assert!(st.is_redirection() && loc == format!("/andy/demo/flow/stamp/{}", fourth.heads), "a bookmark's stamp names its state: {loc}");

    let (st, _, ct) = head(&h.app, &loc).await;

    assert_eq!((st, ct.as_str()), (StatusCode::OK, "image/svg+xml; charset=utf-8"));

    let (_, svg) = call(&h.app, "GET", &loc, None, None).await;

    assert!(svg.starts_with("<svg") && svg.contains("error ·") && svg.contains(&fourth.heads[..12]), "{svg}");

    let (_, svg1) = call(&h.app, "GET", &format!("/andy/demo/flow/stamp/{}", run.heads), None, None).await;

    assert!(svg1.contains("failed ·") && svg1.contains(&format!("run #{} failed", run.id)), "an older state keeps its own result: {svg1}");

    assert!(svg1.contains(&build.receipt_hash), "receipt hashes in the metadata");

    assert_eq!(svg1.matches("text-anchor=\"middle\"").count(), 3, "one glyph per job");

    let (st, loc, _) = head(&h.app, "/andy/demo/file/main/.flow/run").await;

    assert!(st.is_redirection() && loc == "/andy/demo/flow/at/main", "{loc}");

    let (st, loc, _) = head(&h.app, "/andy/demo/flow/at/main").await;

    assert!(st.is_redirection() && loc == format!("/andy/demo/flow/{}", fourth.id), "{loc}");

    let (_, loc, _) = head(&h.app, "/other/plain/flow/stamp/main").await;

    let (_, svg) = call(&h.app, "GET", &loc, None, None).await;

    assert!(svg.contains("no run ·"), "{svg}");

    let (_, loc, _) = head(&h.app, "/other/plain/flow/at/main").await;

    assert_eq!(loc, "/other/plain/flow");

| `/{owner}/{name}/raw/{ref}/{path}` | A file's bytes, as `?plain=1` serves them (images inline with a sniffed type, anything else as an attachment); relative images in rendered Markdown point here |

**State stamps** replace CI badges, which are mutable images claiming that a branch passes, name no state and cannot be checked. A stamp belongs to a state:

- `/{owner}/{name}/flow/stamp/{ref}` is an SVG. For a state address (heads joined with `+`) or a change, it shows the newest run of exactly that state: one glyph per job in dependency order, the run's status, the state's short hash and the run's date. It shows "no run" when there is none. The same data is machine-readable in the SVG's `<metadata>`: heads, run number, status, and each job's status and receipt hash. For a bookmark or tag, it redirects to the stamp of the ref's current state, so even the image names what was checked. A finished run's stamp may be cached for an hour; any other stamp for a minute.

- In rendered Markdown, the relative paths `.flow/stamp` (as an image) and `.flow/run` (as a link) resolve against the state being viewed. The first is served by `/raw/{ref}/.flow/stamp`, the second by `/file/{ref}/.flow/run` (the run page, or the Flow tab when that state has no run). So `[![flow](.flow/stamp)](.flow/run)` in a README shows how the viewed state did: browsing a tag shows the tag's result, not the bookmark's. A real file at either path takes precedence.

- Later, stamps also say which runner key sealed the result (once runner seals exist) and which jobs reused an earlier receipt.