//! Internal hook API, served over a unix socket at `data_dir/internal.sock`.
//!
//! Routes are registered explicitly with [`RouteFn`] instead of `#[route]`. The
//! `#[route]` macro emits an `inventory::submit!` that `Router::builder().discover()`
//! collects, and the public TCP router uses `.discover()`. A `RouteFn` passed to
//! `.route(...)` never enters that inventory, so these endpoints are not reachable
//! on the public listener.

use std::collections::{BTreeMap, HashMap};
use std::path::Path;
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError};

use forge_core::protection::effective;
use forge_core::{Repo, Role, Store};
use forge_git::{GitCommand, GitEnv};
use serde::{Deserialize, Serialize};
use topcoat::context::{Cx, app_context};
use topcoat::router::content::Json;
use topcoat::router::error::{bad_request, forbidden};
use topcoat::router::request::headers;
use topcoat::router::response::IntoResponse;
use topcoat::router::response::Response;
use topcoat::router::{
    Body, HeaderValue, Method, RouteFn, RouteFuture, Router, StatusCode, header, to_bytes,
};

use crate::admin_client::ADMIN_TOKEN_HEADER;
use crate::audit::{self, Actor, Target};
use crate::backup::scheduler::{run_backup_now, run_fsck_now};
use crate::backup::{BackupOptions, Scope};
use crate::post_update::after_ref_update;
use crate::state::AppState;
use forge_core::AuditAction;
use forge_core::store::backup_runs::{RunKind, RunTrigger};

/// Largest accepted hook request body.
const MAX_BODY_BYTES: usize = 1024 * 1024;
/// Header carrying `AppState::internal_token`.
const TOKEN_HEADER: &str = "x-forge-internal-token";
/// Environment variables git sets for the quarantine directory of a push.
pub const QUARANTINE_VARS: [&str; 3] = [
    "GIT_QUARANTINE_PATH",
    "GIT_OBJECT_DIRECTORY",
    "GIT_ALTERNATE_OBJECT_DIRECTORIES",
];

/// One ref update from receive-pack.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct HookUpdate {
    pub old: String,
    pub new: String,
    #[serde(rename = "ref")]
    pub ref_name: String,
}

/// Body of `POST /hook/pre-receive` and `POST /hook/post-receive`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HookRequest {
    pub repo_id: String,
    pub actor_id: String,
    pub protocol: String,
    /// True when the push targets the repo's wiki repo instead of the code repo (see forge-git
    /// `wiki.rs`). Wiki pushes skip the branch rules and the post-push work, and may only touch main.
    #[serde(default)]
    pub wiki: bool,
    pub updates: Vec<HookUpdate>,
    #[serde(default)]
    pub quarantine_env: BTreeMap<String, String>,
}

/// Response of both hook endpoints. `message` is shown to the pusher on stderr.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct HookResponse {
    pub allow: bool,
    pub message: String,
}

impl HookResponse {
    fn allow() -> Self {
        Self {
            allow: true,
            message: String::new(),
        }
    }

    fn deny(message: impl Into<String>) -> Self {
        Self {
            allow: false,
            message: message.into(),
        }
    }
}

/// Router for the internal API. Serve it on the unix socket only.
pub fn router(state: AppState) -> Router {
    Router::builder()
        .route(RouteFn::new(Method::POST, "/hook/pre-receive", pre_receive))
        .route(RouteFn::new(
            Method::POST,
            "/hook/post-receive",
            post_receive,
        ))
        .route(RouteFn::new(
            Method::POST,
            "/internal/admin/backup",
            admin_backup,
        ))
        .route(RouteFn::new(
            Method::POST,
            "/internal/admin/fsck",
            admin_fsck,
        ))
        .route(RouteFn::new(
            Method::POST,
            "/internal/admin/lfs-gc",
            admin_lfs_gc,
        ))
        .app_context(state)
        .build()
}

fn pre_receive(cx: &Cx, body: Body) -> RouteFuture<'_> {
    Box::pin(async move { hook_endpoint(cx, body, Stage::PreReceive).await })
}

fn post_receive(cx: &Cx, body: Body) -> RouteFuture<'_> {
    Box::pin(async move { hook_endpoint(cx, body, Stage::PostReceive).await })
}

#[derive(Clone, Copy)]
enum Stage {
    PreReceive,
    PostReceive,
}

async fn hook_endpoint(
    cx: &Cx,
    body: Body,
    stage: Stage,
) -> topcoat::Result<topcoat::router::response::Response> {
    let state = app_context::<AppState>(cx);
    let presented = headers(cx)
        .get(TOKEN_HEADER)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    if !forge_core::token::constant_time_eq(presented.as_bytes(), state.internal_token.as_bytes()) {
        return Err(forbidden().into());
    }

    let bytes = to_bytes(body, MAX_BODY_BYTES).await?;
    let req: HookRequest = serde_json::from_slice(&bytes)
        .map_err(|e| bad_request(format!("invalid hook request: {e}")))?;

    let verdict = match stage {
        Stage::PreReceive => match evaluate_pre_receive(&state.store, &state.git, &req).await {
            Ok(verdict) => verdict,
            Err(e) => {
                // Fail closed: an internal error rejects the push.
                tracing::error!(error = %e, repo_id = %req.repo_id, "pre-receive check failed");
                HookResponse::deny("internal error while checking the push")
            }
        },
        Stage::PostReceive if req.wiki => {
            // A wiki push changes neither the code repo's refs nor its CI, pull requests or events.
            // It still goes to the standby.
            record_wiki_push(&state.store, &req).await;
            crate::replication::enqueue_repo(state, &req.repo_id);
            HookResponse::allow()
        }
        Stage::PostReceive => {
            let verdict = record_events(&state.store, &req).await;
            trigger_maintenance(&state.git, &req.repo_id);
            after_push_refs(state, &req).await;
            verdict
        }
    };
    Json(verdict).into_response(cx)
}

fn admin_backup(cx: &Cx, body: Body) -> RouteFuture<'_> {
    Box::pin(async move { admin_backup_endpoint(cx, body).await })
}

fn admin_fsck(cx: &Cx, body: Body) -> RouteFuture<'_> {
    Box::pin(async move { admin_fsck_endpoint(cx, body).await })
}

fn admin_lfs_gc(cx: &Cx, body: Body) -> RouteFuture<'_> {
    Box::pin(async move { admin_lfs_gc_endpoint(cx, body).await })
}

/// `POST /internal/admin/lfs-gc`. The body is ignored.
async fn admin_lfs_gc_endpoint(cx: &Cx, _body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    if !admin_authorized(cx, state) {
        return Err(forbidden().into());
    }
    Ok(
        match crate::lfs::gc::gc(
            &state.store,
            &state.config.data_dir,
            crate::lfs::gc::GC_MIN_AGE,
        )
        .await
        {
            Ok(summary) => admin_response(StatusCode::OK, serde_json::json!(summary)),
            Err(e) => admin_error(&e),
        },
    )
}

/// Body of `POST /internal/admin/backup`. `out` is the directory that receives the set.
#[derive(Deserialize)]
struct AdminBackupRequest {
    out: String,
    #[serde(default)]
    include_key: bool,
    #[serde(default)]
    scope: Option<Scope>,
}

/// Body of `POST /internal/admin/fsck`.
#[derive(Deserialize)]
struct AdminFsckRequest {
    #[serde(default)]
    full: bool,
}

/// Admin routes need the per-boot admin token. The socket itself is 0600 in the 0700 data dir.
fn admin_authorized(cx: &Cx, state: &AppState) -> bool {
    let presented = headers(cx)
        .get(ADMIN_TOKEN_HEADER)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    forge_core::token::constant_time_eq(presented.as_bytes(), state.admin_token.as_bytes())
}

async fn admin_backup_endpoint(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    if !admin_authorized(cx, state) {
        return Err(forbidden().into());
    }
    let bytes = to_bytes(body, MAX_BODY_BYTES).await?;
    let req: AdminBackupRequest = serde_json::from_slice(&bytes)
        .map_err(|e| bad_request(format!("invalid backup request: {e}")))?;
    let out = match std::path::absolute(&req.out) {
        Ok(out) => out,
        Err(e) => return Ok(admin_error(&forge_core::Error::Io(e))),
    };
    let guard = match state.jobs.claim(RunKind::Backup) {
        Ok(guard) => guard,
        Err(e) => return Ok(admin_error(&e)),
    };
    let target_name = out.to_string_lossy().into_owned();
    audit::record(
        &state.store,
        None,
        Actor::Cli,
        AuditAction::BackupStarted,
        Target {
            kind: "backup",
            id: "",
            name: &target_name,
        },
        serde_json::json!({ "trigger": "cli" }),
    )
    .await;
    let opts = BackupOptions {
        out,
        include_key: req.include_key,
        scope: req.scope.unwrap_or(Scope::All),
    };
    let result = run_backup_now(state, guard, RunTrigger::Cli, opts, false).await;
    Ok(match result {
        Ok(summary) => admin_response(StatusCode::OK, serde_json::json!(summary)),
        Err(e) => admin_error(&e),
    })
}

async fn admin_fsck_endpoint(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    if !admin_authorized(cx, state) {
        return Err(forbidden().into());
    }
    let bytes = to_bytes(body, MAX_BODY_BYTES).await?;
    let req: AdminFsckRequest = if bytes.is_empty() {
        AdminFsckRequest { full: false }
    } else {
        serde_json::from_slice(&bytes)
            .map_err(|e| bad_request(format!("invalid fsck request: {e}")))?
    };
    let guard = match state.jobs.claim(RunKind::Fsck) {
        Ok(guard) => guard,
        Err(e) => return Ok(admin_error(&e)),
    };
    audit::record(
        &state.store,
        None,
        Actor::Cli,
        AuditAction::FsckStarted,
        Target {
            kind: "fsck",
            id: "",
            name: "",
        },
        serde_json::json!({ "trigger": "cli", "full": req.full }),
    )
    .await;
    let result = run_fsck_now(state, guard, RunTrigger::Cli, req.full).await;
    Ok(match result {
        Ok(summary) => admin_response(StatusCode::OK, serde_json::json!(summary)),
        Err(e) => admin_error(&e),
    })
}

/// A JSON answer with the given status.
fn admin_response(status: StatusCode, value: serde_json::Value) -> Response {
    let mut res = Response::new(Body::from(value.to_string()));
    *res.status_mut() = status;
    res.headers_mut().insert(
        header::CONTENT_TYPE,
        HeaderValue::from_static("application/json"),
    );
    res
}

/// Maps a failed admin action to a status and a `{"error": ...}` body.
fn admin_error(e: &forge_core::Error) -> Response {
    let status = match e {
        forge_core::Error::Conflict(_) => StatusCode::CONFLICT,
        forge_core::Error::Invalid(_) | forge_core::Error::Config(_) => StatusCode::BAD_REQUEST,
        forge_core::Error::Forbidden(_) => StatusCode::FORBIDDEN,
        forge_core::Error::NotFound => StatusCode::NOT_FOUND,
        _ => StatusCode::INTERNAL_SERVER_ERROR,
    };
    let message = match e {
        forge_core::Error::Conflict(text) => text.clone(),
        other => other.to_string(),
    };
    admin_response(status, serde_json::json!({ "error": message }))
}

async fn record_events(store: &Store, req: &HookRequest) -> HookResponse {
    let mut failures = Vec::new();
    for update in &req.updates {
        let recorded = store
            .record_push_event(
                &req.repo_id,
                &req.actor_id,
                &update.ref_name,
                &update.old,
                &update.new,
                &req.protocol,
            )
            .await;
        if let Err(e) = recorded {
            tracing::error!(error = %e, repo_id = %req.repo_id, ref_name = %update.ref_name, "failed to record push event");
            failures.push(format!(
                "warning: could not record push to {:?}",
                update.ref_name
            ));
        }
    }
    // post-receive cannot reject a push, so the verdict is always allow.
    HookResponse {
        allow: true,
        message: failures.join("\n"),
    }
}

/// Repos with post-push maintenance in flight. The value is true when another push
/// arrived while the run was active, so one rerun is due.
static MAINTENANCE: OnceLock<Mutex<HashMap<String, bool>>> = OnceLock::new();

fn maintenance_state() -> MutexGuard<'static, HashMap<String, bool>> {
    MAINTENANCE
        .get_or_init(|| Mutex::new(HashMap::new()))
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
}

/// Marks `repo_id` in flight. Returns true when the caller must start the worker; otherwise
/// the running worker is flagged to run once more.
fn claim_maintenance(map: &mut HashMap<String, bool>, repo_id: &str) -> bool {
    match map.get_mut(repo_id) {
        Some(dirty) => {
            *dirty = true;
            false
        }
        None => {
            map.insert(repo_id.to_owned(), false);
            true
        }
    }
}

/// Called after a run. Returns true when a rerun is due (the entry stays claimed);
/// otherwise releases the repo.
fn finish_maintenance(map: &mut HashMap<String, bool>, repo_id: &str) -> bool {
    let rerun = map
        .get_mut(repo_id)
        .map(|dirty| std::mem::replace(dirty, false));
    if rerun == Some(true) {
        return true;
    }
    map.remove(repo_id);
    false
}

/// Starts `maintenance::after_push` for the repo in the background, coalescing concurrent calls
/// per repo. Failures are logged, never returned, so the push response is not failed. Called once
/// per push by the post-receive hook, and by the web merge route after it updates a branch.
pub fn trigger_maintenance(git: &Arc<GitEnv>, repo_id: &str) {
    let repo_dir = match forge_git::repo::repo_path(&git.data_dir, repo_id) {
        Ok(dir) => dir,
        Err(e) => {
            tracing::error!(error = %e, repo_id = %repo_id, "skipping post-push maintenance");
            return;
        }
    };
    if !claim_maintenance(&mut maintenance_state(), repo_id) {
        return;
    }
    let git = Arc::clone(git);
    let repo_id = repo_id.to_owned();
    tokio::spawn(async move {
        loop {
            if let Err(e) = forge_git::maintenance::after_push(&git, &repo_dir).await {
                tracing::error!(error = %e, repo_id = %repo_id, "post-push maintenance failed");
            }
            let rerun = finish_maintenance(&mut maintenance_state(), &repo_id);
            if !rerun {
                break;
            }
        }
    });
}

/// Runs `post_update::after_ref_update` for every ref the push updated. Errors are logged; the
/// push itself is already accepted at this point.
async fn after_push_refs(state: &AppState, req: &HookRequest) {
    let repo: Repo = match state.store.repo_by_id(&req.repo_id).await {
        Ok(Some(repo)) => repo,
        Ok(None) => return,
        Err(e) => {
            tracing::error!(error = %e, repo_id = %req.repo_id, "post-update: repo lookup failed");
            return;
        }
    };
    for update in &req.updates {
        if let Err(e) = after_ref_update(
            state,
            &repo,
            &req.actor_id,
            &update.ref_name,
            &update.old,
            &update.new,
        )
        .await
        {
            tracing::error!(error = %e, repo_id = %req.repo_id, ref_name = %update.ref_name, "post-update failed");
        }
    }
}

/// Writes a `wiki_pushed` audit entry for an accepted wiki push. The pusher is looked up by id; a
/// push with no known user (a CI job cannot push) is recorded as anonymous.
async fn record_wiki_push(store: &Store, req: &HookRequest) {
    let repo = match store.repo_by_id(&req.repo_id).await {
        Ok(Some(repo)) => repo,
        Ok(None) => return,
        Err(e) => {
            tracing::error!(error = %e, repo_id = %req.repo_id, "could not load repo for wiki audit");
            return;
        }
    };
    let user = match store.user_by_id(&req.actor_id).await {
        Ok(user) => user,
        Err(e) => {
            tracing::error!(error = %e, actor_id = %req.actor_id, "could not load pusher for wiki audit");
            None
        }
    };
    let actor = match user.as_ref() {
        Some(user) => Actor::User(user),
        None => Actor::Anonymous,
    };
    let updates: Vec<serde_json::Value> = req
        .updates
        .iter()
        .map(|u| serde_json::json!({ "ref": u.ref_name, "old": u.old, "new": u.new }))
        .collect();
    audit::record(
        store,
        None,
        actor,
        AuditAction::WikiPushed,
        Target {
            kind: "repo",
            id: &repo.id,
            name: &format!("{}/{}", repo.owner_name, repo.name),
        },
        serde_json::json!({ "updates": updates }),
    )
    .await;
}

/// Decides whether a push may proceed. `Ok(deny)` is a policy decision; `Err` is an
/// internal failure (the caller turns it into a deny).
pub async fn evaluate_pre_receive(
    store: &Store,
    git: &GitEnv,
    req: &HookRequest,
) -> forge_core::Result<HookResponse> {
    let Some(repo) = store.repo_by_id(&req.repo_id).await? else {
        return Ok(HookResponse::deny("repository not found"));
    };
    if repo.archived {
        return Ok(HookResponse::deny("repository is archived"));
    }
    if req.wiki {
        if !store.wiki_enabled(&repo.id).await? {
            return Ok(HookResponse::deny("the wiki is disabled"));
        }
    } else if store.pull_mirror(&repo.id).await?.is_some() {
        // A pull mirror's refs come from its remote; a push here would be overwritten by the next sync.
        // Mirrors do not mirror wikis, so wiki pushes are not refused for this.
        return Ok(HookResponse::deny("this repository is a pull mirror"));
    }
    let repo_dir = if req.wiki {
        forge_git::repo::wiki_path(&git.data_dir, &repo.id)?
    } else {
        forge_git::repo::repo_path(&git.data_dir, &repo.id)?
    };
    // Loaded once per push; every branch update below is checked against the same rules. Wiki
    // pushes skip the rules: the wiki has no protected branches, only its main branch.
    let rules = if req.wiki {
        Vec::new()
    } else {
        store.list_branch_rules(&repo.id).await?
    };

    for update in &req.updates {
        if !is_valid_ref_name(&update.ref_name) {
            return Ok(HookResponse::deny(format!(
                "invalid ref name: {:?}",
                update.ref_name
            )));
        }
        if !is_valid_oid(&update.old) || !is_valid_oid(&update.new) {
            return Ok(HookResponse::deny(format!(
                "invalid object id in update to {:?}",
                update.ref_name
            )));
        }
        if req.wiki {
            if update.ref_name != "refs/heads/main" {
                return Ok(HookResponse::deny(
                    "the wiki only accepts changes to its main branch",
                ));
            }
            if is_zero_oid(&update.new) {
                return Ok(HookResponse::deny(
                    "the wiki's main branch cannot be deleted",
                ));
            }
            continue;
        }
        let Some(branch) = update.ref_name.strip_prefix("refs/heads/") else {
            continue;
        };
        let protection = effective(&rules, &repo.default_branch, branch);
        if protection.restrict_push && !may_push_directly(store, &repo, &req.actor_id).await? {
            return Ok(HookResponse::deny(format!(
                "branch '{branch}' only accepts changes through pull requests"
            )));
        }
        if is_zero_oid(&update.new) {
            if !protection.allow_deletion {
                return Ok(HookResponse::deny(format!(
                    "deleting protected branch '{branch}' is not allowed"
                )));
            }
            continue;
        }
        if is_zero_oid(&update.old) || protection.allow_force_push {
            continue;
        }
        match drops_commits(git, &repo_dir, req, update).await {
            Ok(false) => {}
            Ok(true) => {
                return Ok(HookResponse::deny(format!(
                    "force-push to protected branch '{branch}' is not allowed"
                )));
            }
            Err(e) => {
                return Ok(HookResponse::deny(format!(
                    "could not verify the update to '{branch}': {e}"
                )));
            }
        }
    }
    Ok(HookResponse::allow())
}

/// Whether `actor_id` may push straight to a `restrict_push` branch: an instance admin, the owner
/// of a person-owned repo, or someone whose effective role is Admin (a collaborator Admin, or an
/// org owner or admin team on an org-owned repo). Pull request merges do not go through here.
async fn may_push_directly(store: &Store, repo: &Repo, actor_id: &str) -> forge_core::Result<bool> {
    let Some(actor) = store.user_by_id(actor_id).await? else {
        return Ok(false);
    };
    if actor.is_admin || actor.id == repo.owner_id {
        return Ok(true);
    }
    let collaborator = store.collaborator_role(&repo.id, actor_id).await?;
    let org = store
        .org_repo_role(&repo.owner_id, actor_id, &repo.id)
        .await?;
    Ok(forge_core::authz::max_role(collaborator, org) == Some(Role::Admin))
}

/// True when `old` has commits that `new` does not contain, i.e. the update
/// is not a fast-forward. Runs `git rev-list --max-count=1 <old> ^<new>` with the
/// quarantine environment so that objects received in this push are visible.
async fn drops_commits(
    git: &GitEnv,
    repo_dir: &Path,
    req: &HookRequest,
    update: &HookUpdate,
) -> forge_git::Result<bool> {
    let excluded = format!("^{}", update.new);
    let mut cmd = GitCommand::new(git, "rev-list")
        .git_dir(repo_dir)
        .arg("--max-count=1")
        .user_revs(&[update.old.as_str(), excluded.as_str()])?;
    for key in QUARANTINE_VARS {
        if let Some(value) = req.quarantine_env.get(key) {
            cmd = cmd.env(key, value);
        }
    }
    let stdout = cmd.output().await?;
    Ok(!stdout.is_empty())
}

/// Accepts `refs/...` names without `..`, without a `-`-prefixed component, and
/// without control characters.
pub fn is_valid_ref_name(name: &str) -> bool {
    let Some(rest) = name.strip_prefix("refs/") else {
        return false;
    };
    !rest.is_empty()
        && !name.contains("..")
        && !name.chars().any(char::is_control)
        && !name.split('/').any(|component| component.starts_with('-'))
}

/// True for the all-zero object id (40 or 64 hex digits), which git uses for
/// creation (`old`) and deletion (`new`).
pub fn is_zero_oid(oid: &str) -> bool {
    matches!(oid.len(), 40 | 64) && oid.bytes().all(|b| b == b'0')
}

/// True for a SHA-1 or SHA-256 object id: 40 or 64 hex digits.
pub fn is_valid_oid(oid: &str) -> bool {
    matches!(oid.len(), 40 | 64) && oid.bytes().all(|b| b.is_ascii_hexdigit())
}

#[cfg(test)]
mod tests {
    use super::*;
    use forge_core::protection::RuleFlags;
    use std::path::PathBuf;
    use topcoat::router::RouterBuilderDiscoverExt;

    #[test]
    fn ref_names_must_live_under_refs() {
        assert!(is_valid_ref_name("refs/heads/main"));
        assert!(is_valid_ref_name("refs/tags/v1.0"));
        assert!(!is_valid_ref_name("HEAD"));
        assert!(!is_valid_ref_name("heads/main"));
        assert!(!is_valid_ref_name("refs/"));
    }

    #[test]
    fn ref_names_reject_dotdot_dash_components_and_control_chars() {
        assert!(!is_valid_ref_name("refs/heads/a..b"));
        assert!(!is_valid_ref_name("refs/heads/-evil"));
        assert!(!is_valid_ref_name("refs/-evil/main"));
        assert!(is_valid_ref_name("refs/heads/a-b"));
        assert!(!is_valid_ref_name("refs/heads/a\nb"));
        assert!(!is_valid_ref_name("refs/heads/a\0b"));
        assert!(!is_valid_ref_name("refs/heads/a\x7fb"));
    }

    #[test]
    fn zero_oid_detection() {
        assert!(is_zero_oid(&"0".repeat(40)));
        assert!(is_zero_oid(&"0".repeat(64)));
        assert!(!is_zero_oid(&"0".repeat(39)));
        assert!(!is_zero_oid(&format!("{}1", "0".repeat(39))));
        assert!(!is_zero_oid(""));
    }

    #[test]
    fn oid_validation() {
        assert!(is_valid_oid(&"a".repeat(40)));
        assert!(is_valid_oid(&"F".repeat(64)));
        assert!(is_valid_oid(&"0".repeat(40)));
        assert!(!is_valid_oid(&"g".repeat(40)));
        assert!(!is_valid_oid(&"a".repeat(41)));
        assert!(!is_valid_oid("--all"));
    }

    #[test]
    fn maintenance_claims_coalesce_per_repo() {
        let mut map = HashMap::new();
        // First push starts a worker; pushes during the run only flag a rerun.
        assert!(claim_maintenance(&mut map, "r1"));
        assert!(!claim_maintenance(&mut map, "r1"));
        assert!(!claim_maintenance(&mut map, "r1"));
        // Other repos are independent.
        assert!(claim_maintenance(&mut map, "r2"));
        // One rerun is due after the first run, then the repo is released.
        assert!(finish_maintenance(&mut map, "r1"));
        assert!(!finish_maintenance(&mut map, "r1"));
        assert!(!map.contains_key("r1"));
        assert!(claim_maintenance(&mut map, "r1"));
        assert!(!finish_maintenance(&mut map, "r2"));
    }

    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()
    }

    fn commit(work: &Path, contents: &str) -> String {
        std::fs::write(work.join("file.txt"), contents).expect("write file");
        git(work, &["add", "file.txt"]);
        git(work, &["commit", "-m", contents]);
        git(work, &["rev-parse", "HEAD"])
    }

    #[tokio::test]
    async fn admin_routes_reject_a_missing_or_wrong_token() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = crate::test_support::test_state(dir.path()).await;
        let router = router(state);
        for path in ["/internal/admin/backup", "/internal/admin/fsck"] {
            for token in [None, Some("wrong")] {
                let mut builder = topcoat::router::request::Request::builder()
                    .method("POST")
                    .uri(path)
                    .header("content-type", "application/json");
                if let Some(token) = token {
                    builder = builder.header(ADMIN_TOKEN_HEADER, token);
                }
                let req = builder
                    .body(Body::from(r#"{"out":"/nonexistent/forge-test"}"#))
                    .expect("request");
                let res = router.handle(req).await;
                assert_eq!(res.status(), 403, "{path} with token {token:?}");
            }
        }
        // The right token gets through: an fsck of an instance with no repos is clean.
        let req = topcoat::router::request::Request::builder()
            .method("POST")
            .uri("/internal/admin/fsck")
            .header("content-type", "application/json")
            .header(ADMIN_TOKEN_HEADER, crate::test_support::ADMIN_TOKEN)
            .body(Body::from("{}"))
            .expect("request");
        let res = router.handle(req).await;
        assert_eq!(res.status(), 200);
    }

    #[tokio::test]
    async fn hook_routes_are_not_collected_by_discover() {
        // The public router uses `.discover()`. Hook routes are plain RouteFns and must not be in it.
        let public = Router::builder().discover().build();
        for path in ["/hook/pre-receive", "/hook/post-receive"] {
            let req = topcoat::router::request::Request::builder()
                .method("POST")
                .uri(path)
                .body(Body::empty())
                .expect("request");
            let res = public.handle(req).await;
            assert_eq!(
                res.status(),
                404,
                "{path} is reachable on the discover router"
            );
        }
    }

    #[tokio::test]
    async fn pre_receive_restrict_push_allows_only_repo_admins() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let store = Store::open(&tmp.path().join("forge.db"))
            .await
            .expect("store");
        let owner = store
            .create_user("alice", None, "unused", false)
            .await
            .expect("owner");
        let writer = store
            .create_user("bob", None, "unused", false)
            .await
            .expect("writer");
        let admin = store
            .create_user("carol", None, "unused", false)
            .await
            .expect("admin");
        let repo = store
            .create_repo(
                &owner.id,
                "proj",
                "",
                forge_core::Visibility::Private,
                "main",
            )
            .await
            .expect("repo");
        store
            .set_collaborator(&repo.id, &writer.id, Role::Write)
            .await
            .expect("writer role");
        store
            .set_collaborator(&repo.id, &admin.id, Role::Admin)
            .await
            .expect("admin role");
        let git_env = GitEnv {
            git_bin: PathBuf::from("git"),
            data_dir: tmp.path().to_path_buf(),
        };
        let bare = forge_git::repo::repo_path(&git_env.data_dir, &repo.id).expect("repo path");
        forge_git::repo::init_bare(&git_env, &bare, "main")
            .await
            .expect("init bare");
        let bare_str = bare.to_str().expect("utf8 path");
        let work = tmp.path().join("work");
        git(tmp.path(), &["init", "-b", "main", "work"]);
        let first = commit(&work, "one");
        git(&work, &["push", bare_str, "main"]);
        let second = commit(&work, "two");
        git(&work, &["push", bare_str, "main"]);

        store
            .upsert_branch_rule(
                &repo.id,
                "main",
                RuleFlags {
                    restrict_push: true,
                    ..RuleFlags::default()
                },
            )
            .await
            .expect("upsert rule");

        let request = |actor: &str, old: &str, new: &str| HookRequest {
            repo_id: repo.id.clone(),
            actor_id: actor.into(),
            protocol: "http".into(),
            wiki: false,
            updates: vec![HookUpdate {
                old: old.into(),
                new: new.into(),
                ref_name: "refs/heads/main".into(),
            }],
            quarantine_env: BTreeMap::new(),
        };

        // A Write collaborator's direct fast-forward is refused.
        let refused = evaluate_pre_receive(&store, &git_env, &request(&writer.id, &first, &second))
            .await
            .expect("evaluate");
        assert!(!refused.allow);
        assert_eq!(
            refused.message,
            "branch 'main' only accepts changes through pull requests"
        );
        // So is creating the branch.
        let created = evaluate_pre_receive(
            &store,
            &git_env,
            &request(&writer.id, &"0".repeat(40), &second),
        )
        .await
        .expect("evaluate");
        assert!(!created.allow);

        // Repo admins (Admin collaborators and the owner) may still push.
        for actor in [&admin.id, &owner.id] {
            let ok = evaluate_pre_receive(&store, &git_env, &request(actor, &first, &second))
                .await
                .expect("evaluate");
            assert_eq!(ok, HookResponse::allow());
        }
    }

    #[tokio::test]
    async fn pre_receive_rejects_pushes_to_a_pull_mirror() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let store = Store::open(&tmp.path().join("forge.db"))
            .await
            .expect("store");
        let owner = store
            .create_user("alice", None, "unused", false)
            .await
            .expect("user");
        let repo = store
            .create_repo(
                &owner.id,
                "mirror",
                "",
                forge_core::Visibility::Public,
                "main",
            )
            .await
            .expect("repo");
        let git_env = GitEnv {
            git_bin: "git".into(),
            data_dir: tmp.path().to_path_buf(),
        };
        let bare = forge_git::repo::repo_path(&git_env.data_dir, &repo.id).expect("repo path");
        forge_git::repo::init_bare(&git_env, &bare, "main")
            .await
            .expect("init bare");
        let request = HookRequest {
            repo_id: repo.id.clone(),
            actor_id: owner.id.clone(),
            protocol: "http".into(),
            wiki: false,
            updates: vec![HookUpdate {
                old: "0".repeat(40),
                new: "1".repeat(40),
                ref_name: "refs/heads/main".into(),
            }],
            quarantine_env: BTreeMap::new(),
        };
        // Before the pull mirror exists, the push is not refused for that reason.
        let before = evaluate_pre_receive(&store, &git_env, &request)
            .await
            .expect("evaluate");
        assert_ne!(before.message, "this repository is a pull mirror");

        store
            .create_mirror(&forge_core::mirrors::NewMirror {
                repo_id: &repo.id,
                direction: forge_core::mirrors::MirrorDirection::Pull,
                url: "https://example.com/upstream.git",
                username: "",
                secret: None,
                interval_minutes: 480,
            })
            .await
            .expect("pull mirror");
        let refused = evaluate_pre_receive(&store, &git_env, &request)
            .await
            .expect("evaluate");
        assert!(!refused.allow);
        assert_eq!(refused.message, "this repository is a pull mirror");
    }

    #[tokio::test]
    async fn pre_receive_detects_non_fast_forward_on_default_branch() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let store = Store::open(&tmp.path().join("forge.db"))
            .await
            .expect("store");
        let owner = store
            .create_user("alice", None, "unused", false)
            .await
            .expect("user");
        let repo = store
            .create_repo(
                &owner.id,
                "proj",
                "",
                forge_core::Visibility::Private,
                "main",
            )
            .await
            .expect("repo");
        let git_env = GitEnv {
            git_bin: PathBuf::from("git"),
            data_dir: tmp.path().to_path_buf(),
        };

        // Bare repo at the path forge would use for this repo id.
        let bare = forge_git::repo::repo_path(&git_env.data_dir, &repo.id).expect("repo path");
        forge_git::repo::init_bare(&git_env, &bare, "main")
            .await
            .expect("init bare");
        let bare_str = bare.to_str().expect("utf8 path");

        // Two commits on main, then a divergent commit on a side branch.
        let work = tmp.path().join("work");
        git(tmp.path(), &["init", "-b", "main", "work"]);
        let first = commit(&work, "one");
        git(&work, &["push", bare_str, "main"]);
        let second = commit(&work, "two");
        git(&work, &["push", bare_str, "main"]);
        git(&work, &["reset", "--hard", &first]);
        let divergent = commit(&work, "three");
        git(
            &work,
            &["push", bare_str, &format!("{divergent}:refs/heads/side")],
        );

        let request = |old: &str, new: &str| HookRequest {
            repo_id: repo.id.clone(),
            actor_id: owner.id.clone(),
            protocol: "http".into(),
            wiki: false,
            updates: vec![HookUpdate {
                old: old.into(),
                new: new.into(),
                ref_name: "refs/heads/main".into(),
            }],
            quarantine_env: BTreeMap::new(),
        };

        // Fast-forward: allowed.
        let ff = evaluate_pre_receive(&store, &git_env, &request(&first, &second))
            .await
            .expect("evaluate");
        assert_eq!(ff, HookResponse::allow());

        // Non-fast-forward: denied.
        let forced = evaluate_pre_receive(&store, &git_env, &request(&second, &divergent))
            .await
            .expect("evaluate");
        assert!(!forced.allow);
        assert_eq!(
            forced.message,
            "force-push to protected branch 'main' is not allowed"
        );

        // Deleting the default branch: denied.
        let deleted = evaluate_pre_receive(&store, &git_env, &request(&second, &"0".repeat(40)))
            .await
            .expect("evaluate");
        assert!(!deleted.allow);
        assert_eq!(
            deleted.message,
            "deleting protected branch 'main' is not allowed"
        );

        // A rule that allows force pushes lifts only that check; deletion stays denied.
        store
            .upsert_branch_rule(&repo.id, "main", RuleFlags::basic(true, false))
            .await
            .expect("upsert rule");
        let forced_ok = evaluate_pre_receive(&store, &git_env, &request(&second, &divergent))
            .await
            .expect("evaluate");
        assert_eq!(forced_ok, HookResponse::allow());
        let deleted_again =
            evaluate_pre_receive(&store, &git_env, &request(&second, &"0".repeat(40)))
                .await
                .expect("evaluate");
        assert_eq!(
            deleted_again.message,
            "deleting protected branch 'main' is not allowed"
        );
    }
}