//! 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(&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());

    }

}