//! The history scan (hub spec §15.4, oxygenhub#277): push protection's
//! rules run over what a repository already holds, so a secret committed
//! before push protection existed, or while it was off, is found too.
//!
//! What is scanned is every change reachable from the repository's
//! bookmarks: the lines it inserts and the text blobs it puts in place of
//! a file's content, exactly what push protection reads in a push. A
//! repository's progress is a row in `secret_scans`: the heads whose
//! history is scanned (`done_heads`), and while a scan is under way the
//! heads it works toward (`target_heads`) and how far it got (`position`
//! in its list of changes: those reachable from the target and not from
//! the done heads, in clock order then by hash, so the same list every
//! pass). A pass scans within a [`Budget`] and saves where it stopped; the
//! next one resumes there. Once the list is done the target becomes the
//! done heads, so after a push only the new changes are scanned. A change
//! of rules (`version` in `push_protection.toml`) or a maintainer's "Scan
//! history now" scans the whole history again; findings are kept one per
//! value and file, so a rescan adds nothing already known and keeps what
//! was resolved.
//!
//! The scan is the worker role's `secrets.history_scan` task (docs/roles.md),
//! one repository at a time, one pass each in turn, on one blocking thread
//! that rests as long as it works: at most half a core, so it never starves
//! requests. It reads repositories with the shared lock only, as pages do.
//! Repositories with push protection off are not scanned.

use crate::bus::{Event, Topic};
use crate::config::Config;
use crate::db::{self, push_protection as pdb, repos::RepoRow};
use crate::push_protection::{self as pp, Finding};
use crate::repos::{RepoError, RepoName};
use crate::Shared;
use oxy_core::hash::{ChangeHash, ObjectType};
use oxy_core::objects::{Blob, Edit};
use oxy_core::protocol::Bundle;
use oxy_core::repo::Repo;
use std::collections::{BTreeSet, HashMap, HashSet};
use std::time::{Duration, Instant};

/// How much one pass over one repository may scan: `bytes` of text (a
/// change counts at least one, however little it adds) and `time`
/// wall-clock (resting included), each file or inserted text at most
/// `file` bytes. At least one change is scanned per pass, so a pass always
/// gets somewhere.
#[derive(Clone, Copy, Debug)]
pub struct Budget {
    pub bytes: u64,
    pub time: Duration,
    pub file: u64,
}

impl Default for Budget {
    fn default() -> Self {
        Budget { bytes: 64 << 20, time: Duration::from_secs(5), file: pp::Budget::default().file }
    }
}

impl Budget {
    /// The hub's budget from `[limits]`; `None` when the scan is off.
    pub fn of(config: &Config) -> Option<Budget> {
        let l = &config.limits;
        (l.history_scan_limit.0 > 0).then(|| Budget {
            bytes: l.history_scan_limit.0,
            time: Duration::from_millis(l.history_scan_time_ms),
            file: l.scan_file_limit.0,
        })
    }
}

/// What one pass did.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Pass {
    /// Changes and bytes scanned in this pass.
    pub changes: usize,
    pub bytes: u64,
    /// Findings new to this repository.
    pub found: usize,
    /// Changes still to scan toward the target.
    pub left: usize,
    /// Nothing to scan: the history up to the bookmarks is scanned.
    pub idle: bool,
}

/// How long the worker waits after starting before its first sweep, so a
/// starting hub serves first.
const START_DELAY: Duration = Duration::from_secs(10);

/// How long it rests between repositories.
const PAUSE: Duration = Duration::from_millis(200);

/// Every repository is looked at again this often, for the pushes the bus
/// did not tell of (it delivers at most once) and bookmarks moved without
/// a push (landings).
const SWEEP_EVERY: Duration = Duration::from_secs(15 * 60);

/// The bus topic a "Scan history now" is announced on, its kind the
/// repository's id.
pub const REQUEST_TOPIC: &str = "secrets:scan";

/// The worker role's task: every repository first, then those a push, a
/// request or a sweep says have something new, each a pass at a time.
pub async fn worker(app: Shared) {
    let Some(budget) = Budget::of(&app.config) else {
        // Off in `[limits]`: the task holds its lock and does nothing.
        return std::future::pending().await;
    };
    let mut events = app.bus.subscribe(Topic::Live);
    tokio::time::sleep(START_DELAY).await;
    let mut sweep = true;
    let mut dirty: BTreeSet<i64> = BTreeSet::new();
    loop {
        let mut order = pdb::requested(&app.db).await.unwrap_or_default();
        if sweep {
            order.extend(db::repos::list_with_drops(&app.db).await.unwrap_or_default().into_iter().map(|r| r.id));
            sweep = false;
        }
        order.extend(std::mem::take(&mut dirty));
        let mut seen = HashSet::new();
        order.retain(|id| seen.insert(*id));
        for id in order {
            sweep |= hear(&mut events, &mut dirty);
            let Ok(Some(row)) = db::repos::get_by_id(&app.db, id).await else { continue };
            if !row.push_protection {
                continue;
            }
            match pass(&app, &row, budget).await {
                Ok(p) => {
                    if p.left > 0 {
                        dirty.insert(id);
                    }
                    if p.found > 0 {
                        tracing::warn!(
                            "history scan: {}/{}: {} possible secret(s) found in history; see Settings › Push protection",
                            row.owner,
                            row.name,
                            p.found
                        );
                    }
                }
                Err(e) => tracing::warn!("history scan: {}/{}: {e:#}", row.owner, row.name),
            }
            tokio::time::sleep(PAUSE).await;
        }
        sweep |= hear(&mut events, &mut dirty);
        if sweep || !dirty.is_empty() {
            continue;
        }
        // Nothing left: wait for a push, a request or the next sweep.
        let wake = tokio::time::sleep(SWEEP_EVERY);
        tokio::pin!(wake);
        loop {
            tokio::select! {
                ev = events.recv() => match ev {
                    Ok(ev) => if let Some(id) = wakes(&ev) { dirty.insert(id); break; },
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { sweep = true; break; }
                    Err(tokio::sync::broadcast::error::RecvError::Closed) => { (&mut wake).await; sweep = true; break; }
                },
                _ = &mut wake => { sweep = true; break; }
            }
        }
        if !sweep {
            // Let a burst of pushes settle into one pass each.
            tokio::time::sleep(Duration::from_secs(1)).await;
            hear(&mut events, &mut dirty);
        }
    }
}

/// Take the events waiting: the repositories they name go in `dirty`.
/// Whether events were lost, so everything should be looked at.
fn hear(events: &mut tokio::sync::broadcast::Receiver<Event>, dirty: &mut BTreeSet<i64>) -> bool {
    use tokio::sync::broadcast::error::TryRecvError;
    loop {
        match events.try_recv() {
            Ok(ev) => {
                if let Some(id) = wakes(&ev) {
                    dirty.insert(id);
                }
            }
            Err(TryRecvError::Lagged(_)) => return true,
            Err(_) => return false,
        }
    }
}

/// The repository an event says has something to scan: a push to it
/// (`repo:N`, `push`), or a maintainer's request.
fn wakes(ev: &Event) -> Option<i64> {
    let Event::Live { topic, kind } = ev else { return None };
    if topic == REQUEST_TOPIC {
        return kind.parse().ok();
    }
    if kind != "push" {
        return None;
    }
    topic.strip_prefix("repo:")?.parse().ok()
}

/// Ask for repository `repo_id`'s whole history to be scanned again now
/// ("Scan history now", maintain role).
pub async fn request(app: &Shared, repo_id: i64) -> sqlx::Result<()> {
    pdb::request_scan(&app.db, repo_id).await?;
    wake(app, repo_id);
    Ok(())
}

/// Tell the worker that repository `repo_id` may have history to scan (its
/// push protection was turned back on).
pub fn wake(app: &Shared, repo_id: i64) {
    app.bus.live(REQUEST_TOPIC, &repo_id.to_string());
}

fn heads_text(heads: &[ChangeHash]) -> String {
    heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>().join(",")
}

fn heads_of(text: &str) -> Vec<ChangeHash> {
    text.split(',').filter_map(ChangeHash::from_hex).collect()
}

/// One pass over `row`'s history within `budget`: resume the scan under
/// way, or start one toward the bookmarks if they moved past what was
/// scanned; record what it finds and where it stopped.
pub async fn pass(app: &Shared, row: &RepoRow, budget: Budget) -> anyhow::Result<Pass> {
    let name = RepoName::new(&row.owner, &row.name).ok_or_else(|| anyhow::anyhow!("not a repository name"))?;
    let loaded = pdb::scan_of(&app.db, row.id).await?.unwrap_or(pdb::ScanRow { repo_id: row.id, ..Default::default() });
    let mut s = loaded.clone();
    let version = pp::rules().version as i64;
    if s.rules_version != version {
        // New rules: the whole history again.
        s.rules_version = version;
        s.done_heads.clear();
        s.target_heads.clear();
        s.position = 0;
    }
    let from = Progress { done: heads_of(&s.done_heads), target: heads_of(&s.target_heads), position: s.position.max(0) as usize };
    let default = row.default_bookmark.clone();
    let started = db::now_ms();
    let work = app.repos.read(&name, move |repo, _| scan_in(repo, from, &default, budget)).await;
    let work = match work {
        Ok(w) => w,
        Err(e) => {
            s.error = e.to_string();
            s.pass_ms = Some(started);
            save(app, &loaded, &s).await?;
            return Err(e.into());
        }
    };
    if work.idle && s == loaded {
        // Up to date, as the row already says: nothing to write.
        return Ok(Pass { idle: true, ..Default::default() });
    }
    let records: Vec<pdb::NewHistoryFinding> = work
        .findings
        .iter()
        .map(|(f, bookmark)| pdb::NewHistoryFinding {
            finding: pdb::NewFinding {
                rule: &f.rule,
                description: &f.description,
                path: &f.path,
                line: f.line as i64,
                fingerprint: &f.fingerprint,
                excerpt: &f.excerpt,
            },
            change_hash: f.change_hash.map(|h| h.to_hex()).unwrap_or_default(),
            bookmark: bookmark.clone(),
        })
        .collect();
    let found = pdb::record_history(&app.db, row.id, &records).await?;
    let left = work.pending - work.position;
    if work.idle || left == 0 {
        s.done_heads = heads_text(&work.target);
        s.target_heads.clear();
        s.position = 0;
        s.pending = 0;
        s.requested_ms = None;
        if !work.idle {
            s.finished_ms = Some(db::now_ms());
        }
    } else {
        s.target_heads = heads_text(&work.target);
        s.position = work.position as i64;
        s.pending = work.pending as i64;
    }
    s.changes += work.changes as i64;
    s.bytes += work.bytes as i64;
    s.pass_ms = Some(started);
    s.error.clear();
    save(app, &loaded, &s).await?;
    Ok(Pass { changes: work.changes, bytes: work.bytes, found, left, idle: work.idle })
}

/// Save `s` unless a maintainer asked for a new scan while the pass ran:
/// that request's fresh start wins.
async fn save(app: &Shared, loaded: &pdb::ScanRow, s: &pdb::ScanRow) -> sqlx::Result<()> {
    let now = pdb::scan_of(&app.db, s.repo_id).await?;
    if now.as_ref().and_then(|n| n.requested_ms) != loaded.requested_ms {
        return Ok(());
    }
    pdb::save_scan(&app.db, s).await
}

/// Where a scan stands, as read from its row.
struct Progress {
    done: Vec<ChangeHash>,
    target: Vec<ChangeHash>,
    position: usize,
}

/// What [`scan_in`] did.
struct Work {
    /// The heads it scanned toward, and how far it got of how many.
    target: Vec<ChangeHash>,
    position: usize,
    pending: usize,
    idle: bool,
    changes: usize,
    bytes: u64,
    /// Each finding, placed, with the bookmark it was found on.
    findings: Vec<(Finding, String)>,
}

/// The pass itself, on a blocking thread with the repository open.
fn scan_in(repo: &Repo, from: Progress, default: &str, budget: Budget) -> Result<Work, RepoError> {
    let started = Instant::now();
    let view = repo.view()?;
    let mut bookmarks: Vec<(String, Vec<ChangeHash>)> = Vec::new();
    for name in view.bookmarks.keys() {
        bookmarks.push((name.clone(), repo.heads_of(&view, name)?));
    }
    // The default bookmark first: a finding is named by the first bookmark
    // that reaches it.
    bookmarks.sort_by_key(|(n, _)| (n != default, n.clone()));
    let mut current: Vec<ChangeHash> = bookmarks.iter().flat_map(|(_, h)| h.iter().copied()).collect();
    current.sort();
    current.dedup();
    let mut done = from.done;
    done.sort();
    let (mut target, mut position) = (from.target, from.position);
    if target.is_empty() {
        if current == done {
            return Ok(Work { target: done, position: 0, pending: 0, idle: true, changes: 0, bytes: 0, findings: vec![] });
        }
        target = current.clone();
        position = 0;
    }
    let state = match repo.load_state_history(&target) {
        Ok(s) => s,
        // What the scan worked toward is gone (a bookmark rewritten, then
        // collected): start again toward the bookmarks as they are.
        Err(_) if target != current => {
            target = current.clone();
            position = 0;
            repo.load_state_history(&target)?
        }
        Err(e) => return Err(e.into()),
    };
    // What is scanned already: the done heads' history.
    let scanned: HashSet<ChangeHash> = if done.iter().all(|h| state.contains(h)) {
        closure(&done, |h| state.get(h).map(|m| m.deps.clone()))
    } else {
        match repo.load_state_history(&done) {
            Ok(old) => old.changes().map(|(h, _)| *h).collect(),
            Err(_) => HashSet::new(),
        }
    };
    let mut list: Vec<(oxy_core::objects::Hlc, ChangeHash)> =
        state.changes().filter(|(h, _)| !scanned.contains(*h)).map(|(h, m)| (m.hlc, *h)).collect();
    list.sort();
    let pending = list.len();
    if position > pending {
        position = 0;
    }
    let file = pp::Budget { bytes: u64::MAX, file: budget.file, time: Duration::MAX };
    let mut work = Work { target, position, pending, idle: false, changes: 0, bytes: 0, findings: vec![] };
    let mut reach: HashMap<String, HashSet<ChangeHash>> = HashMap::new();
    let mut busy = Duration::ZERO;
    let mut spent: u64 = 0;
    while work.position < pending {
        if work.changes > 0 && (spent >= budget.bytes || started.elapsed() >= budget.time) {
            break;
        }
        let t = Instant::now();
        let h = list[work.position].1;
        let bundle = bundle_of(repo, &h, budget.file)?;
        let scan = pp::scan(&bundle, file);
        work.bytes += scan.scanned;
        spent += scan.scanned.max(1);
        if !scan.findings.is_empty() {
            let mut findings = scan.findings;
            pp::locate(repo, &bundle, &mut findings);
            let on = bookmarks
                .iter()
                .find(|(name, heads)| {
                    reach.entry(name.clone()).or_insert_with(|| closure(heads, |c| state.get(c).map(|m| m.deps.clone()))).contains(&h)
                })
                .map(|(name, _)| name.clone())
                .unwrap_or_default();
            work.findings.extend(findings.into_iter().map(|f| (f, on.clone())));
        }
        work.position += 1;
        work.changes += 1;
        // Rest as long as it worked, so the scan takes at most half a core.
        busy += t.elapsed();
        if busy >= Duration::from_millis(20) {
            std::thread::sleep(busy);
            busy = Duration::ZERO;
        }
    }
    Ok(work)
}

/// `heads` and every change they depend on, by `deps`.
fn closure(heads: &[ChangeHash], deps: impl Fn(&ChangeHash) -> Option<Vec<ChangeHash>>) -> HashSet<ChangeHash> {
    let mut seen: HashSet<ChangeHash> = HashSet::new();
    let mut stack: Vec<ChangeHash> = heads.to_vec();
    while let Some(h) = stack.pop() {
        if !seen.insert(h) {
            continue;
        }
        if let Some(d) = deps(&h) {
            stack.extend(d.into_iter().filter(|x| !seen.contains(x)));
        }
    }
    seen
}

/// One stored change as push protection reads a push: the change, and the
/// blobs (with their chunks, unless over the file budget) its edits put in
/// place of a file's content. A blob the store does not hold (lazy) is
/// left out, so it is not scanned.
fn bundle_of(repo: &Repo, h: &ChangeHash, file: u64) -> Result<Bundle, RepoError> {
    let store = repo.store();
    let change = store.get_change(h)?;
    let mut objects = Vec::new();
    for e in &change.edits {
        let Edit::Replace { new: Some(r), .. } = e else { continue };
        let Ok(body) = store.get_typed(r, ObjectType::Blob) else { continue };
        let Ok(blob) = Blob::decode(&body) else { continue };
        if blob.length <= file {
            for c in &blob.chunks {
                if let Ok(chunk) = store.get_typed(c, ObjectType::Chunk) {
                    objects.push((ObjectType::Chunk, chunk));
                }
            }
        }
        objects.push((ObjectType::Blob, body));
    }
    Ok(Bundle { changes: vec![(change, vec![])], hashes: vec![*h], objects, ..Default::default() })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn pushes_and_requests_wake_the_scan() {
        let live = |topic: &str, kind: &str| Event::Live { topic: topic.into(), kind: kind.into() };
        assert_eq!(wakes(&live("repo:7", "push")), Some(7));
        assert_eq!(wakes(&live("repo:7", "issue")), None);
        assert_eq!(wakes(&live("repo:7/proposal:2", "push")), None);
        assert_eq!(wakes(&live(REQUEST_TOPIC, "9")), Some(9));
        assert_eq!(wakes(&Event::DeliveryWaiting), None);
    }

    #[test]
    fn heads_round_trip() {
        let a = ChangeHash::from_hex(&"ab".repeat(32)).unwrap();
        let b = ChangeHash::from_hex(&"cd".repeat(32)).unwrap();
        assert_eq!(heads_of(&heads_text(&[a, b])), [a, b]);
        assert!(heads_of("").is_empty());
    }
}