//! Background upkeep for CI. Every 30 seconds it marks running jobs whose runner stopped
//! heartbeating as Error and syncs the statuses of their runs. Once a day it deletes job logs and
//! artifacts past their retention, from `[ci]` in the config.

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

use forge_core::time::stamp_ago;
use tokio::time::{Instant, MissedTickBehavior};

use super::{artifact_key, log_path, sync_run_statuses};
use crate::state::AppState;

/// How often lost jobs are reaped.
const REAP_EVERY: Duration = Duration::from_secs(30);
/// A running job with no heartbeat for this long is lost.
const LOST_AFTER: Duration = Duration::from_secs(120);
/// How often retention runs.
const PURGE_EVERY: Duration = Duration::from_secs(24 * 60 * 60);

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

async fn run(state: AppState) {
    let mut tick = tokio::time::interval(REAP_EVERY);
    tick.set_missed_tick_behavior(MissedTickBehavior::Delay);
    // Retention runs on the first pass after boot, then once a day.
    let mut next_purge = Instant::now();
    loop {
        tick.tick().await;
        reap_once(&state).await;
        if Instant::now() >= next_purge {
            purge_once(&state).await;
            next_purge = Instant::now() + PURGE_EVERY;
        }
    }
}

/// Marks lost jobs as Error and syncs the statuses of every run they belong to. Returns how many
/// jobs were reaped.
pub async fn reap_once(state: &AppState) -> usize {
    reap_before(state, &stamp_ago(LOST_AFTER)).await
}

/// [`reap_once`] with an explicit cutoff: running jobs whose heartbeat is older than `older_than`
/// are lost.
pub async fn reap_before(state: &AppState, older_than: &str) -> usize {
    let jobs = match state.store.reap_lost_jobs(older_than).await {
        Ok(jobs) => jobs,
        Err(e) => {
            tracing::error!(error = %e, "ci: reaping lost jobs failed");
            return 0;
        }
    };
    let mut seen = HashSet::new();
    let mut wake = false;
    for job in &jobs {
        if !seen.insert(job.run_id.clone()) {
            continue;
        }
        match state.store.run_by_id(&job.run_id).await {
            Ok(Some(run)) => match sync_run_statuses(state, &run).await {
                Ok(queued) => wake |= queued,
                Err(e) => {
                    tracing::error!(error = %e, run_id = %run.id, "ci: syncing statuses failed")
                }
            },
            Ok(None) => {}
            Err(e) => tracing::error!(error = %e, run_id = %job.run_id, "ci: loading run failed"),
        }
    }
    if wake {
        state.ci_waker.notify_one();
    }
    if !jobs.is_empty() {
        tracing::warn!(count = jobs.len(), "ci: reaped lost jobs");
    }
    jobs.len()
}

/// Deletes the logs and artifacts past retention. Rows go first (in the store); files follow.
/// Missing files are fine. Returns `(logs deleted, artifacts deleted)` as attempted.
pub async fn purge_once(state: &AppState) -> (usize, usize) {
    let ci = &state.config.ci;
    let day = |days: u64| Duration::from_secs(days.saturating_mul(24 * 60 * 60));
    let logs_before = stamp_ago(day(ci.log_retention_days));
    let artifacts_before = stamp_ago(day(ci.artifact_retention_days));
    let (job_ids, artifact_ids) = match state.store.purge_old(&logs_before, &artifacts_before).await
    {
        Ok(ids) => ids,
        Err(e) => {
            tracing::error!(error = %e, "ci: retention purge failed");
            return (0, 0);
        }
    };
    let logs: Vec<PathBuf> = job_ids
        .iter()
        .map(|id| log_path(&state.config.data_dir, id))
        .collect();
    let log_count = logs.len();
    let removed = tokio::task::spawn_blocking(move || {
        for path in &logs {
            if let Err(e) = std::fs::remove_file(path)
                && e.kind() != std::io::ErrorKind::NotFound
            {
                tracing::error!(error = %e, path = %path.display(), "ci: deleting a log failed");
            }
        }
    })
    .await;
    if let Err(e) = removed {
        tracing::error!(error = %e, "ci: log deletion task failed");
    }
    let artifact_count = artifact_ids.len();
    for id in &artifact_ids {
        if let Err(e) = state.blobs.delete(&artifact_key(id)).await {
            tracing::error!(error = %e, artifact_id = %id, "ci: deleting an artifact failed");
        }
    }
    (log_count, artifact_count)
}