//! Repo mirrors in the web process: the sync worker (see `worker`), the push debounce, the
//! pull-side ref diff, and the SSRF checks run before every sync.
//!
//! A pull mirror is fetched with `forge_git::mirror::mirror_fetch`, then every branch and tag that
//! moved is passed to `post_update::after_ref_update` with the system actor, so webhooks and pull
//! request sync behave as they do after a push. A push mirror runs `mirror_push`.

pub mod worker;

use std::collections::BTreeMap;
use std::future::Future;
use std::net::IpAddr;
use std::path::Path;
use std::time::Duration;

use forge_core::mirrors::{Mirror, MirrorDirection};
use forge_core::time::{now_stamp, stamp_after};
use forge_core::{AuditAction, Repo};
use forge_git::command::GitCommand;
use forge_git::mirror::{PinnedHost, RemoteAuth, mirror_fetch, mirror_push};
use reqwest::Url;
use serde_json::json;

use crate::audit::{self, Actor, Target};
use crate::post_update::after_ref_update;
use crate::state::AppState;
use crate::webhooks::client::{WebhookPolicy, is_forbidden_ip, validate_url};

/// How long a push waits after the last ref update before it is sent.
pub const PUSH_DEBOUNCE: Duration = Duration::from_secs(10);
/// Hard limit on one sync run. The git process is killed when it passes.
pub const RUN_TIMEOUT: Duration = Duration::from_secs(10 * 60);
/// The all-zero object id git uses for a ref that does not exist on one side.
pub const ZERO_OID: &str = "0000000000000000000000000000000000000000";
/// Next run for a mirror whose interval is `0` (manual only).
const FAR_FUTURE: &str = "9999-12-31T23:59:59.999999999Z";

/// Queues every enabled push mirror of a repo for a sync `PUSH_DEBOUNCE` from now, and wakes the
/// worker. Called after a branch or tag update.
pub async fn enqueue_push_mirrors(state: &AppState, repo_id: &str) -> forge_core::Result<()> {
    let due = stamp_after(PUSH_DEBOUNCE);
    let mut queued = false;
    for mirror in state.store.list_mirrors(repo_id).await? {
        if mirror.direction == MirrorDirection::Push && mirror.enabled {
            state.store.schedule_mirror(&mirror.id, &due).await?;
            queued = true;
        }
    }
    if queued {
        state.mirror_waker.notify_one();
    }
    Ok(())
}

/// The policy the mirror worker runs under, from the `[webhooks]` config. An invalid config falls
/// back to the strictest policy: no internal hosts and no plain HTTP.
pub fn policy(state: &AppState) -> WebhookPolicy {
    WebhookPolicy::from_config(&state.config.webhooks).unwrap_or_else(|e| {
        tracing::error!(error = %e, "mirrors: invalid [webhooks] config, using the default policy");
        WebhookPolicy {
            allow_http: false,
            allowed_hosts: Vec::new(),
        }
    })
}

/// Checks a mirror URL against the webhook policy before a run: the URL shape, and every address
/// the host resolves to. Returns the address git must connect to (see [`pin_checked`]). `None`
/// when there is nothing to pin: an address literal was checked by `validate_url`, and a host in
/// `allowed_hosts` is trusted.
pub async fn check_remote(url: &str, policy: &WebhookPolicy) -> Result<Option<PinnedHost>, String> {
    let parsed = validate_url(url, policy).map_err(|e| format!("mirror URL refused: {e}"))?;
    let host = parsed.host_str().unwrap_or_default();
    let bare = host.trim_start_matches('[').trim_end_matches(']');
    if bare.parse::<IpAddr>().is_ok() || policy.host_allowed(host) {
        return Ok(None);
    }
    let port = parsed.port_or_known_default().unwrap_or(443);
    // One lookup for the whole run: the addresses checked here are the ones git is pinned to.
    let resolved: Vec<IpAddr> = tokio::net::lookup_host((host, port))
        .await
        .map_err(|e| format!("cannot resolve the mirror host: {e}"))?
        .map(|addr| addr.ip())
        .collect();
    pin_checked(host, port, &resolved, policy).map(Some)
}

/// Checks every resolved address of `host` and pins the first one. One forbidden address refuses
/// the whole host, so a mixed answer cannot slip a private address past the check.
pub fn pin_checked(
    host: &str,
    port: u16,
    resolved: &[IpAddr],
    policy: &WebhookPolicy,
) -> Result<PinnedHost, String> {
    let Some(ip) = resolved.first().copied() else {
        return Err("the mirror host did not resolve".into());
    };
    if resolved
        .iter()
        .any(|ip| is_forbidden_ip(*ip) && !policy.ip_allowed(*ip))
    {
        return Err("mirror host resolves to a forbidden address".into());
    }
    Ok(PinnedHost {
        host: host.to_string(),
        port,
        ip,
    })
}

/// The URL with userinfo and query removed: scheme, host (and port) and path. Used in audit
/// entries and anywhere a URL is shown.
pub fn redact_url(url: &str) -> String {
    match Url::parse(url) {
        Ok(parsed) => {
            let host = parsed.host_str().unwrap_or_default();
            let port = parsed.port().map(|p| format!(":{p}")).unwrap_or_default();
            format!("{}://{host}{port}{}", parsed.scheme(), parsed.path())
        }
        Err(_) => "[invalid url]".to_string(),
    }
}

/// The credentials a run presents. `None` when the mirror has neither a user nor a password.
fn auth_for(mirror: &Mirror, secret: Option<Vec<u8>>) -> Option<RemoteAuth> {
    match secret {
        Some(secret) => Some(RemoteAuth {
            username: mirror.username.clone(),
            secret,
        }),
        None if !mirror.username.is_empty() => Some(RemoteAuth {
            username: mirror.username.clone(),
            secret: Vec::new(),
        }),
        None => None,
    }
}

/// Every branch and tag of the bare repo at `repo_dir`, as full ref name to object id.
pub async fn ref_snapshot(
    state: &AppState,
    repo_dir: &Path,
) -> Result<BTreeMap<String, String>, String> {
    let out = GitCommand::new(&state.git, "for-each-ref")
        .git_dir(repo_dir)
        .arg("--format=%(objectname) %(refname)")
        .arg("refs/heads/")
        .arg("refs/tags/")
        .output()
        .await
        .map_err(|e| e.to_string())?;
    let mut refs = BTreeMap::new();
    for line in String::from_utf8_lossy(&out).lines() {
        if let Some((oid, name)) = line.split_once(' ') {
            refs.insert(name.to_string(), oid.to_string());
        }
    }
    Ok(refs)
}

/// The refs that differ between two snapshots as `(name, old, new)`. A new ref has the zero old
/// id; a removed ref has the zero new id.
pub fn changed_refs(
    before: &BTreeMap<String, String>,
    after: &BTreeMap<String, String>,
) -> Vec<(String, String, String)> {
    let mut changes = Vec::new();
    for (name, new) in after {
        match before.get(name) {
            Some(old) if old == new => {}
            Some(old) => changes.push((name.clone(), old.clone(), new.clone())),
            None => changes.push((name.clone(), ZERO_OID.to_string(), new.clone())),
        }
    }
    for (name, old) in before {
        if !after.contains_key(name) {
            changes.push((name.clone(), old.clone(), ZERO_OID.to_string()));
        }
    }
    changes
}

/// Snapshots the refs, runs `fetch`, snapshots again, and passes every changed ref to
/// `after_ref_update` as the system actor. Returns how many refs changed. A failed `after_ref_update`
/// is logged and does not stop the others.
pub async fn pull_refs<F>(state: &AppState, repo: &Repo, fetch: F) -> Result<usize, String>
where
    F: Future<Output = Result<(), String>>,
{
    let repo_dir =
        forge_git::repo::repo_path(&state.git.data_dir, &repo.id).map_err(|e| e.to_string())?;
    let before = ref_snapshot(state, &repo_dir).await?;
    fetch.await?;
    let after = ref_snapshot(state, &repo_dir).await?;
    let changes = changed_refs(&before, &after);
    for (name, old, new) in &changes {
        if let Err(e) = after_ref_update(state, repo, "", name, old, new).await {
            tracing::error!(error = %e, repo_id = %repo.id, refname = %name, "mirrors: post-update work failed");
        }
    }
    Ok(changes.len())
}

/// One sync of one mirror, without recording it. `Err` is the scrubbed error text.
async fn sync_one(state: &AppState, mirror: &Mirror, policy: &WebhookPolicy) -> Result<(), String> {
    let repo = state
        .store
        .repo_by_id(&mirror.repo_id)
        .await
        .map_err(|e| e.to_string())?
        .ok_or("the repository is gone")?;
    let secret = state
        .store
        .mirror_secret(&mirror.id)
        .await
        .map_err(|e| e.to_string())?;
    let auth = auth_for(mirror, secret);
    let pin = check_remote(&mirror.url, policy).await?;
    let repo_dir =
        forge_git::repo::repo_path(&state.git.data_dir, &repo.id).map_err(|e| e.to_string())?;
    let allow_http = policy.allow_http;
    match mirror.direction {
        MirrorDirection::Pull => {
            let fetch = async {
                mirror_fetch(
                    &state.git,
                    &repo_dir,
                    &mirror.url,
                    auth.as_ref(),
                    allow_http,
                    pin.as_ref(),
                    RUN_TIMEOUT,
                )
                .await
                .map_err(|e| e.to_string())
            };
            pull_refs(state, &repo, fetch).await.map(|_| ())
        }
        MirrorDirection::Push => mirror_push(
            &state.git,
            &repo_dir,
            &mirror.url,
            auth.as_ref(),
            allow_http,
            pin.as_ref(),
            RUN_TIMEOUT,
        )
        .await
        .map_err(|e| e.to_string()),
    }
}

/// When the next run of a mirror is due, from now.
fn next_run_at(interval_minutes: u32) -> String {
    if interval_minutes == 0 {
        return FAR_FUTURE.to_string();
    }
    stamp_after(Duration::from_secs(u64::from(interval_minutes) * 60))
}

/// Runs one sync and records the result. Audits only the transition from not-failed to failed.
/// A debounce or "sync now" that arrives during the run keeps its earlier due time.
pub async fn run_one(state: &AppState, mirror: &Mirror, policy: &WebhookPolicy) {
    let started = now_stamp();
    let outcome = sync_one(state, mirror, policy).await;
    let ok = outcome.is_ok();
    let error = outcome.err();
    let mut next = next_run_at(mirror.interval_minutes);
    if let Ok(Some(current)) = state.store.mirror(&mirror.id).await
        && current.next_sync_at > started
        && current.next_sync_at < next
    {
        next = current.next_sync_at;
    }
    if let Err(e) = state
        .store
        .record_mirror_result(&mirror.id, ok, error.as_deref(), &next)
        .await
    {
        tracing::error!(error = %e, mirror_id = %mirror.id, "mirrors: could not record the sync result");
    }
    if let Some(error) = &error {
        tracing::warn!(mirror_id = %mirror.id, error = %error, "mirrors: sync failed");
        if mirror.last_ok != Some(false) {
            let data = json!({
                "direction": mirror.direction.as_str(),
                "url": redact_url(&mirror.url),
                "error": error,
            });
            audit::record(
                &state.store,
                None,
                Actor::Anonymous,
                AuditAction::MirrorSyncFailed,
                Target {
                    kind: "mirror",
                    id: &mirror.id,
                    name: &redact_url(&mirror.url),
                },
                data,
            )
            .await;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn refs(pairs: &[(&str, &str)]) -> BTreeMap<String, String> {
        pairs
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect()
    }

    #[test]
    fn changed_refs_finds_new_moved_and_removed() {
        let before = refs(&[
            ("refs/heads/main", "a"),
            ("refs/heads/gone", "b"),
            ("refs/tags/v1", "c"),
        ]);
        let after = refs(&[
            ("refs/heads/main", "d"),
            ("refs/heads/new", "e"),
            ("refs/tags/v1", "c"),
        ]);
        let mut changes = changed_refs(&before, &after);
        changes.sort();
        assert_eq!(
            changes,
            vec![
                (
                    "refs/heads/gone".to_string(),
                    "b".to_string(),
                    ZERO_OID.to_string()
                ),
                (
                    "refs/heads/main".to_string(),
                    "a".to_string(),
                    "d".to_string()
                ),
                (
                    "refs/heads/new".to_string(),
                    ZERO_OID.to_string(),
                    "e".to_string()
                ),
            ]
        );
        assert!(changed_refs(&after, &after).is_empty());
    }

    #[test]
    fn redacted_url_drops_userinfo_and_query() {
        assert_eq!(
            redact_url("https://user:pw@example.com:8443/a/b.git?token=x"),
            "https://example.com:8443/a/b.git"
        );
    }

    #[test]
    fn auth_needs_a_user_or_a_secret() {
        let mut mirror = sample_mirror();
        assert!(auth_for(&mirror, None).is_none());
        mirror.username = "bot".into();
        let anon = auth_for(&mirror, None).expect("user only");
        assert!(anon.secret.is_empty());
        let full = auth_for(&mirror, Some(b"tok".to_vec())).expect("with secret");
        assert_eq!(full.secret, b"tok".to_vec());
    }

    #[test]
    fn manual_interval_is_far_future() {
        assert_eq!(next_run_at(0), FAR_FUTURE);
        assert!(next_run_at(480).as_str() < FAR_FUTURE);
    }

    #[test]
    fn pin_uses_the_first_checked_address() {
        let policy = WebhookPolicy {
            allow_http: false,
            allowed_hosts: Vec::new(),
        };
        let public_v4: IpAddr = "93.184.216.34".parse().expect("ip");
        let public_v6: IpAddr = "2606:2800:220:1::1".parse().expect("ip");
        let pinned = pin_checked("example.com", 443, &[public_v4, public_v6], &policy)
            .expect("public addresses pass");
        assert_eq!(pinned.ip, public_v4);
        assert_eq!(pinned.port, 443);
        assert_eq!(
            pinned.config_pair().expect("pair"),
            "http.curloptResolve=example.com:443:93.184.216.34"
        );
        let pinned_v6 = pin_checked("example.com", 80, &[public_v6], &policy).expect("v6");
        assert_eq!(
            pinned_v6.config_pair().expect("pair"),
            "http.curloptResolve=example.com:80:[2606:2800:220:1::1]"
        );
        // A private address anywhere in the answer refuses the host, even when listed second.
        let loopback: IpAddr = "127.0.0.1".parse().expect("ip");
        assert!(pin_checked("example.com", 443, &[public_v4, loopback], &policy).is_err());
        assert!(pin_checked("example.com", 443, &[loopback, public_v4], &policy).is_err());
        assert!(pin_checked("example.com", 443, &[], &policy).is_err());
    }

    #[tokio::test]
    async fn remote_checks_refuse_loopback_and_userinfo() {
        let policy = WebhookPolicy {
            allow_http: true,
            allowed_hosts: Vec::new(),
        };
        assert!(
            check_remote("https://127.0.0.1/x.git", &policy)
                .await
                .is_err()
        );
        assert!(check_remote("https://[::1]/x.git", &policy).await.is_err());
        assert!(
            check_remote("https://u:p@example.com/x.git", &policy)
                .await
                .is_err()
        );
        assert!(check_remote("file:///etc/passwd", &policy).await.is_err());
        let allowed = WebhookPolicy {
            allow_http: false,
            allowed_hosts: vec![crate::webhooks::client::AllowEntry::Net(
                "127.0.0.0/8".parse().expect("cidr"),
            )],
        };
        assert!(
            check_remote("https://127.0.0.1/x.git", &allowed)
                .await
                .is_ok()
        );
    }

    pub(super) fn sample_mirror() -> Mirror {
        Mirror {
            id: "m1".into(),
            repo_id: "r1".into(),
            direction: MirrorDirection::Pull,
            url: "https://example.com/x.git".into(),
            username: String::new(),
            has_secret: false,
            interval_minutes: 480,
            enabled: true,
            next_sync_at: now_stamp(),
            last_sync_at: None,
            last_ok: None,
            last_error: None,
            created_at: now_stamp(),
        }
    }
}