//! The advisory worker. When `[advisories] enabled = true`, it keeps an OSV dump per configured
//! ecosystem in `<data_dir>/advisories/<ecosystem>/all.zip`. Each dump is downloaded again
//! `refresh_hours` after its last refresh (the file's modification time), sleeping until then
//! (capped at [`MAX_DUMP_BYTES`], with timeouts, no redirects). A failed download is retried with
//! a backoff that doubles from [`RETRY_FIRST`] up to [`RETRY_MAX`]. A dump that is newer than the
//! last import is imported, and then every repo is rescanned. At boot, a fresh dump is imported
//! without a download.

use std::collections::HashMap;
use std::fs::{self, File};
use std::io::{BufReader, Write};
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime};

use forge_core::advisories::{KNOWN_ECOSYSTEMS, read_dump};
use forge_core::{Error, Result};

use super::scan::scan_all;
use crate::state::AppState;

/// First wait after a failed download. It doubles with each further failure.
const RETRY_FIRST: Duration = Duration::from_secs(5 * 60);
/// Longest wait after a failed download.
const RETRY_MAX: Duration = Duration::from_secs(60 * 60);
/// Largest dump accepted, in bytes.
const MAX_DUMP_BYTES: u64 = 500 * 1024 * 1024;
const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
/// Time allowed for one whole download.
const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(30 * 60);
const USER_AGENT: &str = concat!("forge-advisories/", env!("CARGO_PKG_VERSION"));

/// Starts the worker task, if advisories are enabled. It runs until the process exits.
pub fn spawn(state: AppState) {
    if !state.config.advisories.enabled {
        return;
    }
    tokio::spawn(run(state));
}

/// The OSV dump URL for an ecosystem.
pub fn dump_url(ecosystem: &str) -> String {
    format!("https://osv-vulnerabilities.storage.googleapis.com/{ecosystem}/all.zip")
}

/// Where the dump of `ecosystem` is kept.
pub fn dump_path(data_dir: &Path, ecosystem: &str) -> PathBuf {
    data_dir.join("advisories").join(ecosystem).join("all.zip")
}

async fn run(state: AppState) {
    let client = match reqwest::Client::builder()
        .redirect(reqwest::redirect::Policy::none())
        .connect_timeout(CONNECT_TIMEOUT)
        .timeout(DOWNLOAD_TIMEOUT)
        .user_agent(USER_AGENT)
        .build()
    {
        Ok(client) => client,
        Err(e) => {
            tracing::error!(error = %e, "advisories: could not build the HTTP client");
            return;
        }
    };
    let refresh_after = Duration::from_secs(state.config.advisories.refresh_hours.max(1) * 3600);
    let mut imported: HashMap<String, SystemTime> = HashMap::new();
    // When each ecosystem's dump is next downloaded, and how many downloads in a row failed.
    let mut due: HashMap<String, SystemTime> = HashMap::new();
    let mut failures: HashMap<String, u32> = HashMap::new();
    loop {
        let mut wake_at: Option<SystemTime> = None;
        for ecosystem in &state.config.advisories.ecosystems {
            if !KNOWN_ECOSYSTEMS.contains(&ecosystem.as_str()) {
                tracing::warn!(ecosystem = %ecosystem, "advisories: unknown ecosystem, skipped");
                continue;
            }
            let path = dump_path(&state.config.data_dir, ecosystem);
            let mut refresh_at = *due
                .entry(ecosystem.clone())
                .or_insert_with(|| next_refresh_at(&path, refresh_after));
            let now = SystemTime::now();
            if refresh_at <= now {
                refresh_at = match download(&client, ecosystem, &path).await {
                    Ok(()) => {
                        failures.remove(ecosystem);
                        next_refresh_at(&path, refresh_after)
                    }
                    Err(e) => {
                        tracing::error!(error = %e, ecosystem = %ecosystem, "advisories: refresh failed");
                        let count = failures.entry(ecosystem.clone()).or_insert(0);
                        *count += 1;
                        now + retry_delay(*count)
                    }
                };
                due.insert(ecosystem.clone(), refresh_at);
            }
            if let Err(e) = import_if_changed(&state, ecosystem, &path, &mut imported).await {
                tracing::error!(error = %e, ecosystem = %ecosystem, "advisories: import failed");
            }
            wake_at = Some(wake_at.map_or(refresh_at, |at| at.min(refresh_at)));
        }
        let Some(wake_at) = wake_at else {
            tracing::info!("advisories: no known ecosystems configured, worker stopped");
            return;
        };
        let wait = wake_at
            .duration_since(SystemTime::now())
            .unwrap_or(Duration::ZERO);
        tokio::time::sleep(wait).await;
    }
}

/// When the dump at `path` is due for its next download: `refresh_after` past its last refresh.
/// A missing dump is due now.
fn next_refresh_at(path: &Path, refresh_after: Duration) -> SystemTime {
    match modified_time(path) {
        None => SystemTime::now(),
        Some(last) => last + refresh_after,
    }
}

/// The wait after the `failures`-th failed download in a row: 5 minutes, doubling, capped at an hour.
fn retry_delay(failures: u32) -> Duration {
    let shift = failures.saturating_sub(1).min(16);
    (RETRY_FIRST * 2u32.pow(shift)).min(RETRY_MAX)
}

/// Imports the dump of `ecosystem` if it changed since the last import, then rescans every repo.
async fn import_if_changed(
    state: &AppState,
    ecosystem: &str,
    path: &Path,
    imported: &mut HashMap<String, SystemTime>,
) -> Result<()> {
    let Some(modified) = modified_time(path) else {
        return Ok(());
    };
    if imported.get(ecosystem) == Some(&modified) {
        return Ok(());
    }
    import(state, ecosystem, path).await?;
    imported.insert(ecosystem.to_owned(), modified);
    scan_all(state).await;
    Ok(())
}

fn modified_time(path: &Path) -> Option<SystemTime> {
    fs::metadata(path).and_then(|m| m.modified()).ok()
}

/// Streams the dump to `<path>.part` and renames it into place once complete. A response that is
/// too large, or that fails midway, leaves the previous dump in place.
async fn download(client: &reqwest::Client, ecosystem: &str, path: &Path) -> Result<()> {
    let url = dump_url(ecosystem);
    let dir = path
        .parent()
        .ok_or_else(|| Error::Config("advisory dump path has no directory".into()))?;
    fs::create_dir_all(dir)?;
    let part = path.with_extension("zip.part");
    let outcome = fetch_into(client, &url, &part).await;
    match outcome {
        Ok(bytes) => {
            fs::rename(&part, path)?;
            tracing::info!(ecosystem, bytes, "advisories: dump downloaded");
            Ok(())
        }
        Err(e) => {
            let _ = fs::remove_file(&part);
            Err(e)
        }
    }
}

async fn fetch_into(client: &reqwest::Client, url: &str, part: &Path) -> Result<u64> {
    let mut response = client
        .get(url)
        .send()
        .await
        .and_then(reqwest::Response::error_for_status)
        .map_err(|e| Error::Invalid(format!("advisory download failed: {e}")))?;
    if response
        .content_length()
        .is_some_and(|len| len > MAX_DUMP_BYTES)
    {
        return Err(Error::Invalid(
            "advisory dump is larger than 500 MiB".into(),
        ));
    }
    let mut file = File::create(part)?;
    let mut total: u64 = 0;
    while let Some(chunk) = response
        .chunk()
        .await
        .map_err(|e| Error::Invalid(format!("advisory download interrupted: {e}")))?
    {
        total += chunk.len() as u64;
        if total > MAX_DUMP_BYTES {
            return Err(Error::Invalid(
                "advisory dump is larger than 500 MiB".into(),
            ));
        }
        file.write_all(&chunk)?;
    }
    file.sync_all()?;
    Ok(total)
}

/// Parses the dump on a blocking thread and replaces the stored advisories of `ecosystem`.
async fn import(state: &AppState, ecosystem: &str, path: &Path) -> Result<()> {
    let owned_path = path.to_owned();
    let eco = ecosystem.to_owned();
    let (advisories, skipped) = tokio::task::spawn_blocking(move || -> Result<_> {
        let file = File::open(&owned_path)?;
        read_dump(BufReader::new(file), &eco)
    })
    .await
    .map_err(|e| Error::Invalid(format!("advisory import task failed: {e}")))??;
    state
        .store
        .replace_advisories(ecosystem, &advisories)
        .await?;
    tracing::info!(
        ecosystem,
        imported = advisories.len(),
        skipped,
        "advisories: dump imported"
    );
    Ok(())
}