Reports: keep a job's report file by file, index its declaration, metrics and files a36bab6b9ffc

-- Reports (oxygenhub#127, docs/reports.md): a directory a Flow job writes

-- with report.toml at its top. Its files are blobs of the repository and

-- its manifest is the job's artifact of that name; these tables index it

-- for the Reports tab, its history and checks, and name every file blob

-- so garbage collection keeps it.

CREATE TABLE reports (

    id          INTEGER PRIMARY KEY,

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

    name        TEXT NOT NULL,

    kind        TEXT NOT NULL,

    title       TEXT NOT NULL,

    summary     TEXT NOT NULL,

    context     TEXT NOT NULL,

    run_id      INTEGER REFERENCES flow_runs(id) ON DELETE CASCADE,

    job_id      INTEGER,

    run_number  INTEGER,

    job         TEXT NOT NULL,

    bookmark    TEXT NOT NULL,

    state       TEXT NOT NULL,

    manifest    TEXT,

    files       INTEGER NOT NULL,

    bytes       INTEGER NOT NULL,

    created_ms  INTEGER NOT NULL

);

CREATE UNIQUE INDEX reports_job ON reports (job_id, name);

CREATE INDEX reports_history ON reports (repo_id, name, bookmark, created_ms);

CREATE TABLE report_metrics (

    report_id   INTEGER NOT NULL REFERENCES reports(id) ON DELETE CASCADE,

    key         TEXT NOT NULL,

    value       REAL NOT NULL,

    unit        TEXT NOT NULL,

    better      TEXT NOT NULL,

    fail_above  REAL,

    fail_below  REAL,

    PRIMARY KEY (report_id, key)

);

CREATE TABLE report_files (

    report_id   INTEGER NOT NULL REFERENCES reports(id) ON DELETE CASCADE,

    path        TEXT NOT NULL,

    blob        TEXT NOT NULL,

    size        INTEGER NOT NULL,

    PRIMARY KEY (report_id, path)

);

CREATE INDEX report_files_blob ON report_files (blob);

//! Reports (oxygenhub#127, docs/reports.md): a directory a Flow job writes

//! with `report.toml` at its top. The runner uploads each file as a blob of

//! the run's repository, then a manifest naming them as the job's artifact

//! under the report's name. The hub checks the manifest, reads the

//! declaration and indexes the report, its metrics and its files, so the

//! Reports tab, its history and checks read only the database, and garbage

//! collection keeps every file a kept report names.

use crate::db::{now_ms, Pool};

use crate::Shared;

use serde::Deserialize;

/// Files in one report, and their total size; the runner keeps to these.

pub const MAX_FILES: usize = 2000;

pub const MAX_BYTES: u64 = 512 << 20;

/// Largest `report.toml` read.

const MAX_DECLARATION: usize = 64 << 10;

/// Metrics kept from one report.

const MAX_METRICS: usize = 200;

/// The kinds a report may declare.

pub const KINDS: [&str; 4] = ["coverage", "tests", "performance", "custom"];

/// A report file's path: `/`-separated names of letters, digits, `.`,

/// `_`, `-` and `+`, never `.` or `..`, so it travels in a header and a URL

/// as it is.

pub fn path_ok(p: &str) -> bool {

    p.len() <= 1024

        && p.split('/').count() <= 32

        && p.split('/').all(|n| {

            !n.is_empty() && n != "." && n != ".." && n.len() <= 255 && n.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-+".contains(&b))

        })

}

/// A report's manifest, as the runner uploads it.

#[derive(Clone, Debug, Deserialize)]

pub struct Manifest {

    pub version: u32,

    pub files: Vec<ManifestFile>,

}

#[derive(Clone, Debug, Deserialize)]

pub struct ManifestFile {

    pub path: String,

    pub blob: String,

    pub size: u64,

}

impl Manifest {

    pub fn bytes(&self) -> u64 {

        self.files.iter().map(|f| f.size).sum()

    }

    /// The blob of the file at `path`.

    pub fn blob(&self, path: &str) -> Option<&str> {

        self.files.iter().find(|f| f.path == path).map(|f| f.blob.as_str())

    }

}

/// Read and check a manifest: version 1, clean unique paths, blob names,

/// within the limits, with `report.toml`.

pub fn parse_manifest(bytes: &[u8]) -> Result<Manifest, String> {

    let m: Manifest = serde_json::from_slice(bytes).map_err(|e| format!("the report manifest is not valid: {e}"))?;

    if m.version != 1 {

        return Err(format!("report manifest version {} is not known; this hub reads version 1", m.version));

    }

    if m.files.len() > MAX_FILES {

        return Err(format!("the report holds {} files; at most {MAX_FILES}", m.files.len()));

    }

    if m.bytes() > MAX_BYTES {

        return Err(format!("the report holds {} MiB; at most {} MiB", m.bytes() >> 20, MAX_BYTES >> 20));

    }

    let mut seen = std::collections::HashSet::new();

    for f in &m.files {

        if !path_ok(&f.path) {

            return Err(format!("report file {:?}: names may hold letters, digits, '.', '_', '-' and '+'", f.path));

        }

        if !seen.insert(f.path.as_str()) {

            return Err(format!("report file {} is named twice", f.path));

        }

        if oxy_core::hash::ObjectRef::from_hex(&f.blob).is_none() {

            return Err(format!("report file {}: {:?} is not a blob name", f.path, f.blob));

        }

    }

    if m.blob("report.toml").is_none() {

        return Err("the report has no report.toml at its top".into());

    }

    Ok(m)

}

/// What `report.toml` declares.

#[derive(Clone, Debug, PartialEq)]

pub struct Declaration {

    pub kind: String,

    pub title: String,

    pub summary: String,

    /// The data file of a built-in kind (`lcov.info`, `junit.xml`).

    pub file: Option<String>,

    pub metrics: Vec<Metric>,

    /// The free-form `[context]` table, as JSON.

    pub context: serde_json::Value,

}

/// A number the hub tracks across runs.

#[derive(Clone, Debug, PartialEq, sqlx::FromRow)]

pub struct Metric {

    pub key: String,

    pub value: f64,

    pub unit: String,

    /// `higher`, `lower`, or empty when neither is better.

    pub better: String,

    pub fail_above: Option<f64>,

    pub fail_below: Option<f64>,

}

impl Declaration {

    /// The declaration of a report whose `report.toml` could not be read:

    /// kept as a custom report saying why, so its files stay and the reader

    /// sees what to fix.

    pub fn unreadable(name: &str, why: &str) -> Declaration {

        Declaration {

            kind: "custom".into(),

            title: name.into(),

            summary: format!("report.toml: {why}"),

            file: None,

            metrics: vec![],

            context: serde_json::json!({}),

        }

    }

}

fn number(v: &toml::Value) -> Option<f64> {

    match v {

        toml::Value::Integer(i) => Some(*i as f64),

        toml::Value::Float(f) if f.is_finite() => Some(*f),

        _ => None,

    }

}

fn metric_key_ok(k: &str) -> bool {

    !k.is_empty() && k.len() <= 64 && k.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b))

}

/// Read `report.toml` for report `name`; the error says what to fix.

pub fn parse_declaration(name: &str, bytes: &[u8]) -> Result<Declaration, String> {

    if bytes.len() > MAX_DECLARATION {

        return Err(format!("{} KiB; at most {} KiB", bytes.len() >> 10, MAX_DECLARATION >> 10));

    }

    let text = std::str::from_utf8(bytes).map_err(|_| "not UTF-8 text".to_string())?;

    let table: toml::Table = toml::from_str(text).map_err(|e| {

        let line = e.span().map(|s| text[..s.start.min(text.len())].lines().count().max(1));

        let what = if e.message().trim().is_empty() { "not valid TOML" } else { e.message().trim() };

        match line {

            Some(n) => format!("line {n}: {what}"),

            None => what.to_string(),

        }

    })?;

    let text_of = |key: &str, max: usize| -> Result<Option<String>, String> {

        match table.get(key) {

            None => Ok(None),

            Some(toml::Value::String(s)) if s.len() <= max => Ok(Some(s.clone())),

            Some(toml::Value::String(_)) => Err(format!("`{key}` is longer than {max} bytes")),

            Some(_) => Err(format!("`{key}` is a string")),

        }

    };

    for key in table.keys() {

        if !matches!(key.as_str(), "kind" | "title" | "summary" | "file" | "metrics" | "context") {

            return Err(format!("`{key}` is not known (kind, title, summary, file, metrics, context)"));

        }

    }

    let kind = text_of("kind", 32)?.ok_or("`kind` is required: one of coverage, tests, performance, custom")?;

    if !KINDS.contains(&kind.as_str()) {

        return Err(format!("kind {kind:?} is not known: one of coverage, tests, performance, custom"));

    }

    let file = text_of("file", 1024)?;

    if let Some(f) = &file {

        if !path_ok(f) {

            return Err(format!("file {f:?} is not a path in the report"));

        }

    }

    let mut metrics = Vec::new();

    match table.get("metrics") {

        None => {}

        Some(toml::Value::Table(t)) => {

            if t.len() > MAX_METRICS {

                return Err(format!("{} metrics; at most {MAX_METRICS}", t.len()));

            }

            for (key, v) in t {

                if !metric_key_ok(key) {

                    return Err(format!("metric {key:?}: names hold letters, digits, '.', '_' and '-'"));

                }

                metrics.push(parse_metric(key, v).map_err(|e| format!("metric {key}: {e}"))?);

            }

        }

        Some(_) => return Err("`metrics` is a table".into()),

    }

    let context = match table.get("context") {

        None => serde_json::json!({}),

        Some(v @ toml::Value::Table(_)) => serde_json::to_value(v).map_err(|e| format!("context: {e}"))?,

        Some(_) => return Err("`context` is a table".into()),

    };

    Ok(Declaration {

        kind,

        title: text_of("title", 200)?.unwrap_or_else(|| name.to_string()),

        summary: text_of("summary", 1000)?.unwrap_or_default(),

        file,

        metrics,

        context,

    })

}

fn parse_metric(key: &str, v: &toml::Value) -> Result<Metric, String> {

    let mut m = Metric { key: key.into(), value: 0.0, unit: String::new(), better: String::new(), fail_above: None, fail_below: None };

    if let Some(n) = number(v) {

        m.value = n;

        return Ok(m);

    }

    let toml::Value::Table(t) = v else {

        return Err("a number, or a table with `value`".into());

    };

    for (k, v) in t {

        match k.as_str() {

            "value" => m.value = number(v).ok_or("`value` is a number")?,

            "unit" => match v {

                toml::Value::String(s) if s.len() <= 32 => m.unit = s.clone(),

                _ => return Err("`unit` is a short string".into()),

            },

            "better" => match v.as_str() {

                Some(s @ ("higher" | "lower")) => m.better = s.into(),

                _ => return Err("`better` is \"higher\" or \"lower\"".into()),

            },

            "fail_above" => m.fail_above = Some(number(v).ok_or("`fail_above` is a number")?),

            "fail_below" => m.fail_below = Some(number(v).ok_or("`fail_below` is a number")?),

            _ => return Err(format!("`{k}` is not known (value, unit, better, fail_above, fail_below)")),

        }

    }

    if !t.contains_key("value") {

        return Err("`value` is required".into());

    }

    Ok(m)

}

/// Where a report came from.

pub struct Origin<'a> {

    pub repo_id: i64,

    pub run_id: i64,

    pub run_number: i64,

    pub job_id: i64,

    pub job: &'a str,

    pub bookmark: &'a str,

    pub state: &'a str,

}

/// Index report `name` of a job, replacing any the job kept before.

pub async fn keep(app: &Shared, o: &Origin<'_>, name: &str, d: &Declaration, m: &Manifest, manifest: &str) -> sqlx::Result<i64> {

    let mut tx = crate::db::begin_write(&app.db).await?;

    sqlx::query("DELETE FROM reports WHERE job_id = ? AND name = ?").bind(o.job_id).bind(name).execute(&mut *tx).await?;

    let id: i64 = sqlx::query_scalar(

        "INSERT INTO reports (repo_id, name, kind, title, summary, context, run_id, job_id, run_number, job, bookmark, state, \

         manifest, files, bytes, created_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",

    )

    .bind(o.repo_id)

    .bind(name)

    .bind(&d.kind)

    .bind(&d.title)

    .bind(&d.summary)

    .bind(d.context.to_string())

    .bind(o.run_id)

    .bind(o.job_id)

    .bind(o.run_number)

    .bind(o.job)

    .bind(o.bookmark)

    .bind(o.state)

    .bind(manifest)

    .bind(m.files.len() as i64)

    .bind(m.bytes() as i64)

    .bind(now_ms())

    .fetch_one(&mut *tx)

    .await?;

    for x in &d.metrics {

        sqlx::query(

            "INSERT INTO report_metrics (report_id, key, value, unit, better, fail_above, fail_below) VALUES (?, ?, ?, ?, ?, ?, ?)",

        )

        .bind(id)

        .bind(&x.key)

        .bind(x.value)

        .bind(&x.unit)

        .bind(&x.better)

        .bind(x.fail_above)

        .bind(x.fail_below)

        .execute(&mut *tx)

        .await?;

    }

    for f in &m.files {

        sqlx::query("INSERT INTO report_files (report_id, path, blob, size) VALUES (?, ?, ?, ?)")

            .bind(id)

            .bind(&f.path)

            .bind(&f.blob)

            .bind(f.size as i64)

            .execute(&mut *tx)

            .await?;

    }

    tx.commit().await?;

    Ok(id)

}

/// Report `id`'s metrics, by key.

pub async fn metrics(pool: &Pool, id: i64) -> sqlx::Result<Vec<Metric>> {

    sqlx::query_as("SELECT key, value, unit, better, fail_above, fail_below FROM report_metrics WHERE report_id = ? ORDER BY key")

        .bind(id)

        .fetch_all(pool)

        .await

}

/// Every file blob a repository's kept reports name, for garbage collection

/// to keep too: the database is the only thing that names them.

pub async fn file_blobs(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {

    sqlx::query_scalar("SELECT DISTINCT f.blob FROM report_files f JOIN reports r ON r.id = f.report_id WHERE r.repo_id = ?")

        .bind(repo_id)

        .fetch_all(pool)

        .await

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn declarations_are_read_with_their_metrics_and_context() {

        let d = parse_declaration(

            "frame-timing",

            br#"

kind    = "performance"

title   = "Frame timing"

summary = "Median frame 16.4 ms"

[metrics]

median_frame_ms = { value = 16.4, unit = "ms", better = "lower" }

p99_frame_ms    = { value = 18.1, unit = "ms", better = "lower", fail_above = 20 }

scenes          = 12

[context]

target_fps    = 30

budget_cycles = 268800

"#,

        )

        .unwrap();

        assert_eq!((d.kind.as_str(), d.title.as_str()), ("performance", "Frame timing"));

        let keys: Vec<&str> = d.metrics.iter().map(|m| m.key.as_str()).collect();

        assert_eq!(keys, ["median_frame_ms", "p99_frame_ms", "scenes"]);

        assert_eq!(d.metrics[1].fail_above, Some(20.0));

        assert_eq!((d.metrics[2].value, d.metrics[2].unit.as_str()), (12.0, ""));

        assert_eq!(d.context["budget_cycles"], 268800);

        let minimal = parse_declaration("cov", b"kind = \"coverage\"\nfile = \"lcov.info\"\n").unwrap();

        assert_eq!((minimal.title.as_str(), minimal.file.as_deref()), ("cov", Some("lcov.info")));

        let bad = |t: &str| parse_declaration("r", t.as_bytes()).unwrap_err();

        assert!(bad("title = \"x\"").contains("`kind` is required"));

        assert!(bad("kind = \"graphs\"").contains("not known"));

        assert!(bad("kind = \"custom\"\ntitel = \"x\"").contains("`titel` is not known"));

        assert!(bad("kind = \"custom\"\n[metrics]\nfps = \"fast\"").contains("metric fps"));

        assert!(bad("kind = \"custom\"\n[metrics]\nfps = { value = 1, better = \"more\" }").contains("better"));

        assert!(bad("kind = \"custom\"\n[metrics]\nfps = { unit = \"Hz\" }").contains("`value` is required"));

        assert!(bad("kind = \"custom\"\ntitle = ").starts_with("line 2"), "{}", bad("kind = \"custom\"\ntitle = "));

    }

    #[test]

    fn manifests_are_checked() {

        let blob = "a".repeat(64);

        let m = |files: &str| parse_manifest(format!("{{\"version\":1,\"files\":[{files}]}}").as_bytes());

        let file = |p: &str| format!("{{\"path\":\"{p}\",\"blob\":\"{blob}\",\"size\":3}}");

        let ok = m(&[file("report.toml"), file("data/a.csv")].join(",")).unwrap();

        assert_eq!((ok.files.len(), ok.bytes()), (2, 6));

        assert!(m(&file("index.html")).unwrap_err().contains("report.toml"));

        assert!(m(&[file("report.toml"), file("../x")].join(",")).is_err());

        assert!(m(&[file("report.toml"), file("a b")].join(",")).is_err());

        assert!(m(&[file("report.toml"), file("report.toml")].join(",")).unwrap_err().contains("twice"));

        assert!(m("{\"path\":\"report.toml\",\"blob\":\"zz\",\"size\":1}").unwrap_err().contains("blob"));

        assert!(parse_manifest(b"{\"version\":2,\"files\":[]}").unwrap_err().contains("version 2"));

        assert!(path_ok("assets/js/chart.min.js") && !path_ok("/abs") && !path_ok("a//b") && !path_ok(""));

    }

}

        .route(

            "/flow/jobs/{id}/reports/{name}/files",

            post(crate::flow::report_file).layer(axum::extract::DefaultBodyLimit::max(crate::flow::MAX_ARTIFACT_BYTES + (1 << 20))),

        )

/// Job `job_id`'s artifacts, by name.

    // A report's artifact is its manifest (#127): checked, and its files

    // found in the store, before it is kept.

    let report = match out.kind {

        flow_core::OutputKind::Report => Some(report_manifest(&app, &repo, &name, &body).await?),

        flow_core::OutputKind::Artifact => None,

    };

    let artifact = report.is_none();

    let report_bytes = (artifact && name == crate::coverage::ARTIFACT).then(|| body.clone());

    let junit_bytes = (artifact && name == crate::junit::ARTIFACT).then(|| body.clone());

    if let Some((manifest, declaration)) = report {

        let origin = crate::reports::Origin {

            repo_id: row.id,

            run_id: run.id,

            run_number: run.number,

            job_id: job.id,

            job: &job.name,

            bookmark: &run.bookmark,

            state: &run.state,

        };

        let said = match declaration {

            Ok(d) => match crate::reports::keep(&app, &origin, &a.name, &d, &manifest, &a.blob).await {

                Ok(_) => {

                    let s = if d.metrics.len() == 1 { "" } else { "s" };

                    format!("report {}: {} \"{}\", {} metric{s}", a.name, d.kind, d.title, d.metrics.len())

                }

                Err(e) => format!("report {}: not indexed: {e}", a.name),

            },

            Err(why) => {

                let d = crate::reports::Declaration::unreadable(&a.name, &why);

                crate::reports::keep(&app, &origin, &a.name, &d, &manifest, &a.blob).await.ok();

                format!("report {}: report.toml: {why}", a.name)

            }

        };

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

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

    }

}

/// Check report manifest `body` (#127) and that the repository holds every

/// file it names; answers it with the report's declaration, or why that

/// could not be read.

async fn report_manifest(

    app: &Shared,

    repo: &RepoName,

    name: &str,

    body: &[u8],

) -> Result<(crate::reports::Manifest, Result<crate::reports::Declaration, String>), ApiError> {

    let manifest = crate::reports::parse_manifest(body).map_err(|e| ApiError(StatusCode::BAD_REQUEST, e))?;

    let refs: Vec<(String, oxy_core::hash::ObjectRef)> =

        manifest.files.iter().filter_map(|f| Some((f.path.clone(), oxy_core::hash::ObjectRef::from_hex(&f.blob)?))).collect();

    let declaration = manifest.blob("report.toml").and_then(oxy_core::hash::ObjectRef::from_hex);

    let found = app

        .repos

        .read(repo, move |r, _| {

            if let Some((path, _)) = refs.iter().find(|(_, b)| !r.store().has(b)) {

                return Ok(Err(path.clone()));

            }

            Ok(Ok(declaration.map(|b| r.read_blob(&b)).transpose()?))

        })

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("read the report: {e}")))?;

    let toml = found.map_err(|path| ApiError(StatusCode::BAD_REQUEST, format!("report file {path} was not uploaded; upload it first")))?;

    let declaration = crate::reports::parse_declaration(name, &toml.unwrap_or_default());

    Ok((manifest, declaration))

}

/// `POST /api/v1/flow/jobs/{id}/reports/{name}/files`: one file of a report

/// the job declares (#127), its path in `X-Flow-Report-Path`, kept as a

/// blob of the run's repository. Answers `{"blob"}`; the manifest naming

/// the files follows as the report's artifact.

pub async fn report_file(

    State(app): State<Shared>,

    headers: HeaderMap,

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

    body: axum::body::Bytes,

) -> Result<Response, ApiError> {

    runner_auth(&app, &headers).await?;

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

    let spec: flow_core::Job =

        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;

    if !spec.outputs.iter().any(|o| o.name == name && o.kind == flow_core::OutputKind::Report) {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no report {name:?}", job.name)));

    }

    let path = headers.get("x-flow-report-path").and_then(|v| v.to_str().ok()).unwrap_or_default();

    if !crate::reports::path_ok(path) {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("report file {path:?}: give its path in X-Flow-Report-Path")));

    }

    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()))?;

    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;

    let blob = app

        .repos

        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the report file: {e}")))?;

    Ok(Json(serde_json::json!({ "blob": blob.to_hex() })).into_response())

pub mod reports;

        // Flow's artifacts and reports' files are named only in the

        // database: kept as roots, or gc would prune them after the grace

        // period (#126, #127). A listing that fails skips this repository

        // rather than risk them.

        let listed = match db::flow::artifact_blobs(&app.db, row.id).await {

            Ok(mut blobs) => crate::reports::file_blobs(&app.db, row.id).await.map(|files| {

                blobs.extend(files);

                blobs

            }),

            Err(e) => Err(e),

        };

        let keep: Vec<oxy_core::hash::ObjectRef> = match listed {

async fn upload_report_file(app: &Shared, job: i64, name: &str, path: &str, bytes: &[u8]) -> (StatusCode, String) {

    let req = Request::post(format!("/api/v1/flow/jobs/{job}/reports/{name}/files"))

        .header(header::AUTHORIZATION, "Bearer runner-secret")

        .header(header::CONTENT_TYPE, "application/octet-stream")

        .header("X-Flow-Report-Path", path)

        .body(Body::from(bytes.to_vec()))

        .unwrap();

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

    let status = res.status();

    (status, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())

}

/// #127: a report's files arrive one by one, then its manifest as the

/// report's artifact; the hub indexes the declaration, metrics and files,

/// and gc keeps every file.

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

async fn reports_are_kept_file_by_file_and_indexed() {

    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",

        &format!(

            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"perf\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n        report \"frame-timing\" {{\n            path \"/src/out/report\"\n            when \"always\"\n        }}\n    }}\n}}\n"

        ),

    );

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

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

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

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

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

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

    let toml = b"kind = \"performance\"\ntitle = \"Frame timing\"\nsummary = \"Median 16.4 ms\"\n\n[metrics]\nmedian_ms = { value = 16.4, unit = \"ms\", better = \"lower\", fail_above = 20 }\n\n[context]\ntarget_fps = 30\n";

    let page = b"<p>frames</p>".to_vec();

    let data: Vec<u8> = (0..100_000u32).map(|i| (i % 241) as u8).collect();

    let files: [(&str, &[u8]); 3] = [("data/frames.csv", &data), ("index.html", &page), ("report.toml", toml)];

    assert_eq!(upload_report_file(&h.app, id, "tool", "report.toml", toml).await.0, StatusCode::BAD_REQUEST, "not a report");

    assert_eq!(upload_report_file(&h.app, id, "frame-timing", "../x", toml).await.0, StatusCode::BAD_REQUEST, "not a clean path");

    let mut listed = vec![];

    for (path, bytes) in files {

        let (st, text) = upload_report_file(&h.app, id, "frame-timing", path, bytes).await;

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

        let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();

        assert_eq!(blob, oxy_core::repo::Repo::blob_ref(bytes).to_hex());

        listed.push(serde_json::json!({ "path": path, "blob": blob, "size": bytes.len() }));

    }

    let mut missing = listed.clone();

    missing.push(serde_json::json!({ "path": "assets/x.js", "blob": "ab".repeat(32), "size": 1 }));

    let (st, text) =

        upload(&h.app, id, "frame-timing", &serde_json::to_vec(&serde_json::json!({ "version": 1, "files": missing })).unwrap()).await;

    assert_eq!(st, StatusCode::BAD_REQUEST, "a file never uploaded: {text}");

    assert!(text.contains("assets/x.js"), "{text}");

    let manifest = serde_json::to_vec(&serde_json::json!({ "version": 1, "files": listed })).unwrap();

    let (st, text) = upload(&h.app, id, "frame-timing", &manifest).await;

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

    let (kind, title, summary, files_n, bytes): (String, String, String, i64, i64) =

        sqlx::query_as("SELECT kind, title, summary, files, bytes FROM reports WHERE job_id = ? AND name = 'frame-timing'")

            .bind(id)

            .fetch_one(&h.app.db)

            .await

            .unwrap();

    assert_eq!((kind.as_str(), title.as_str(), summary.as_str()), ("performance", "Frame timing", "Median 16.4 ms"));

    assert_eq!((files_n, bytes), (3, (data.len() + page.len() + toml.len()) as i64));

    let rid: i64 = sqlx::query_scalar("SELECT id FROM reports WHERE job_id = ?").bind(id).fetch_one(&h.app.db).await.unwrap();

    let m = oxygenhub::reports::metrics(&h.app.db, rid).await.unwrap();

    assert_eq!((m[0].key.as_str(), m[0].value, m[0].fail_above), ("median_ms", 16.4, Some(20.0)));

    let log = db::flow::log(&h.app.db, id, 0, 1000).await.unwrap();

    assert!(log.iter().any(|r| r.text.contains("report frame-timing: performance \"Frame timing\", 1 metric")), "{log:?}");

    let _ = run;

    // gc keeps every file: only the database names them.

    {

        let name = RepoName::parse("andy/demo").unwrap();

        let mut hosted = oxy_core::repo::Repo::open(&h.app.repos.dir(&name)).unwrap();

        hosted.config.set("core.gc-grace", "0").unwrap();

        hosted.save_config().unwrap();

    }

    let done = oxygenhub::maintenance::gc_all(&h.app).await;

    assert!(done.iter().all(|d| !d.failed), "{done:?}");

    let hosted = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap();

    for (path, bytes) in files {

        assert_eq!(hosted.read_blob(&oxy_core::repo::Repo::blob_ref(bytes)).unwrap(), bytes, "{path} survives gc");

    }

}

The bundle travels file by file: each to

`POST /api/v1/flow/jobs/{job}/reports/{name}/files` with its path in

`X-Flow-Report-Path`, answered with its blob; then the manifest,

`{"version":1,"files":[{"path","blob","size"}]}`, as the job's artifact

under the report's name, which the receipt names. The hub refuses a

manifest naming a file it does not hold, reads `report.toml` and indexes

the report; a `report.toml` it cannot read is kept as a custom report whose

summary says why, so the job's log and the tab show what to fix. Paths are

`/`-separated names of letters, digits, `.`, `_`, `-` and `+`; a report

holds at most 2000 files and 512 MiB.

   - `report_files`: one row per file per report: path, blob, size; what

     garbage collection keeps and the frame serves.