//! The advisory worker. When `[advisories] enabled = true`, it keeps an OSV dump per configured
//! ecosystem in `<data_dir>/advisories/<ecosystem>/all.zip`. A dump older than `refresh_hours` is
//! downloaded again (capped at [`MAX_DUMP_BYTES`], with timeouts, no redirects). A dump that is
//! newer than the last import is imported, and then every repo is rescanned. It checks once an
//! hour, and once at boot, so an existing 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;

/// How often the worker checks the dumps.
const CHECK_EVERY: 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 mut imported: HashMap<String, SystemTime> = HashMap::new();
    loop {
        for ecosystem in &state.config.advisories.ecosystems {
            if !KNOWN_ECOSYSTEMS.contains(&ecosystem.as_str()) {
                tracing::warn!(ecosystem = %ecosystem, "advisories: unknown ecosystem, skipped");
                continue;
            }
            if let Err(e) = refresh_one(&state, &client, ecosystem, &mut imported).await {
                tracing::error!(error = %e, ecosystem = %ecosystem, "advisories: refresh failed");
            }
        }
        tokio::time::sleep(CHECK_EVERY).await;
    }
}

/// Downloads `ecosystem`'s dump if it is stale, then imports it if it changed since the last import.
async fn refresh_one(
    state: &AppState,
    client: &reqwest::Client,
    ecosystem: &str,
    imported: &mut HashMap<String, SystemTime>,
) -> Result<()> {
    let path = dump_path(&state.config.data_dir, ecosystem);
    let max_age = Duration::from_secs(state.config.advisories.refresh_hours.max(1) * 3600);
    if is_older_than(&path, max_age) {
        download(client, ecosystem, &path).await?;
    }
    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()
}

/// True when the file is missing or was written more than `max_age` ago.
fn is_older_than(path: &Path, max_age: Duration) -> bool {
    match modified_time(path) {
        None => true,
        Some(at) => SystemTime::now()
            .duration_since(at)
            .is_ok_and(|age| age > max_age),
    }
}

/// 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(())
}