//! Scheduled backups and fsck, the run slot they share with manual and CLI triggers, and the job
//! bodies. Every run is recorded in `backup_runs`. Only one run is active at a time: a second
//! trigger gets "already running" while the first is in progress.

use std::path::PathBuf;
use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, SystemTime};

use forge_core::store::backup_runs::{BackupRun, RunKind, RunTrigger};
use forge_core::{Error, Result};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;

use super::{
    BackupOptions, BackupSource, BackupSummary, FsckSummary, create_backup, healthcheck, retention,
    run_fsck, verify_backup,
};
use crate::state::AppState;

/// After a failed run, the next try waits at least this long.
const RETRY_AFTER_SECS: i64 = 60 * 60;
/// Pause after each pass. A due job that did not clear its due time (it could not take the run
/// slot while a manual run held it, say) waits this long instead of spinning the loop.
const RECHECK: Duration = Duration::from_secs(30);

/// The one run slot of this process.
#[derive(Default)]
pub struct Jobs {
    running: Mutex<Option<RunKind>>,
}

/// Holds the run slot. Dropping it frees the slot.
pub struct JobGuard {
    jobs: Arc<Jobs>,
    kind: RunKind,
}

impl Jobs {
    /// Takes the run slot for `kind`. Fails with "already running" while another run holds it.
    pub fn claim(self: &Arc<Self>, kind: RunKind) -> Result<JobGuard> {
        let mut slot = self.running.lock().unwrap_or_else(PoisonError::into_inner);
        if let Some(current) = *slot {
            return Err(Error::Conflict(format!(
                "already running: a {} run is in progress",
                current.as_str()
            )));
        }
        *slot = Some(kind);
        Ok(JobGuard {
            jobs: Arc::clone(self),
            kind,
        })
    }

    /// The kind of run in progress, if any.
    pub fn running(&self) -> Option<RunKind> {
        *self.running.lock().unwrap_or_else(PoisonError::into_inner)
    }
}

impl Drop for JobGuard {
    fn drop(&mut self) {
        let mut slot = self
            .jobs
            .running
            .lock()
            .unwrap_or_else(PoisonError::into_inner);
        if *slot == Some(self.kind) {
            *slot = None;
        }
    }
}

fn started_unix(run: &BackupRun) -> Option<i64> {
    OffsetDateTime::parse(&run.started_at, &Rfc3339)
        .ok()
        .map(|t| t.unix_timestamp())
}

/// Seconds until a job of this kind is due. Zero means run now.
///
/// The base is the start of the last successful run plus `every_secs`. With no success on record
/// the job is due at once. When the latest attempt failed, the next try also waits
/// `retry_after_secs` from that attempt, so a broken setup does not retry every tick.
pub fn secs_until_due(
    now: OffsetDateTime,
    last_ok: Option<&BackupRun>,
    last_attempt: Option<&BackupRun>,
    every_secs: i64,
    retry_after_secs: i64,
) -> i64 {
    let mut due = match last_ok.and_then(started_unix) {
        Some(t) => t.saturating_add(every_secs),
        None => now.unix_timestamp(),
    };
    if let Some(attempt) = last_attempt.filter(|run| run.ok != Some(true))
        && let Some(t) = started_unix(attempt)
    {
        due = due.max(t.saturating_add(retry_after_secs));
    }
    (due - now.unix_timestamp()).max(0)
}

/// Starts the scheduler loop. The caller checks that `[backup] dir` is set.
///
/// Each pass sleeps until the earliest of the backup and fsck is due, runs what is due, and then
/// works out the next instant from the run history again.
pub fn spawn(state: AppState) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            match next_due_in(&state).await {
                Ok(wait) => {
                    let until = tokio::time::Instant::now()
                        + Duration::from_secs(u64::try_from(wait).unwrap_or(0));
                    tokio::time::sleep_until(until).await;
                }
                Err(e) => {
                    tracing::error!(error = %e, "backup scheduler could not read its history");
                    tokio::time::sleep(RECHECK).await;
                    continue;
                }
            }
            run_due(&state).await;
            tokio::time::sleep(RECHECK).await;
        }
    })
}

/// How often a job of this kind is due, in seconds.
fn interval_secs(state: &AppState, kind: RunKind) -> i64 {
    let cfg = &state.config.backup;
    match kind {
        RunKind::Backup => {
            i64::try_from(cfg.interval_hours.saturating_mul(3600)).unwrap_or(i64::MAX)
        }
        RunKind::Fsck => {
            i64::try_from(cfg.fsck_interval_days.saturating_mul(86_400)).unwrap_or(i64::MAX)
        }
    }
}

/// Seconds until a job of this kind is due, from its run history. Zero means run now.
async fn secs_until_run(state: &AppState, kind: RunKind) -> Result<i64> {
    let (ok, attempt) = due_inputs(state, kind).await?;
    Ok(secs_until_due(
        OffsetDateTime::now_utc(),
        ok.as_ref(),
        attempt.as_ref(),
        interval_secs(state, kind),
        RETRY_AFTER_SECS,
    ))
}

/// Seconds until the earlier of the backup and the fsck is due. Zero means one is due now.
async fn next_due_in(state: &AppState) -> Result<i64> {
    let backup = secs_until_run(state, RunKind::Backup).await?;
    let fsck = secs_until_run(state, RunKind::Fsck).await?;
    Ok(backup.min(fsck))
}

/// Runs each job that is due now. A job that cannot take the run slot, because a manual run holds
/// it, is left for the next pass.
async fn run_due(state: &AppState) {
    for kind in [RunKind::Backup, RunKind::Fsck] {
        let due = match secs_until_run(state, kind).await {
            Ok(wait) => wait == 0,
            Err(e) => {
                tracing::error!(error = %e, kind = kind.as_str(), "scheduler could not read its history");
                false
            }
        };
        if !due {
            continue;
        }
        let Ok(guard) = state.jobs.claim(kind) else {
            continue;
        };
        let outcome = match kind {
            RunKind::Backup => scheduled_backup(state).await.map(|_| ()),
            RunKind::Fsck => fsck_job(state, RunTrigger::Schedule, false)
                .await
                .map(|_| ()),
        };
        if let Err(e) = outcome {
            tracing::error!(error = %e, kind = kind.as_str(), "scheduled run failed");
        }
        drop(guard);
    }
}

async fn due_inputs(
    state: &AppState,
    kind: RunKind,
) -> Result<(Option<BackupRun>, Option<BackupRun>)> {
    let ok = state.store.last_ok(kind).await?;
    let attempt = state.store.last_run(kind).await?;
    Ok((ok, attempt))
}

/// The configured backup directory as an absolute path.
pub fn configured_dir(state: &AppState) -> Result<PathBuf> {
    std::path::absolute(&state.config.backup.dir).map_err(Error::Io)
}

async fn scheduled_backup(state: &AppState) -> Result<BackupSummary> {
    let opts = BackupOptions {
        out: configured_dir(state)?,
        include_key: false,
        scope: super::Scope::All,
    };
    backup_job(state, RunTrigger::Schedule, opts, true).await
}

/// Runs one backup set and records it. The caller holds the run slot.
///
/// The set is verified before the run counts as ok. `prune` applies retention to `opts.out` after
/// a good run. Healthcheck pings are sent for scheduled runs only.
pub async fn backup_job(
    state: &AppState,
    trigger: RunTrigger,
    opts: BackupOptions,
    prune: bool,
) -> Result<BackupSummary> {
    let run_id = state.store.start_run(RunKind::Backup, trigger).await?;
    let healthcheck_url = Some(state.config.backup.healthcheck_url.as_str())
        .filter(|url| trigger == RunTrigger::Schedule && !url.is_empty());
    if let Some(url) = healthcheck_url {
        healthcheck::start(url).await;
    }

    let outcome = backup_set(state, &opts).await;
    match &outcome {
        Ok(summary) => {
            let path = summary.path.to_string_lossy().into_owned();
            record_finish(state, &run_id, true, Some(&path), Some(summary.bytes), None).await;
        }
        Err(e) => {
            record_finish(state, &run_id, false, None, None, Some(&e.to_string())).await;
        }
    }
    if outcome.is_ok() && prune {
        match retention::prune(&opts.out, state.config.backup.keep, SystemTime::now()) {
            Ok(report) => tracing::info!(
                sets = report.sets_removed.len(),
                partials = report.partials_removed.len(),
                "backup retention applied"
            ),
            Err(e) => tracing::warn!(error = %e, "backup retention failed"),
        }
    }
    if let Some(url) = healthcheck_url {
        match &outcome {
            Ok(_) => healthcheck::success(url).await,
            Err(e) => healthcheck::failure(url, &e.to_string()).await,
        }
    }
    outcome
}

async fn backup_set(state: &AppState, opts: &BackupOptions) -> Result<BackupSummary> {
    let source = BackupSource {
        store: state.store.clone(),
        git: (*state.git).clone(),
        base_url: state.config.base_url.clone(),
    };
    let summary = create_backup(&source, opts).await?;
    let report = verify_backup(&summary.path, &source.git).await?;
    if !report.is_ok() {
        return Err(Error::Invalid(format!(
            "the new backup set failed verification: {}",
            report.problems.join("; ")
        )));
    }
    Ok(summary)
}

/// Runs fsck over every repo and the database, and records it. The caller holds the run slot.
pub async fn fsck_job(state: &AppState, trigger: RunTrigger, full: bool) -> Result<FsckSummary> {
    let run_id = state.store.start_run(RunKind::Fsck, trigger).await?;
    let outcome = run_fsck(&state.store, &state.git, full).await;
    match &outcome {
        Ok(summary) if summary.is_ok() => {
            record_finish(state, &run_id, true, None, None, None).await;
        }
        Ok(summary) => {
            let text = summary.problems.join("\n");
            record_finish(state, &run_id, false, None, None, Some(&text)).await;
        }
        Err(e) => {
            record_finish(state, &run_id, false, None, None, Some(&e.to_string())).await;
        }
    }
    outcome
}

/// Writes the outcome of a run. A failure to record is logged, because the run itself has already
/// happened.
async fn record_finish(
    state: &AppState,
    run_id: &str,
    ok: bool,
    path: Option<&str>,
    bytes: Option<u64>,
    error: Option<&str>,
) {
    if let Err(e) = state.store.finish_run(run_id, ok, path, bytes, error).await {
        tracing::error!(error = %e, run_id, "could not record the outcome of a backup run");
    }
}

/// Starts a manual run in the background and returns at once. The caller claimed the run slot
/// (so it could refuse with "already running" and record the audit entry first). Fails with a
/// config error, freeing the slot, when no backup directory is set.
pub fn start_manual(state: &AppState, guard: JobGuard) -> Result<()> {
    if !state.config.backup.enabled() {
        return Err(Error::Config(
            "backups are not configured: set [backup] dir in the config file".into(),
        ));
    }
    let kind = guard.kind;
    let state = state.clone();
    tokio::spawn(async move {
        let _guard = guard;
        let outcome: Result<()> = match kind {
            RunKind::Backup => match configured_dir(&state) {
                Ok(out) => backup_job(
                    &state,
                    RunTrigger::Manual,
                    BackupOptions {
                        out,
                        include_key: false,
                        scope: super::Scope::All,
                    },
                    true,
                )
                .await
                .map(|_| ()),
                Err(e) => Err(e),
            },
            RunKind::Fsck => fsck_job(&state, RunTrigger::Manual, false)
                .await
                .map(|_| ()),
        };
        if let Err(e) = outcome {
            tracing::error!(error = %e, kind = kind.as_str(), "manual run failed");
        }
    });
    Ok(())
}

/// Runs a backup now and waits for it, holding the run slot in `guard`. For the CLI through the
/// internal socket.
pub async fn run_backup_now(
    state: &AppState,
    guard: JobGuard,
    trigger: RunTrigger,
    opts: BackupOptions,
    prune: bool,
) -> Result<BackupSummary> {
    let _guard = guard;
    backup_job(state, trigger, opts, prune).await
}

/// Runs fsck now and waits for it, holding the run slot in `guard`. For the CLI through the
/// internal socket.
pub async fn run_fsck_now(
    state: &AppState,
    guard: JobGuard,
    trigger: RunTrigger,
    full: bool,
) -> Result<FsckSummary> {
    let _guard = guard;
    fsck_job(state, trigger, full).await
}

#[cfg(test)]
mod tests {
    use super::*;
    use forge_core::model::Visibility;
    use forge_git::repo::{init_bare, repo_path};
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpListener;

    fn at(stamp: &str) -> OffsetDateTime {
        OffsetDateTime::parse(stamp, &Rfc3339).expect("stamp")
    }

    fn run(kind: RunKind, started: &str, ok: Option<bool>) -> BackupRun {
        BackupRun {
            id: format!("id-{started}"),
            kind,
            trigger: RunTrigger::Schedule,
            started_at: started.to_string(),
            finished_at: None,
            ok,
            path: None,
            bytes: None,
            error: None,
        }
    }

    const DAY: i64 = 86_400;
    const HOUR: i64 = 3600;

    #[test]
    fn never_run_is_due_now() {
        let now = at("2026-10-15T12:00:00Z");
        assert_eq!(secs_until_due(now, None, None, DAY, HOUR), 0);
    }

    #[test]
    fn next_run_follows_the_last_success() {
        let now = at("2026-10-15T12:00:00Z");
        let last = run(RunKind::Backup, "2026-10-15T06:00:00Z", Some(true));
        // Last success six hours ago with a 24 hour interval: due in eighteen hours.
        assert_eq!(
            secs_until_due(now, Some(&last), Some(&last), DAY, HOUR),
            18 * HOUR
        );
        let old = run(RunKind::Backup, "2026-10-13T00:00:00Z", Some(true));
        assert_eq!(secs_until_due(now, Some(&old), Some(&old), DAY, HOUR), 0);
    }

    #[test]
    fn a_failed_attempt_waits_before_retrying() {
        let now = at("2026-10-15T12:00:00Z");
        let old_ok = run(RunKind::Backup, "2026-10-13T00:00:00Z", Some(true));
        // The latest attempt failed 10 minutes ago: the retry waits the rest of the hour.
        let failed = run(RunKind::Backup, "2026-10-15T11:50:00Z", Some(false));
        assert_eq!(
            secs_until_due(now, Some(&old_ok), Some(&failed), DAY, HOUR),
            50 * 60
        );
        // A run that never finished counts as a failed attempt.
        let unfinished = run(RunKind::Backup, "2026-10-15T11:50:00Z", None);
        assert_eq!(
            secs_until_due(now, None, Some(&unfinished), DAY, HOUR),
            50 * 60
        );
        // Failed long ago: due now.
        let stale = run(RunKind::Backup, "2026-10-15T09:00:00Z", Some(false));
        assert_eq!(secs_until_due(now, None, Some(&stale), DAY, HOUR), 0);
    }

    #[test]
    fn the_slot_is_exclusive_and_freed_on_drop() {
        let jobs = Arc::new(Jobs::default());
        let guard = jobs.claim(RunKind::Backup).expect("first claim");
        assert_eq!(jobs.running(), Some(RunKind::Backup));
        match jobs.claim(RunKind::Fsck) {
            Err(Error::Conflict(message)) => assert!(message.contains("already running")),
            other => panic!("expected a conflict, got {:?}", other.map(|_| ())),
        }
        drop(guard);
        assert_eq!(jobs.running(), None);
        assert!(jobs.claim(RunKind::Fsck).is_ok());
    }

    /// A tiny server that answers `count` requests and records the request lines.
    async fn ping_server(count: usize) -> (String, Arc<Mutex<Vec<String>>>) {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("addr");
        let seen = Arc::new(Mutex::new(Vec::new()));
        let log = Arc::clone(&seen);
        tokio::spawn(async move {
            for _ in 0..count {
                let Ok((mut sock, _)) = listener.accept().await else {
                    return;
                };
                let mut buf = vec![0u8; 8192];
                let n = sock.read(&mut buf).await.unwrap_or(0);
                let text = String::from_utf8_lossy(&buf[..n]).into_owned();
                log.lock()
                    .unwrap_or_else(PoisonError::into_inner)
                    .push(text.lines().next().unwrap_or_default().to_owned());
                let _ = sock
                    .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
                    .await;
            }
        });
        (format!("http://{addr}/hc/token-123"), seen)
    }

    #[tokio::test]
    async fn manual_start_without_a_dir_is_a_config_error_and_frees_the_slot() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = crate::test_support::test_state(dir.path()).await;
        let guard = state.jobs.claim(RunKind::Backup).expect("claim");
        assert!(matches!(start_manual(&state, guard), Err(Error::Config(_))));
        assert_eq!(state.jobs.running(), None);
    }

    #[tokio::test]
    async fn a_fresh_install_wakes_at_once() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = crate::test_support::test_state(dir.path()).await;
        assert_eq!(next_due_in(&state).await.expect("next due"), 0);
    }

    #[tokio::test]
    async fn scheduled_backup_is_recorded_verified_and_pinged() {
        let dir = tempfile::tempdir().expect("tempdir");
        let data_dir = dir.path().join("data");
        std::fs::create_dir_all(&data_dir).expect("data dir");
        let out = dir.path().join("out");
        let (url, pings) = ping_server(2).await;

        let mut state = crate::test_support::test_state(&data_dir).await;
        let mut config = crate::test_support::test_config(&data_dir);
        config.backup.dir = out.clone();
        config.backup.keep = 1;
        config.backup.healthcheck_url = url;
        state.config = Arc::new(config);

        let owner = state
            .store
            .create_user("alice", None, "hash", false)
            .await
            .expect("user");
        let repo = state
            .store
            .create_repo(&owner.id, "demo", "", Visibility::Private, "main")
            .await
            .expect("repo");
        let path = repo_path(&data_dir, &repo.id).expect("path");
        init_bare(&state.git, &path, "main").await.expect("init");

        let first = scheduled_backup(&state).await.expect("first backup");
        assert!(first.path.starts_with(&out));
        // A second run: retention keeps only the newest complete set. Set names have one-second
        // stamps, so the runs must not share a second.
        tokio::time::sleep(Duration::from_millis(1100)).await;
        let second = scheduled_backup(&state).await.expect("second backup");
        assert_ne!(first.path, second.path);
        let sets: Vec<_> = std::fs::read_dir(&out)
            .expect("read out")
            .filter_map(|e| e.ok().map(|e| e.file_name()))
            .collect();
        assert_eq!(sets.len(), 1, "{sets:?}");

        let runs = state.store.recent_runs(10).await.expect("runs");
        assert_eq!(runs.len(), 2);
        assert!(runs.iter().all(|r| r.ok == Some(true)));
        assert!(runs.iter().all(|r| r.trigger == RunTrigger::Schedule));
        assert!(runs[0].path.is_some());

        // The first scheduled run pings start and success; the second run's pings are not needed.
        for _ in 0..100 {
            if pings.lock().unwrap_or_else(PoisonError::into_inner).len() >= 2 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        let seen = pings.lock().unwrap_or_else(PoisonError::into_inner).clone();
        assert_eq!(seen[0], "GET /hc/token-123/start HTTP/1.1");
        assert_eq!(seen[1], "GET /hc/token-123 HTTP/1.1");
    }
}