//! Turns pushes, tags, pull requests and manual requests into CI runs.
//!
//! Every entry point reads `.forge/ci.toml` at the commit it runs (64 KiB cap). No file means no
//! run. A file that does not parse gives an Error run with the message in `config_error`, plus a
//! `ci/config` status. A valid file gives one run whose jobs are created in dependency order.

use forge_ci::config::{CiConfig, Event, parse_config};
use forge_core::authz::{Action, authorize};
use forge_core::ci::{CiRun, NewJob};
use forge_core::store::commit_status::StatusState;
use forge_core::{PullRequest, Repo};
use forge_git::browse::{read_blob, resolve_commit};

use super::{publish, sync_run_statuses};
use crate::internal::is_zero_oid;
use crate::state::AppState;

/// The workflow file, relative to the repo root.
pub const CONFIG_PATH: &str = ".forge/ci.toml";
/// Largest workflow file read, in bytes.
const MAX_CONFIG_BYTES: usize = 64 * 1024;

/// A job ready to store: name, `JobSpec` as JSON, and its needs.
type PlannedJob = (String, String, Vec<String>);

/// What `.forge/ci.toml` held at one commit.
enum Loaded {
    Missing,
    Invalid(String),
    Valid(CiConfig),
}

/// What a request should do with the loaded config.
enum Plan {
    /// No workflow file at this commit.
    Missing,
    /// The file is valid but this event does not trigger it.
    NotTriggered,
    /// Create a run with these jobs, or an Error run when `config_error` is set.
    Run {
        jobs: Vec<PlannedJob>,
        config_error: Option<String>,
    },
}

/// The inputs of one run to create.
struct Request<'a> {
    repo: &'a Repo,
    trigger: &'static str,
    git_ref: &'a str,
    commit: &'a str,
    pull_number: Option<i64>,
    actor_id: &'a str,
    needs_approval: bool,
}

/// A branch or tag update. Called from `post_update::after_ref_update` for every update of
/// `refs/heads/*` or `refs/tags/*`, including pull mirror updates (with an empty actor). Deletions
/// are ignored. Failures are logged; they never fail the push.
pub async fn on_ref_update(
    state: &AppState,
    repo: &Repo,
    actor_id: &str,
    refname: &str,
    new_oid: &str,
) {
    if !state.config.ci.enabled || is_zero_oid(new_oid) {
        return;
    }
    let (trigger, event) = if let Some(branch) = refname.strip_prefix("refs/heads/") {
        ("push", Event::Push { branch })
    } else if let Some(tag) = refname.strip_prefix("refs/tags/") {
        ("tag", Event::Tag { tag })
    } else {
        return;
    };
    let result = async {
        let loaded = load(state, &repo.id, new_oid).await;
        if let Plan::Run { jobs, config_error } = plan_for(loaded, &event)? {
            create(
                state,
                Request {
                    repo,
                    trigger,
                    git_ref: refname,
                    commit: new_oid,
                    pull_number: None,
                    actor_id,
                    needs_approval: false,
                },
                jobs,
                config_error.as_deref(),
            )
            .await?;
        }
        forge_core::Result::Ok(())
    }
    .await;
    if let Err(e) = result {
        tracing::error!(error = %e, repo_id = %repo.id, refname, "ci: scheduling a push failed");
    }
}

/// A pull request was opened, synchronized (its head moved) or reopened. The config is read at
/// the head commit. A fork head, or an author without Write on the base repo, creates the run in
/// WaitingApproval: its jobs wait until someone approves it.
pub async fn on_pull_event(state: &AppState, repo: &Repo, pull: &PullRequest, actor_id: &str) {
    if !state.config.ci.enabled {
        return;
    }
    if let Err(e) = pull_run(state, repo, pull, actor_id).await {
        tracing::error!(error = %e, repo_id = %repo.id, number = pull.issue.number, "ci: scheduling a pull request failed");
    }
}

async fn pull_run(
    state: &AppState,
    repo: &Repo,
    pull: &PullRequest,
    actor_id: &str,
) -> forge_core::Result<()> {
    let loaded = load(state, &repo.id, &pull.head_oid).await;
    let Plan::Run { jobs, config_error } = plan_for(loaded, &Event::PullRequest)? else {
        return Ok(());
    };
    let needs_approval = needs_approval(state, repo, pull).await?;
    let git_ref = format!("refs/pull/{}/head", pull.issue.number);
    create(
        state,
        Request {
            repo,
            trigger: "pull_request",
            git_ref: &git_ref,
            commit: &pull.head_oid,
            pull_number: Some(pull.issue.number),
            actor_id,
            needs_approval,
        },
        jobs,
        config_error.as_deref(),
    )
    .await?;
    Ok(())
}

/// Starts a run for `branch` by hand. Fails when the branch does not exist, when there is no
/// workflow file, or when the workflow does not allow manual runs. A file that does not parse
/// gives an Error run, as a push does.
pub async fn manual_run(
    state: &AppState,
    repo: &Repo,
    branch: &str,
    actor_id: &str,
) -> forge_core::Result<CiRun> {
    if !state.config.ci.enabled {
        return Err(forge_core::Error::Invalid(
            "CI is disabled on this instance".into(),
        ));
    }
    let git_ref = format!("refs/heads/{branch}");
    let repo_dir = forge_git::repo::repo_path(&state.git.data_dir, &repo.id)
        .map_err(|e| forge_core::Error::Git(e.to_string()))?;
    // `resolve_commit` takes the short branch name and looks under refs/heads/ first.
    let short = branch.to_owned();
    let Some(commit) = blocking(move || resolve_commit(&repo_dir, &short)).await? else {
        return Err(forge_core::Error::NotFound);
    };
    let loaded = load(state, &repo.id, &commit).await;
    match plan_for(loaded, &Event::Manual)? {
        Plan::Run { jobs, config_error } => {
            create(
                state,
                Request {
                    repo,
                    trigger: "manual",
                    git_ref: &git_ref,
                    commit: &commit,
                    pull_number: None,
                    actor_id,
                    needs_approval: false,
                },
                jobs,
                config_error.as_deref(),
            )
            .await
        }
        Plan::NotTriggered => Err(forge_core::Error::Invalid(
            "manual runs are not enabled in .forge/ci.toml".into(),
        )),
        Plan::Missing => Err(forge_core::Error::Invalid(
            "this branch has no .forge/ci.toml".into(),
        )),
    }
}

/// Creates the run, sets the statuses of its jobs (and `ci/config` for an error run), and wakes
/// the runner claims when a job is queued.
async fn create(
    state: &AppState,
    req: Request<'_>,
    jobs: Vec<PlannedJob>,
    config_error: Option<&str>,
) -> forge_core::Result<CiRun> {
    let new_jobs: Vec<NewJob<'_>> = jobs
        .iter()
        .map(|(name, spec_json, needs)| NewJob {
            name,
            spec_json,
            needs: needs.as_slice(),
        })
        .collect();
    let run = state
        .store
        .create_run(
            &req.repo.id,
            req.trigger,
            req.git_ref,
            req.commit,
            req.pull_number,
            req.actor_id,
            req.needs_approval,
            &new_jobs,
            config_error,
        )
        .await?;
    if let Some(message) = config_error {
        let url = format!(
            "{}/{}/{}/ci/runs/{}",
            state.config.base_url.trim_end_matches('/'),
            req.repo.owner_name,
            req.repo.name,
            run.number
        );
        publish(
            state,
            req.repo,
            &run,
            "ci/config",
            StatusState::Error,
            message,
            &url,
        )
        .await;
    }
    if sync_run_statuses(state, &run).await? {
        state.ci_waker.notify_one();
    }
    Ok(run)
}

/// Decides what a request does, given the loaded config and the event.
fn plan_for(loaded: Loaded, event: &Event<'_>) -> forge_core::Result<Plan> {
    match loaded {
        Loaded::Missing => Ok(Plan::Missing),
        Loaded::Invalid(message) => Ok(Plan::Run {
            jobs: Vec::new(),
            config_error: Some(message),
        }),
        Loaded::Valid(cfg) => {
            if !cfg.triggered_by(event) {
                return Ok(Plan::NotTriggered);
            }
            Ok(Plan::Run {
                jobs: plan_jobs(&cfg)?,
                config_error: None,
            })
        }
    }
}

/// The jobs of a valid config in dependency order, each with its spec as JSON.
fn plan_jobs(cfg: &CiConfig) -> forge_core::Result<Vec<PlannedJob>> {
    let mut jobs = Vec::new();
    for name in cfg.job_order() {
        let Some(spec) = cfg.jobs.get(name) else {
            continue;
        };
        let spec_json =
            serde_json::to_string(spec).map_err(|e| forge_core::Error::Invalid(e.to_string()))?;
        jobs.push((name.to_owned(), spec_json, spec.needs.clone()));
    }
    Ok(jobs)
}

/// Reads and parses `.forge/ci.toml` at `commit` from the repo `repo_id`.
async fn load(state: &AppState, repo_id: &str, commit: &str) -> Loaded {
    let repo_dir = match forge_git::repo::repo_path(&state.git.data_dir, repo_id) {
        Ok(path) => path,
        Err(e) => {
            tracing::error!(error = %e, repo_id, "ci: bad repo path");
            return Loaded::Invalid(format!("could not read {CONFIG_PATH}"));
        }
    };
    let (dir, rev) = (repo_dir, commit.to_owned());
    let blob = match blocking(move || read_blob(&dir, &rev, CONFIG_PATH, MAX_CONFIG_BYTES)).await {
        Ok(blob) => blob,
        Err(e) => {
            tracing::error!(error = %e, repo_id, commit, "ci: reading the workflow failed");
            return Loaded::Invalid(format!("could not read {CONFIG_PATH}"));
        }
    };
    let Some(blob) = blob else {
        return Loaded::Missing;
    };
    if blob.truncated {
        return Loaded::Invalid(format!("{CONFIG_PATH} is larger than 64 KiB"));
    }
    if blob.is_binary {
        return Loaded::Invalid(format!("{CONFIG_PATH} is not text"));
    }
    let Ok(text) = String::from_utf8(blob.data) else {
        return Loaded::Invalid(format!("{CONFIG_PATH} is not valid UTF-8"));
    };
    match parse_config(&text) {
        Ok(cfg) => Loaded::Valid(cfg),
        Err(e) => Loaded::Invalid(e.to_string()),
    }
}

/// Whether a pull request needs approval before its jobs run: a fork head, or an author who lacks
/// Write on the base repo (or no longer exists).
async fn needs_approval(
    state: &AppState,
    repo: &Repo,
    pull: &PullRequest,
) -> forge_core::Result<bool> {
    if pull.head_repo_id.as_deref() != Some(repo.id.as_str()) {
        return Ok(true);
    }
    let Some(author) = state.store.user_by_id(&pull.issue.author_id).await? else {
        return Ok(true);
    };
    if author.is_org() {
        return Ok(true);
    }
    let role = state.store.collaborator_role(&repo.id, &author.id).await?;
    let org_role = state
        .store
        .org_repo_role(&repo.owner_id, &author.id, &repo.id)
        .await?;
    Ok(authorize(Some(&author), repo, role, org_role, Action::Write).is_err())
}

/// Runs blocking git work off the async runtime, mapping git and join errors to `forge_core`.
async fn blocking<T, F>(work: F) -> forge_core::Result<T>
where
    F: FnOnce() -> forge_git::Result<T> + Send + 'static,
    T: Send + 'static,
{
    match tokio::task::spawn_blocking(work).await {
        Ok(Ok(value)) => Ok(value),
        Ok(Err(e)) => Err(forge_core::Error::Git(e.to_string())),
        Err(e) => Err(forge_core::Error::Git(format!("git task failed: {e}"))),
    }
}

#[cfg(test)]
mod tests {
    use std::path::{Path, PathBuf};
    use std::sync::Arc;
    use std::time::Duration;

    use forge_core::ci::{CiJob, CiRunner, JobState, RunState};
    use forge_core::store::commit_status::{CommitStatus, StatusState};
    use forge_core::time::stamp_after;
    use forge_core::{User, Visibility};
    use forge_git::repo::{init_bare, repo_path};
    use tempfile::TempDir;

    use super::*;
    use crate::ci::log_path;
    use crate::ci::reaper::{purge_once, reap_before};
    use crate::post_update::after_ref_update;
    use crate::test_support::test_state;

    const ZERO: &str = "0000000000000000000000000000000000000000";

    const CONFIG: &str = r#"
[on]
push = ["main"]
pull_request = true

[jobs.build]
image = "alpine:3"
run = "make"

[jobs.test]
image = "alpine:3"
run = "make test"
needs = ["build"]
"#;

    const CONFIG_GLOBS: &str = r#"
[on]
push = ["release/*"]

[jobs.build]
image = "alpine:3"
run = "make"
"#;

    const CONFIG_TAGS: &str = r#"
[on]
push = []
tags = ["v*"]

[jobs.build]
image = "alpine:3"
run = "make"
"#;

    const CONFIG_ONE_JOB: &str = r#"
[on]
push = ["main"]

[jobs.build]
image = "alpine:3"
run = "make"
"#;

    const CONFIG_NO_MANUAL: &str = r#"
[on]
push = ["main"]
manual = false

[jobs.build]
image = "alpine:3"
run = "make"
"#;

    fn git(dir: &Path, args: &[&str]) -> String {
        let out = std::process::Command::new("git")
            .arg("-C")
            .arg(dir)
            .args(args)
            .env("GIT_AUTHOR_NAME", "test")
            .env("GIT_AUTHOR_EMAIL", "test@example.com")
            .env("GIT_COMMITTER_NAME", "test")
            .env("GIT_COMMITTER_EMAIL", "test@example.com")
            .output()
            .expect("git runs");
        assert!(
            out.status.success(),
            "git {args:?} failed: {}",
            String::from_utf8_lossy(&out.stderr)
        );
        String::from_utf8_lossy(&out.stdout).trim().to_string()
    }

    struct Fixture {
        _tmp: TempDir,
        state: AppState,
        repo: Repo,
        owner: User,
        bare: PathBuf,
        work: PathBuf,
    }

    async fn fixture() -> Fixture {
        let tmp = tempfile::tempdir().expect("tempdir");
        let state = test_state(tmp.path()).await;
        let owner = state
            .store
            .create_user("alice", None, "unused", false)
            .await
            .expect("user");
        let repo = state
            .store
            .create_repo(&owner.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");
        let work = tmp.path().join("work");
        git(tmp.path(), &["init", "-b", "main", "work"]);
        Fixture {
            _tmp: tmp,
            state,
            repo,
            owner,
            bare,
            work,
        }
    }

    /// Commits `files` in the work tree and returns the new commit.
    fn commit(work: &Path, files: &[(&str, &str)]) -> String {
        for (path, text) in files {
            let full = work.join(path);
            if let Some(parent) = full.parent() {
                std::fs::create_dir_all(parent).expect("dirs");
            }
            std::fs::write(&full, text).expect("write");
            git(work, &["add", path]);
        }
        git(work, &["commit", "--allow-empty", "-m", "change"]);
        git(work, &["rev-parse", "HEAD"])
    }

    /// Makes `oid` the tip of `refname` in the bare repo. Fetch runs no hooks, unlike a push.
    fn land(fx: &Fixture, oid: &str, refname: &str) {
        let work = fx.work.to_str().expect("utf8 path");
        git(
            &fx.bare,
            &["fetch", "--no-tags", work, &format!("+{oid}:{refname}")],
        );
    }

    /// Commits `files`, lands them on `refname`, and runs the post-update hook for the update.
    async fn push(fx: &Fixture, files: &[(&str, &str)], refname: &str) -> String {
        let oid = commit(&fx.work, files);
        land(fx, &oid, refname);
        after_ref_update(&fx.state, &fx.repo, &fx.owner.id, refname, ZERO, &oid)
            .await
            .expect("post update");
        oid
    }

    fn status<'a>(statuses: &'a [CommitStatus], context: &str) -> Option<&'a CommitStatus> {
        statuses.iter().find(|s| s.context == context)
    }

    fn state_of(jobs: &[CiJob], name: &str) -> Option<JobState> {
        jobs.iter().find(|j| j.name == name).map(|j| j.state)
    }

    async fn runs_total(fx: &Fixture) -> u64 {
        fx.state
            .store
            .list_runs(&fx.repo.id, 1, 50)
            .await
            .expect("list runs")
            .1
    }

    async fn runner(fx: &Fixture) -> CiRunner {
        let token = fx
            .state
            .store
            .create_registration_token(&fx.owner.id, 24)
            .await
            .expect("token");
        let (runner, _secret) = fx
            .state
            .store
            .register_runner(&token, "runner-1", &[])
            .await
            .expect("register");
        runner
    }

    #[tokio::test]
    async fn push_with_config_creates_run_and_pending_statuses() {
        let fx = fixture().await;
        let oid = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/main").await;

        let run = fx
            .state
            .store
            .run_by_number(&fx.repo.id, 1)
            .await
            .expect("query")
            .expect("run exists");
        assert_eq!(run.trigger, "push");
        assert_eq!(run.git_ref, "refs/heads/main");
        assert_eq!(run.commit_oid, oid);
        assert!(run.config_error.is_none());

        let jobs = fx.state.store.jobs_for_run(&run.id).await.expect("jobs");
        assert_eq!(state_of(&jobs, "build"), Some(JobState::Queued));
        assert_eq!(state_of(&jobs, "test"), Some(JobState::Blocked));

        let statuses = fx
            .state
            .store
            .statuses_for(&fx.repo.id, &oid)
            .await
            .expect("statuses");
        let build = status(&statuses, "ci/build").expect("ci/build status");
        assert_eq!(build.state, StatusState::Pending);
        let url = build.target_url.clone().unwrap_or_default();
        let job = jobs.iter().find(|j| j.name == "build").expect("build job");
        assert_eq!(
            url,
            format!("https://forge.test/alice/proj/ci/jobs/{}", job.id)
        );
        let test = status(&statuses, "ci/test").expect("ci/test status");
        assert_eq!(test.state, StatusState::Pending);
    }

    #[tokio::test]
    async fn no_file_means_no_run() {
        let fx = fixture().await;
        let oid = push(&fx, &[("README.md", "hello")], "refs/heads/main").await;
        assert_eq!(runs_total(&fx).await, 0);
        let statuses = fx
            .state
            .store
            .statuses_for(&fx.repo.id, &oid)
            .await
            .expect("statuses");
        assert!(statuses.is_empty());
    }

    #[tokio::test]
    async fn bad_file_gives_error_run_and_config_status() {
        let fx = fixture().await;
        let oid = push(
            &fx,
            &[(CONFIG_PATH, "[jobs.build\nimage = ")],
            "refs/heads/main",
        )
        .await;
        let run = fx
            .state
            .store
            .run_by_number(&fx.repo.id, 1)
            .await
            .expect("query")
            .expect("run exists");
        assert_eq!(run.state, RunState::Error);
        assert!(run.config_error.is_some());
        let jobs = fx.state.store.jobs_for_run(&run.id).await.expect("jobs");
        assert!(jobs.is_empty());

        let statuses = fx
            .state
            .store
            .statuses_for(&fx.repo.id, &oid)
            .await
            .expect("statuses");
        let config = status(&statuses, "ci/config").expect("ci/config status");
        assert_eq!(config.state, StatusState::Error);
        assert!(!config.description.as_deref().unwrap_or("").is_empty());
    }

    #[tokio::test]
    async fn branch_globs_select_runs() {
        let fx = fixture().await;
        push(&fx, &[(CONFIG_PATH, CONFIG_GLOBS)], "refs/heads/main").await;
        assert_eq!(runs_total(&fx).await, 0, "main is not in the globs");

        push(
            &fx,
            &[(CONFIG_PATH, CONFIG_GLOBS)],
            "refs/heads/release/1.0",
        )
        .await;
        assert_eq!(runs_total(&fx).await, 1);
        let run = fx
            .state
            .store
            .run_by_number(&fx.repo.id, 1)
            .await
            .expect("query")
            .expect("run");
        assert_eq!(run.git_ref, "refs/heads/release/1.0");
    }

    #[tokio::test]
    async fn tag_push_creates_tag_run() {
        let fx = fixture().await;
        push(&fx, &[(CONFIG_PATH, CONFIG_TAGS)], "refs/tags/nightly").await;
        assert_eq!(runs_total(&fx).await, 0, "nightly does not match v*");

        push(&fx, &[(CONFIG_PATH, CONFIG_TAGS)], "refs/tags/v1.0").await;
        assert_eq!(runs_total(&fx).await, 1);
        let run = fx
            .state
            .store
            .run_by_number(&fx.repo.id, 1)
            .await
            .expect("query")
            .expect("run");
        assert_eq!(run.trigger, "tag");
        assert_eq!(run.git_ref, "refs/tags/v1.0");
    }

    /// Opens a pull request `author` -> `branch` whose head repo is `head_repo_id`, and runs the
    /// PR hook for it.
    async fn open_pull(
        fx: &Fixture,
        author_id: &str,
        head_repo_id: &str,
        branch: &str,
        head_oid: &str,
    ) -> PullRequest {
        let pr = fx
            .state
            .store
            .create_pull(
                &fx.repo.id,
                author_id,
                "Change",
                "",
                "main",
                head_repo_id,
                branch,
                head_oid,
                head_oid,
            )
            .await
            .expect("create pull");
        on_pull_event(&fx.state, &fx.repo, &pr, author_id).await;
        pr
    }

    #[tokio::test]
    async fn fork_pull_request_waits_for_approval() {
        let fx = fixture().await;
        let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/feature").await;
        let bob = fx
            .state
            .store
            .create_user("bob", None, "unused", false)
            .await
            .expect("user");
        let fork = fx
            .state
            .store
            .create_repo(&bob.id, "proj", "", Visibility::Public, "main")
            .await
            .expect("fork");
        let pr = open_pull(&fx, &fx.owner.id, &fork.id, "feature", &head).await;

        let run = fx
            .state
            .store
            .run_by_number(&fx.repo.id, 1)
            .await
            .expect("query")
            .expect("run");
        assert_eq!(run.trigger, "pull_request");
        assert_eq!(run.state, RunState::WaitingApproval);
        assert_eq!(run.pull_number, Some(pr.issue.number));

        let statuses = fx
            .state
            .store
            .statuses_for(&fx.repo.id, &head)
            .await
            .expect("statuses");
        let build = status(&statuses, "ci/build").expect("ci/build");
        assert_eq!(build.state, StatusState::Pending);
        assert_eq!(build.description.as_deref(), Some("Waiting for approval"));
    }

    #[tokio::test]
    async fn non_writer_pull_request_waits_for_approval() {
        let fx = fixture().await;
        let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/feature").await;
        let bob = fx
            .state
            .store
            .create_user("bob", None, "unused", false)
            .await
            .expect("user");
        open_pull(&fx, &bob.id, &fx.repo.id, "feature", &head).await;
        let run = fx
            .state
            .store
            .run_by_number(&fx.repo.id, 1)
            .await
            .expect("query")
            .expect("run");
        assert_eq!(run.state, RunState::WaitingApproval);
    }

    #[tokio::test]
    async fn writer_pull_request_runs_at_once() {
        let fx = fixture().await;
        let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/feature").await;
        open_pull(&fx, &fx.owner.id, &fx.repo.id, "feature", &head).await;
        let run = fx
            .state
            .store
            .run_by_number(&fx.repo.id, 1)
            .await
            .expect("query")
            .expect("run");
        assert_ne!(run.state, RunState::WaitingApproval);
    }

    #[tokio::test]
    async fn manual_run_uses_branch_head() {
        let fx = fixture().await;
        let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/main").await;
        let before = runs_total(&fx).await;
        let run = manual_run(&fx.state, &fx.repo, "main", &fx.owner.id)
            .await
            .expect("manual run");
        assert_eq!(run.trigger, "manual");
        assert_eq!(run.git_ref, "refs/heads/main");
        assert_eq!(run.commit_oid, head);
        assert_eq!(runs_total(&fx).await, before + 1);
    }

    #[tokio::test]
    async fn manual_run_refuses_missing_branch_and_disabled_manual() {
        let fx = fixture().await;
        push(&fx, &[(CONFIG_PATH, CONFIG_NO_MANUAL)], "refs/heads/main").await;
        let before = runs_total(&fx).await;
        assert!(
            manual_run(&fx.state, &fx.repo, "nope", &fx.owner.id)
                .await
                .is_err()
        );
        let refused = manual_run(&fx.state, &fx.repo, "main", &fx.owner.id).await;
        assert!(matches!(refused, Err(forge_core::Error::Invalid(_))));
        assert_eq!(runs_total(&fx).await, before);
    }

    #[tokio::test]
    async fn reaper_marks_lost_jobs_and_syncs_statuses() {
        let fx = fixture().await;
        let head = push(&fx, &[(CONFIG_PATH, CONFIG)], "refs/heads/main").await;
        let runner = runner(&fx).await;
        let (job, _run, _token) = fx
            .state
            .store
            .claim_job(&runner)
            .await
            .expect("claim")
            .expect("a queued job");
        assert_eq!(job.name, "build");

        // A cutoff in the future makes every running job look silent.
        let reaped = reap_before(&fx.state, &stamp_after(Duration::from_secs(3600))).await;
        assert_eq!(reaped, 1);

        let job = fx
            .state
            .store
            .job_by_id(&job.id)
            .await
            .expect("query")
            .expect("job");
        assert_eq!(job.state, JobState::Error);
        assert_eq!(job.message.as_deref(), Some("runner lost"));

        let statuses = fx
            .state
            .store
            .statuses_for(&fx.repo.id, &head)
            .await
            .expect("statuses");
        let build = status(&statuses, "ci/build").expect("ci/build");
        assert_eq!(build.state, StatusState::Error);
    }

    #[tokio::test]
    async fn purge_deletes_log_files() {
        let fx = fixture().await;
        push(&fx, &[(CONFIG_PATH, CONFIG_ONE_JOB)], "refs/heads/main").await;
        let runner = runner(&fx).await;
        let (job, _run, _token) = fx
            .state
            .store
            .claim_job(&runner)
            .await
            .expect("claim")
            .expect("a queued job");
        // Purge only takes jobs that wrote a log and whose run has finished.
        fx.state
            .store
            .add_log_bytes(&job.id, 12)
            .await
            .expect("log bytes");
        fx.state
            .store
            .finish_job(&job.id, JobState::Success, Some(0), None)
            .await
            .expect("finish");

        let path = log_path(&fx.state.config.data_dir, &job.id);
        std::fs::create_dir_all(path.parent().expect("parent")).expect("log dir");
        std::fs::write(&path, "build output\n").expect("write log");

        // Zero days of retention: everything finished before now is purged.
        let mut cfg = (*fx.state.config).clone();
        cfg.ci.log_retention_days = 0;
        let mut state = fx.state.clone();
        state.config = Arc::new(cfg);
        tokio::time::sleep(Duration::from_millis(1100)).await;

        let (logs, _artifacts) = purge_once(&state).await;
        assert_eq!(logs, 1);
        assert!(!path.exists(), "log file is deleted");
    }
}