//! The mirror sync worker. It polls for due mirrors every minute, and wakes at once when
//! `AppState::mirror_waker` is notified (a "sync now", a new mirror, or a push). At most two runs
//! are in flight, and never two on the same repo.

use std::collections::HashSet;
use std::time::Duration;

use forge_core::time::now_stamp;
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use tokio::sync::mpsc;

use super::{policy, run_one};
use crate::state::AppState;
use crate::webhooks::client::WebhookPolicy;

/// The longest the worker sleeps when nothing wakes it.
const POLL: Duration = Duration::from_secs(60);
/// The shortest sleep, so a due time that has already passed does not spin the loop.
const MIN_SLEEP: Duration = Duration::from_secs(1);
/// Runs in flight at once.
const MAX_RUNNING: usize = 2;
/// Due mirrors read per pass.
const CLAIM_BATCH: u32 = 20;

/// Starts the worker task. It runs until the process exits.
pub fn spawn(state: AppState) {
    tokio::spawn(run(state));
}

async fn run(state: AppState) {
    let policy = policy(&state);
    let (done_tx, mut done_rx) = mpsc::unbounded_channel::<(String, String)>();
    let mut running_mirrors: HashSet<String> = HashSet::new();
    let mut running_repos: HashSet<String> = HashSet::new();
    loop {
        claim(
            &state,
            &policy,
            &done_tx,
            &mut running_mirrors,
            &mut running_repos,
        )
        .await;
        // Wake when the earliest enabled mirror is due, so a push mirror queued with a short
        // debounce runs on time instead of waiting for the next poll.
        let next = match state.store.next_mirror_due().await {
            Ok(next) => next,
            Err(e) => {
                tracing::error!(error = %e, "mirrors: could not read the next due time");
                None
            }
        };
        let nap = sleep_for(next.as_deref(), OffsetDateTime::now_utc());
        tokio::select! {
            _ = tokio::time::sleep(nap) => {}
            _ = state.mirror_waker.notified() => {}
            Some((mirror_id, repo_id)) = done_rx.recv() => {
                running_mirrors.remove(&mirror_id);
                running_repos.remove(&repo_id);
            }
        }
    }
}

/// How long the worker sleeps before its next pass: until `next_due` (an RFC 3339 stamp, the
/// earliest enabled mirror), clamped to `MIN_SLEEP..=POLL`. `None` or an unparsable stamp gives
/// `POLL`.
pub fn sleep_for(next_due: Option<&str>, now: OffsetDateTime) -> Duration {
    let Some(due) = next_due.and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok()) else {
        return POLL;
    };
    let until = Duration::try_from(due - now).unwrap_or(Duration::ZERO);
    until.clamp(MIN_SLEEP, POLL)
}

/// Starts runs for due mirrors until the concurrency limit, skipping mirrors and repos already
/// in flight.
async fn claim(
    state: &AppState,
    policy: &WebhookPolicy,
    done_tx: &mpsc::UnboundedSender<(String, String)>,
    running_mirrors: &mut HashSet<String>,
    running_repos: &mut HashSet<String>,
) {
    let due = match state.store.due_mirrors(&now_stamp(), CLAIM_BATCH).await {
        Ok(due) => due,
        Err(e) => {
            tracing::error!(error = %e, "mirrors: could not list due mirrors");
            return;
        }
    };
    for mirror in due {
        if running_mirrors.len() >= MAX_RUNNING {
            break;
        }
        if running_mirrors.contains(&mirror.id) || running_repos.contains(&mirror.repo_id) {
            continue;
        }
        running_mirrors.insert(mirror.id.clone());
        running_repos.insert(mirror.repo_id.clone());
        let (mirror_id, repo_id) = (mirror.id.clone(), mirror.repo_id.clone());
        let state = state.clone();
        let policy = policy.clone();
        let done = done_tx.clone();
        tokio::spawn(async move {
            run_one(&state, &mirror, &policy).await;
            let _ = done.send((mirror_id, repo_id));
        });
    }
}

#[cfg(test)]
mod tests {
    use std::path::Path;

    use forge_core::secret;
    use forge_core::webhooks::WebhookEvent;

    use super::super::pull_refs;
    use crate::test_support::test_state;

    #[test]
    fn sleep_stops_at_the_next_due_time_and_is_clamped() {
        use std::time::Duration;

        use time::OffsetDateTime;
        use time::format_description::well_known::Rfc3339;
        use time::macros::datetime;

        let now = datetime!(2026-10-15 12:00:00 UTC);
        let stamp = |t: OffsetDateTime| t.format(&Rfc3339).expect("stamp");
        // Nothing enabled: the full poll interval.
        assert_eq!(super::sleep_for(None, now), super::POLL);
        // A push queued with a 10 s debounce: sleep until it is due, not the full poll.
        let in_10s = stamp(now + time::Duration::seconds(10));
        assert_eq!(
            super::sleep_for(Some(&in_10s), now),
            Duration::from_secs(10)
        );
        // Due later than the poll interval: wake at the poll interval.
        let in_1h = stamp(now + time::Duration::hours(1));
        assert_eq!(super::sleep_for(Some(&in_1h), now), super::POLL);
        // Already due (or in flight and still due): do not spin, wait the minimum.
        let past = stamp(now - time::Duration::seconds(5));
        assert_eq!(super::sleep_for(Some(&past), now), super::MIN_SLEEP);
        // A stamp that does not parse falls back to the poll interval.
        assert_eq!(super::sleep_for(Some("soon"), now), super::POLL);
    }

    /// Plain git against a bare repo, for setting up the fake upstream.
    fn plumbing(git_dir: &Path, args: &[&str]) -> String {
        let out = std::process::Command::new("git")
            .arg(format!("--git-dir={}", git_dir.display()))
            .args(args)
            .env("GIT_AUTHOR_NAME", "t")
            .env("GIT_AUTHOR_EMAIL", "t@example.com")
            .env("GIT_COMMITTER_NAME", "t")
            .env("GIT_COMMITTER_EMAIL", "t@example.com")
            .output()
            .expect("git runs");
        assert!(out.status.success(), "git {args:?}: {out:?}");
        String::from_utf8_lossy(&out.stdout).trim().to_string()
    }

    /// Makes a new commit in `git_dir` with the given message and points `refname` at it.
    fn commit_to(git_dir: &Path, refname: &str, message: &str) {
        let tree = plumbing(git_dir, &["hash-object", "-t", "tree", "-w", "--stdin"]);
        let commit = plumbing(git_dir, &["commit-tree", &tree, "-m", message]);
        plumbing(git_dir, &["update-ref", refname, &commit]);
    }

    #[tokio::test]
    async fn pull_sync_passes_only_changed_refs_to_post_update() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = test_state(dir.path()).await;
        let owner = state
            .store
            .create_user("alice", None, "unused", false)
            .await
            .expect("user");
        let repo = state
            .store
            .create_repo(
                &owner.id,
                "mirrored",
                "",
                forge_core::Visibility::Public,
                "main",
            )
            .await
            .expect("repo");
        let bare = forge_git::repo::repo_path(&state.git.data_dir, &repo.id).expect("repo path");
        forge_git::repo::init_bare(&state.git, &bare, "main")
            .await
            .expect("init bare");
        // A webhook on push, so a changed ref shows up as a queued delivery.
        let sealed = secret::seal(&[9u8; 32], b"hook-secret")
            .expect("seal")
            .into_bytes();
        let hook = state
            .store
            .create_webhook(
                &repo.id,
                &owner.id,
                "https://example.com/hook",
                &sealed,
                &[WebhookEvent::Push],
                true,
            )
            .await
            .expect("hook");

        // First sync: the upstream has a branch and a tag.
        let upstream = bare.clone();
        let first = pull_refs(&state, &repo, async {
            commit_to(&upstream, "refs/heads/main", "one");
            commit_to(&upstream, "refs/tags/v1", "tagged");
            Ok(())
        })
        .await
        .expect("first sync");
        assert_eq!(first, 2);
        let deliveries = state
            .store
            .list_deliveries(&hook.id, 10)
            .await
            .expect("deliveries");
        assert!(
            deliveries.iter().any(|d| d.event == WebhookEvent::Push),
            "a changed branch queues a push event"
        );

        // A sync with nothing new changes nothing.
        let idle = pull_refs(&state, &repo, async { Ok(()) })
            .await
            .expect("idle sync");
        assert_eq!(idle, 0);

        // Moving the branch is detected as one change; the tag stays put.
        let moved = pull_refs(&state, &repo, async {
            commit_to(&upstream, "refs/heads/main", "two");
            Ok(())
        })
        .await
        .expect("moved sync");
        assert_eq!(moved, 1);

        // A failed fetch reports its error and changes nothing.
        let failed = pull_refs(&state, &repo, async { Err("boom".to_string()) }).await;
        assert_eq!(failed.unwrap_err(), "boom");
    }
}