//! Route tests for the CI run and job pages. The fixture: alice owns the private repo `proj`, bob
//! has Write, dave has Read, and carol is outside the repo.

use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use forge_core::ci::{CiJob, CiRun, JobState, NewJob, RunState};
use forge_core::{Role, Store, Visibility};
use forge_git::repo::{init_bare, repo_path};
use topcoat::router::request::Request;
use topcoat::router::{Body, HeaderMap, Router, to_bytes};

use crate::auth::SESSION_COOKIE;
use crate::blobs::BlobStore;
use crate::ci::{CiLive, artifact_key, log_path};
use crate::test_support::{app_router, create_session, test_state};

const CSRF: &str = "ci-test-csrf";
const SHA: &str = "0123456789abcdef0123456789abcdef01234567";
const SPEC: &str = r#"{"runs_on":[]}"#;
const RUN_BASE: &str = "/alice/proj/ci";

struct Fixture {
    store: Store,
    app: Router,
    live: Arc<CiLive>,
    repo: forge_core::Repo,
    alice: String,
    data_dir: PathBuf,
    blobs: BlobStore,
}

async fn fixture(dir: &Path) -> Fixture {
    let state = test_state(dir).await;
    let store = state.store.clone();
    let alice = store
        .create_user("alice", None, "unused", false)
        .await
        .expect("alice");
    let bob = store
        .create_user("bob", None, "unused", false)
        .await
        .expect("bob");
    let dave = store
        .create_user("dave", None, "unused", false)
        .await
        .expect("dave");
    let carol = store
        .create_user("carol", None, "unused", false)
        .await
        .expect("carol");
    let repo = store
        .create_repo(&alice.id, "proj", "", Visibility::Private, "main")
        .await
        .expect("repo");
    let bare = repo_path(&state.git.data_dir, &repo.id).expect("repo path");
    init_bare(&state.git, &bare, "main")
        .await
        .expect("init bare");
    store
        .set_collaborator(&repo.id, &bob.id, Role::Write)
        .await
        .expect("bob writes");
    store
        .set_collaborator(&repo.id, &dave.id, Role::Read)
        .await
        .expect("dave reads");
    create_session(&store, &alice.id, "alice-token", CSRF).await;
    create_session(&store, &bob.id, "bob-token", CSRF).await;
    create_session(&store, &dave.id, "dave-token", CSRF).await;
    create_session(&store, &carol.id, "carol-token", CSRF).await;
    let data_dir = state.config.data_dir.clone();
    let blobs = state.blobs.clone();
    let live = Arc::clone(&state.ci_live);
    Fixture {
        store,
        app: app_router(state),
        live,
        repo,
        alice: alice.id,
        data_dir,
        blobs,
    }
}

fn cookie_for(token: &str) -> String {
    format!("__Host-{SESSION_COOKIE}={token}")
}

/// A response: status, headers and the whole body.
struct Reply {
    status: u16,
    headers: HeaderMap,
    body: Vec<u8>,
}

impl Reply {
    fn text(&self) -> String {
        String::from_utf8_lossy(&self.body).into_owned()
    }

    fn location(&self) -> Option<String> {
        self.headers
            .get("location")
            .and_then(|value| value.to_str().ok())
            .map(str::to_owned)
    }
}

async fn get(app: &Router, uri: &str, cookie: Option<&str>) -> Reply {
    let mut builder = Request::builder().method("GET").uri(uri);
    if let Some(cookie) = cookie {
        builder = builder.header("cookie", cookie);
    }
    let req = builder.body(Body::from(String::new())).expect("request");
    collect(app.handle(req).await).await
}

/// A POST with the CSRF token and the given form fields.
async fn post(app: &Router, uri: &str, cookie: &str, fields: &[(&str, &str)]) -> Reply {
    let mut form = form_urlencoded::Serializer::new(String::new());
    form.append_pair("csrf", CSRF);
    for (name, value) in fields {
        form.append_pair(name, value);
    }
    let req = Request::builder()
        .method("POST")
        .uri(uri)
        .header("cookie", cookie)
        .header("content-type", "application/x-www-form-urlencoded")
        .body(Body::from(form.finish()))
        .expect("request");
    collect(app.handle(req).await).await
}

async fn collect(res: topcoat::router::response::Response) -> Reply {
    let status = res.status().as_u16();
    let headers = res.headers().clone();
    let body = to_bytes(res.into_body(), usize::MAX).await.expect("body");
    Reply {
        status,
        headers,
        body: body.to_vec(),
    }
}

/// Creates a run on `main` with one job per name. Jobs wait for approval when `needs_approval`.
async fn run_with(fx: &Fixture, jobs: &[&str], needs_approval: bool) -> CiRun {
    let needs: &[String] = &[];
    let specs: Vec<NewJob<'_>> = jobs
        .iter()
        .map(|name| NewJob {
            name,
            spec_json: SPEC,
            needs,
        })
        .collect();
    fx.store
        .create_run(
            &fx.repo.id,
            "push",
            "refs/heads/main",
            SHA,
            None,
            &fx.alice,
            needs_approval,
            &specs,
            None,
        )
        .await
        .expect("run")
}

/// Claims the one queued job with a fresh runner and finishes it as a success.
async fn finish_job(fx: &Fixture) -> CiJob {
    let token = fx
        .store
        .create_registration_token(&fx.alice, 24)
        .await
        .expect("token");
    let (runner, _) = fx
        .store
        .register_runner(&token, "runner-1", &["linux".to_owned()])
        .await
        .expect("runner");
    let (job, _, _) = fx
        .store
        .claim_job(&runner)
        .await
        .expect("claim")
        .expect("a queued job");
    fx.store
        .finish_job(&job.id, JobState::Success, Some(0), None)
        .await
        .expect("finish");
    job
}

async fn job_of(fx: &Fixture, run: &CiRun) -> CiJob {
    fx.store
        .jobs_for_run(&run.id)
        .await
        .expect("jobs")
        .into_iter()
        .next()
        .expect("a job")
}

#[tokio::test]
async fn private_repo_is_not_found_to_outsiders_and_anonymous() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let urls = [
        RUN_BASE.to_owned(),
        format!("{RUN_BASE}/runs/{}", run.number),
        format!("/alice/proj/ci/jobs/{}", job.id),
        format!("/alice/proj/ci/jobs/{}/log.txt", job.id),
        format!("/alice/proj/ci/jobs/{}/artifacts/x.bin", job.id),
    ];
    for url in &urls {
        let anon = get(&fx.app, url, None).await;
        assert_eq!(anon.status, 404, "anonymous {url}");
        let outsider = get(&fx.app, url, Some(&cookie_for("carol-token"))).await;
        assert_eq!(outsider.status, 404, "outsider {url}");
    }
    let reader = get(&fx.app, RUN_BASE, Some(&cookie_for("dave-token"))).await;
    assert_eq!(reader.status, 200);
}

#[tokio::test]
async fn buttons_and_manual_form_need_write() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let run_url = format!("{RUN_BASE}/runs/{}", run.number);
    let writer = get(&fx.app, &run_url, Some(&cookie_for("bob-token"))).await;
    assert!(writer.text().contains(">Re-run<"), "writer sees re-run");
    assert!(writer.text().contains(">Cancel<"), "writer sees cancel");
    let reader = get(&fx.app, &run_url, Some(&cookie_for("dave-token"))).await;
    assert_eq!(reader.status, 200);
    assert!(!reader.text().contains(">Re-run<"), "reader sees no re-run");
    assert!(!reader.text().contains(">Cancel<"), "reader sees no cancel");
    let list = get(&fx.app, RUN_BASE, Some(&cookie_for("dave-token"))).await;
    assert!(
        !list.text().contains("Run workflow"),
        "reader has no manual form"
    );
    let list = get(&fx.app, RUN_BASE, Some(&cookie_for("bob-token"))).await;
    assert!(
        list.text().contains("Run workflow"),
        "writer has the manual form"
    );
    let refused = post(
        &fx.app,
        &format!("{run_url}/cancel"),
        &cookie_for("dave-token"),
        &[],
    )
    .await;
    assert_eq!(refused.status, 403, "reader cannot cancel");
    let still = fx
        .store
        .run_by_id(&run.id)
        .await
        .expect("run")
        .expect("run");
    assert_eq!(still.state, RunState::Queued);
}

#[tokio::test]
async fn run_list_pages_25_at_a_time() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    for _ in 0..30 {
        run_with(&fx, &[], false).await;
    }
    let first = get(&fx.app, RUN_BASE, Some(&cookie_for("alice-token"))).await;
    let text = first.text();
    assert_eq!(text.matches("href=\"/alice/proj/ci/runs/").count(), 25);
    assert!(text.contains("?page=2"), "older link");
    assert!(!text.contains(">Newer<"));
    let second = get(
        &fx.app,
        &format!("{RUN_BASE}?page=2"),
        Some(&cookie_for("alice-token")),
    )
    .await;
    let text = second.text();
    assert_eq!(text.matches("href=\"/alice/proj/ci/runs/").count(), 5);
    assert!(text.contains(">Newer<"), "newer link");
    assert!(!text.contains("?page=3"), "no older link on the last page");
    let bogus = get(
        &fx.app,
        &format!("{RUN_BASE}?page=abc"),
        Some(&cookie_for("alice-token")),
    )
    .await;
    assert_eq!(bogus.status, 200);
    assert_eq!(
        bogus.text().matches("href=\"/alice/proj/ci/runs/").count(),
        25
    );
}

/// The pages no longer reload themselves. A page that is not finished carries its live marker, and
/// a finished one carries none.
#[tokio::test]
async fn pages_are_live_while_not_finished_and_never_meta_refresh() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let url = format!("/alice/proj/ci/jobs/{}", job.id);
    let run_url = format!("{RUN_BASE}/runs/{}", run.number);
    let job_marker = format!("data-live=\"{url}/events\"");
    let run_marker = format!("data-live=\"{run_url}/events\"");

    let queued = get(&fx.app, &url, Some(&cookie_for("alice-token"))).await;
    assert!(queued.text().contains(&job_marker), "job live while queued");
    let run_page = get(&fx.app, &run_url, Some(&cookie_for("alice-token"))).await;
    assert!(
        run_page.text().contains(&run_marker),
        "run live while queued"
    );
    for page in [&queued, &run_page] {
        assert!(
            !page.text().contains("http-equiv=\"refresh\""),
            "no meta refresh"
        );
    }

    finish_job(&fx).await;
    let done = get(&fx.app, &url, Some(&cookie_for("alice-token"))).await;
    assert!(
        !done.text().contains("data-live="),
        "job not live when finished"
    );
    assert!(
        !done.text().contains("http-equiv=\"refresh\""),
        "no meta refresh"
    );
}

/// Opens an event stream without reading it. The response head carries the status.
async fn open(
    app: &Router,
    uri: &str,
    cookie: Option<&str>,
) -> topcoat::router::response::Response {
    let mut builder = Request::builder().method("GET").uri(uri);
    if let Some(cookie) = cookie {
        builder = builder.header("cookie", cookie);
    }
    app.handle(builder.body(Body::from(String::new())).expect("request"))
        .await
}

/// The next frame of an event stream, or None once the stream has ended. Fails after five seconds.
async fn next_event<S, E>(frames: &mut S) -> Option<Vec<u8>>
where
    S: futures_util::Stream<Item = Result<bytes::Bytes, E>> + Unpin,
    E: std::fmt::Debug,
{
    use futures_util::StreamExt;

    let frame = tokio::time::timeout(Duration::from_secs(5), frames.next())
        .await
        .expect("an event within five seconds")?;
    Some(frame.expect("data").to_vec())
}

#[tokio::test]
async fn live_events_need_the_same_access_as_the_page() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let urls = [
        format!("{RUN_BASE}/runs/{}/events", run.number),
        format!("/alice/proj/ci/jobs/{}/events", job.id),
    ];
    for url in &urls {
        let anon = open(&fx.app, url, None).await;
        assert_eq!(anon.status().as_u16(), 404, "anonymous {url}");
        let outsider = open(&fx.app, url, Some(&cookie_for("carol-token"))).await;
        assert_eq!(outsider.status().as_u16(), 404, "outsider {url}");
        let reader = open(&fx.app, url, Some(&cookie_for("dave-token"))).await;
        assert_eq!(reader.status().as_u16(), 200, "reader {url}");
    }
}

/// Each stream opens with its status. The run stream follows every bump; the job stream sends only
/// status changes, so a log append is silent, and it ends once the job is terminal.
#[tokio::test]
async fn live_events_open_with_the_status_and_follow_changes() {
    use futures_util::StreamExt;

    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let alice = cookie_for("alice-token");

    let mut run_frames = open(
        &fx.app,
        &format!("{RUN_BASE}/runs/{}/events", run.number),
        Some(&alice),
    )
    .await
    .into_body()
    .into_data_stream();
    let first = next_event(&mut run_frames).await.expect("the run status");
    assert_eq!(
        first,
        format!("event: changed\ndata: {}\n\n", run.state.as_str()).into_bytes()
    );

    let mut job_frames = open(
        &fx.app,
        &format!("/alice/proj/ci/jobs/{}/events", job.id),
        Some(&alice),
    )
    .await
    .into_body()
    .into_data_stream();
    let first = next_event(&mut job_frames).await.expect("the job status");
    assert_eq!(
        first,
        format!("event: changed\ndata: {}\n\n", job.state.as_str()).into_bytes()
    );

    // A bump with no new log bytes and no status change sends nothing.
    fx.live.bump_job(&job.id);
    let quiet = tokio::time::timeout(Duration::from_millis(200), job_frames.next()).await;
    assert!(quiet.is_err(), "a log append sends no status event");

    // The run stream sends its status on every bump of the run.
    fx.live.bump_run(&run.id);
    let bumped = next_event(&mut run_frames).await.expect("the run status");
    let now = fx
        .store
        .run_by_id(&run.id)
        .await
        .expect("run")
        .expect("run");
    assert_eq!(
        bumped,
        format!("event: changed\ndata: {}\n\n", now.state.as_str()).into_bytes()
    );

    // The job finishes: its final status is sent, then the stream ends.
    finish_job(&fx).await;
    fx.live.close_job(&job.id);
    let last = next_event(&mut job_frames).await.expect("the final status");
    assert_eq!(
        last,
        format!("event: changed\ndata: {}\n\n", JobState::Success.as_str()).into_bytes()
    );
    assert!(next_event(&mut job_frames).await.is_none(), "stream ends");
}

/// The job's log events: each append is sent as a `log` event with its bytes, the id is the offset
/// past them, and the bytes before `?from=` are not sent again.
#[tokio::test]
async fn job_events_send_each_log_append_as_a_log_event() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let path = log_path(&fx.data_dir, &job.id);
    std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
    std::fs::write(&path, b"one\n").expect("log");

    let mut frames = open(
        &fx.app,
        &format!("/alice/proj/ci/jobs/{}/events?from=4", job.id),
        Some(&cookie_for("alice-token")),
    )
    .await
    .into_body()
    .into_data_stream();
    let first = next_event(&mut frames).await.expect("the status");
    assert_eq!(
        first,
        format!("event: changed\ndata: {}\n\n", job.state.as_str()).into_bytes()
    );

    let mut file = std::fs::OpenOptions::new()
        .append(true)
        .open(&path)
        .expect("log");
    file.write_all(b"two\nthree\n").expect("append");
    fx.live.bump_job(&job.id);
    let log = next_event(&mut frames).await.expect("the log event");
    assert_eq!(
        log,
        b"event: log\ndata: two\ndata: three\ndata: \nid: 14\n\n".to_vec()
    );
}

/// An append longer than 64 KiB is sent in several `log` events, in order.
#[tokio::test]
async fn job_events_split_a_long_append_into_64_kib_events() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let path = log_path(&fx.data_dir, &job.id);
    std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
    std::fs::write(&path, b"").expect("log");

    let mut frames = open(
        &fx.app,
        &format!("/alice/proj/ci/jobs/{}/events", job.id),
        Some(&cookie_for("alice-token")),
    )
    .await
    .into_body()
    .into_data_stream();
    next_event(&mut frames).await.expect("the status");

    let long = "a".repeat(150 * 1024);
    let mut file = std::fs::OpenOptions::new()
        .append(true)
        .open(&path)
        .expect("log");
    file.write_all(long.as_bytes()).expect("append");
    fx.live.bump_job(&job.id);

    let mut sizes = Vec::new();
    for _ in 0..3 {
        let frame = next_event(&mut frames).await.expect("a log event");
        let text = String::from_utf8(frame).expect("text");
        assert!(text.starts_with("event: log\n"), "{text}");
        let data = text
            .lines()
            .find_map(|line| line.strip_prefix("data: "))
            .expect("data line");
        sizes.push(data.len());
        if sizes.len() == 3 {
            assert!(
                text.contains("id: 153600\n"),
                "the last id is the end offset"
            );
        }
    }
    assert_eq!(sizes, vec![64 * 1024, 64 * 1024, 150 * 1024 - 128 * 1024]);
}

/// A reconnect sends `Last-Event-ID`, and the stream resumes after that offset, not at `?from=`.
#[tokio::test]
async fn job_events_resume_after_the_last_event_id() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let path = log_path(&fx.data_dir, &job.id);
    std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
    std::fs::write(&path, b"one\ntwo\n").expect("log");

    let request = Request::builder()
        .method("GET")
        .uri(format!("/alice/proj/ci/jobs/{}/events?from=0", job.id))
        .header("cookie", cookie_for("alice-token"))
        .header("last-event-id", "4")
        .body(Body::from(String::new()))
        .expect("request");
    let mut frames = fx.app.handle(request).await.into_body().into_data_stream();
    next_event(&mut frames).await.expect("the status");
    let log = next_event(&mut frames).await.expect("the log event");
    assert_eq!(log, b"event: log\ndata: two\ndata: \nid: 8\n\n".to_vec());
}

/// The job page's log tail says where its text ends, so the events stream starts from there.
#[tokio::test]
async fn job_page_tail_carries_its_offset() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let path = log_path(&fx.data_dir, &job.id);
    std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
    std::fs::write(&path, b"one\ntwo\n").expect("log");

    let page = get(
        &fx.app,
        &format!("/alice/proj/ci/jobs/{}", job.id),
        Some(&cookie_for("alice-token")),
    )
    .await;
    let text = page.text();
    assert!(
        text.contains("data-live-log=\"true\" data-offset=\"8\""),
        "tail offset"
    );
    assert!(
        text.contains(&format!("data-live-state=\"{}\"", job.state.as_str())),
        "rendered status"
    );
}

#[tokio::test]
async fn job_page_tails_the_log_at_512_kib() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let path = log_path(&fx.data_dir, &job.id);
    std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
    let mut body = b"HEAD-MARK\n".to_vec();
    let line = format!("{}\n", "x".repeat(99));
    for _ in 0..6_000 {
        body.extend_from_slice(line.as_bytes());
    }
    body.extend_from_slice(b"TAIL-MARK\n");
    std::fs::write(&path, &body).expect("log");
    let page = get(
        &fx.app,
        &format!("/alice/proj/ci/jobs/{}", job.id),
        Some(&cookie_for("alice-token")),
    )
    .await;
    let text = page.text();
    assert!(text.contains("TAIL-MARK"));
    assert!(!text.contains("HEAD-MARK"));
    assert!(text.contains("Showing the last 512 KiB"));
    assert!(
        page.body.len() < 530 * 1024,
        "page is {} bytes",
        page.body.len()
    );
}

#[tokio::test]
async fn log_txt_is_the_full_file_for_a_finished_job() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let path = log_path(&fx.data_dir, &job.id);
    std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
    std::fs::write(&path, b"line one\nline two\n").expect("log");
    finish_job(&fx).await;
    let res = get(
        &fx.app,
        &format!("/alice/proj/ci/jobs/{}/log.txt", job.id),
        Some(&cookie_for("alice-token")),
    )
    .await;
    assert_eq!(res.status, 200);
    assert_eq!(res.body, b"line one\nline two\n");
    assert_eq!(
        res.headers
            .get("content-type")
            .and_then(|v| v.to_str().ok()),
        Some("text/plain; charset=utf-8")
    );
    assert_eq!(
        res.headers
            .get("x-content-type-options")
            .and_then(|v| v.to_str().ok()),
        Some("nosniff")
    );
}

/// The next chunk a follower sends. Fails the test when none comes within five seconds.
async fn next_chunk(
    chunks: &mut tokio::sync::mpsc::UnboundedReceiver<bytes::Bytes>,
) -> Option<bytes::Bytes> {
    tokio::time::timeout(Duration::from_secs(5), chunks.recv())
        .await
        .expect("a chunk within five seconds")
}

/// A follower sends each append as soon as the job's signal is bumped. The wait is far longer
/// than the timeouts here, so a chunk that arrives in time was woken by the bump, not by a timer.
#[tokio::test]
async fn log_follower_is_woken_by_bumps_and_ends_on_close() {
    use futures_util::StreamExt;

    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let path = log_path(&fx.data_dir, &job.id);
    std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
    std::fs::write(&path, b"one\n").expect("log");
    let live = Arc::new(CiLive::default());
    let mut stream = Box::pin(super::ci_job::follow_log(
        fx.store.clone(),
        Arc::clone(&live),
        job.id.clone(),
        path.clone(),
    ));
    let (tx, mut chunks) = tokio::sync::mpsc::unbounded_channel();
    tokio::spawn(async move {
        while let Some(Ok(chunk)) = stream.next().await {
            if tx.send(chunk).is_err() {
                break;
            }
        }
    });
    assert_eq!(
        next_chunk(&mut chunks).await.as_deref(),
        Some(&b"one\n"[..])
    );

    let mut file = std::fs::OpenOptions::new()
        .append(true)
        .open(&path)
        .expect("log");
    file.write_all(b"two\n").expect("append");
    live.bump_job(&job.id);
    assert_eq!(
        next_chunk(&mut chunks).await.as_deref(),
        Some(&b"two\n"[..])
    );

    finish_job(&fx).await;
    live.close_job(&job.id);
    let end = tokio::time::timeout(Duration::from_secs(5), chunks.recv())
        .await
        .expect("the stream ends once the job is terminal");
    assert!(end.is_none());
}

#[tokio::test]
async fn artifact_download_sends_attachment_headers() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let job = job_of(&fx, &run).await;
    let payload = b"artifact bytes";
    let artifact = fx
        .store
        .add_artifact(&job.id, "report.txt", payload.len() as i64, SHA)
        .await
        .expect("artifact row");
    let mut writer = fx
        .blobs
        .begin(&artifact_key(&artifact.id), 1024)
        .await
        .expect("begin");
    writer.push(payload).await.expect("push");
    writer.finish().await.expect("finish");
    let res = get(
        &fx.app,
        &format!("/alice/proj/ci/jobs/{}/artifacts/report.txt", job.id),
        Some(&cookie_for("alice-token")),
    )
    .await;
    assert_eq!(res.status, 200);
    assert_eq!(res.body, payload);
    let header = |name: &str| {
        res.headers
            .get(name)
            .and_then(|v| v.to_str().ok())
            .map(str::to_owned)
    };
    assert_eq!(
        header("content-disposition").as_deref(),
        Some("attachment; filename=\"report.txt\"")
    );
    assert_eq!(header("content-length").as_deref(), Some("14"));
    assert_eq!(header("x-content-type-options").as_deref(), Some("nosniff"));
    assert_eq!(
        header("content-type").as_deref(),
        Some("application/octet-stream")
    );
}

#[tokio::test]
async fn approve_only_works_while_waiting_for_approval() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], true).await;
    let run_url = format!("{RUN_BASE}/runs/{}", run.number);
    let waiting = get(&fx.app, &run_url, Some(&cookie_for("bob-token"))).await;
    assert!(
        waiting.text().contains(">Approve and run<"),
        "approve button while waiting"
    );
    let reader = get(&fx.app, &run_url, Some(&cookie_for("dave-token"))).await;
    assert!(
        !reader.text().contains(">Approve and run<"),
        "no approve button for readers"
    );

    let approved = post(
        &fx.app,
        &format!("{run_url}/approve"),
        &cookie_for("bob-token"),
        &[],
    )
    .await;
    assert!(
        approved
            .location()
            .unwrap_or_default()
            .ends_with("?notice=approved")
    );
    let now = fx
        .store
        .run_by_id(&run.id)
        .await
        .expect("run")
        .expect("run");
    assert_eq!(now.state, RunState::Queued);
    let page = get(&fx.app, &run_url, Some(&cookie_for("bob-token"))).await;
    assert!(
        !page.text().contains(">Approve and run<"),
        "no approve button after approval"
    );

    let again = post(
        &fx.app,
        &format!("{run_url}/approve"),
        &cookie_for("bob-token"),
        &[],
    )
    .await;
    assert!(
        again
            .location()
            .unwrap_or_default()
            .ends_with("?notice=not-waiting")
    );
    let still = fx
        .store
        .run_by_id(&run.id)
        .await
        .expect("run")
        .expect("run");
    assert_eq!(still.state, RunState::Queued);
}

#[tokio::test]
async fn cancel_and_rerun_post_and_redirect() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let run = run_with(&fx, &["build"], false).await;
    let run_url = format!("{RUN_BASE}/runs/{}", run.number);
    let cancelled = post(
        &fx.app,
        &format!("{run_url}/cancel"),
        &cookie_for("bob-token"),
        &[],
    )
    .await;
    assert!(
        cancelled
            .location()
            .unwrap_or_default()
            .ends_with("?notice=cancelled")
    );
    let now = fx
        .store
        .run_by_id(&run.id)
        .await
        .expect("run")
        .expect("run");
    assert_eq!(now.state, RunState::Cancelled);
    let finished = post(
        &fx.app,
        &format!("{run_url}/cancel"),
        &cookie_for("bob-token"),
        &[],
    )
    .await;
    assert!(
        finished
            .location()
            .unwrap_or_default()
            .ends_with("?notice=finished")
    );

    let rerun = post(
        &fx.app,
        &format!("{run_url}/rerun"),
        &cookie_for("bob-token"),
        &[],
    )
    .await;
    assert_eq!(
        rerun.location().as_deref(),
        Some("/alice/proj/ci/runs/2?notice=rerun")
    );
    assert!(
        fx.store
            .run_by_number(&fx.repo.id, 2)
            .await
            .expect("lookup")
            .is_some()
    );
}

#[tokio::test]
async fn manual_run_refuses_an_unknown_branch() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let res = post(
        &fx.app,
        RUN_BASE,
        &cookie_for("bob-token"),
        &[("branch", "no-such-branch")],
    )
    .await;
    assert!(
        res.location()
            .unwrap_or_default()
            .ends_with("?notice=no-branch")
    );
    let reader = post(
        &fx.app,
        RUN_BASE,
        &cookie_for("dave-token"),
        &[("branch", "main")],
    )
    .await;
    assert_eq!(reader.status, 403, "reader cannot start a manual run");
}

#[tokio::test]
async fn badge_follows_read_access_and_reports_the_latest_run() {
    let dir = tempfile::tempdir().expect("tempdir");
    let fx = fixture(dir.path()).await;
    let open = fx
        .store
        .create_repo(&fx.alice, "open", "", Visibility::Public, "main")
        .await
        .expect("public repo");
    let svg_path = "/alice/proj/badge.svg";

    // The private repo: a run exists, but only readers see its badge.
    run_with(&fx, &["build"], false).await;
    finish_job(&fx).await;
    assert_eq!(get(&fx.app, svg_path, None).await.status, 404, "anonymous");
    let outsider = get(&fx.app, svg_path, Some(&cookie_for("carol-token"))).await;
    assert_eq!(outsider.status, 404, "outsider");
    let reader = get(&fx.app, svg_path, Some(&cookie_for("dave-token"))).await;
    assert_eq!(reader.status, 200, "reader");
    assert!(reader.text().contains("passing"));
    assert_eq!(
        reader
            .headers
            .get("content-type")
            .and_then(|v| v.to_str().ok()),
        Some("image/svg+xml")
    );
    assert_eq!(
        reader
            .headers
            .get("cache-control")
            .and_then(|v| v.to_str().ok()),
        Some("no-cache, max-age=0")
    );

    // The public repo: anyone sees the badge. Unknown branches and jobs have no status.
    let needs: &[String] = &[];
    let specs = [NewJob {
        name: "build",
        spec_json: SPEC,
        needs,
    }];
    fx.store
        .create_run(
            &open.id,
            "push",
            "refs/heads/main",
            SHA,
            None,
            &fx.alice,
            false,
            &specs,
            None,
        )
        .await
        .expect("run");
    finish_job(&fx).await;
    let anon = get(&fx.app, "/alice/open/badge.svg", None).await;
    assert_eq!(anon.status, 200);
    assert!(anon.text().contains(">passing</text>"));
    let unknown = get(&fx.app, "/alice/open/badge.svg?branch=nope", None).await;
    assert_eq!(unknown.status, 200);
    assert!(unknown.text().contains(">no status</text>"));
    let job = get(&fx.app, "/alice/open/badge.svg?job=build", None).await;
    assert!(job.text().contains(">build</text>"));
    assert!(job.text().contains(">passing</text>"));
    let missing_job = get(&fx.app, "/alice/open/badge.svg?job=lint", None).await;
    assert!(missing_job.text().contains(">no status</text>"));
    let unknown_repo = get(&fx.app, "/alice/nothere/badge.svg", None).await;
    assert_eq!(unknown_repo.status, 404);
}