// Copyright (C) 2025-2026  The Software Heritage developers
// See the AUTHORS file at the top-level directory of this distribution
// License: GNU General Public License version 3, or any later version
// See top-level LICENSE file for more information

use std::borrow::Cow;
use std::collections::HashMap;
use std::fs::File;
use std::io::{BufReader, Write};
use std::ops::Range;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime};

use anyhow::{Context, Result, anyhow, bail};
use dashmap::{DashMap, DashSet};
#[cfg(feature = "arrow")]
use dataset_writer::{CsvZstTableWriter, ParallelDatasetWriter};
use dsi_progress_logger::ProgressLog;
use itertools::Itertools;
use rapidhash::{RapidHashMap, RapidHashSet};
use rayon::prelude::*;
use swh_contents::blame::blame::{Blame, BlameError, Blamer, LineProvenance, RevisionSkipper};
use swh_contents::blame::rangemap::RangeMap;
use swh_contents::diff::contents::{FileAtRev, FileContents};
use swh_contents::diff::diff::{DiffHunk, FileDiff};
use swh_contents::diff::differ::RevDiffer;
use swh_contents::diff::tokenizers::Tokenizer;
use swh_contents::retriever::ContentRetriever;
use swh_graph::graph::{
    NodeId, SwhGraphWithProperties, SwhLabeledBackwardGraph, SwhLabeledForwardGraph,
};
use swh_graph::labels::LabelNameId;
use swh_graph::properties;
use swh_graph::{NodeType, SWHID};
use swh_graph_stdlib::find_root_dir;
use swh_graph_stdlib::fs::resolve_path_by_id;

use crate::git;

pub mod strategies;
use strategies::{LineRange, Strategy, StrategyFactory, VulnerableFileLines};

pub mod tokenizers;
mod utils;

pub const DEFAULT_MAX_LINE_SIMILARITY_DISTANCE: u32 = 100;
pub const DEFAULT_MAX_SIMILAR_LINE_OFFSET: u32 = 1000;

#[cfg(feature = "arrow")]
/// Aliased to [`CsvZstTableWriter`] when the `arrow` feature is available, `()` otherwise
type MiddleRevWriter<'a> = CsvZstTableWriter<'a>;
#[cfg(not(feature = "arrow"))]
/// Aliased to `CsvZstTableWriter` when the `arrow` feature is available, `()` otherwise
type MiddleRevWriter<'a> = ();

fn make_path(
    graph: &impl SwhGraphWithProperties<LabelNames: properties::LabelNames>,
    path: &[LabelNameId],
) -> String {
    path.iter()
        .map(|&label_name| {
            String::from_utf8_lossy(&graph.properties().label_name(label_name)).into_owned()
        })
        .join("/")
}

/// Given a range in the new file and the diff hunks between old and new,
/// computes the corresponding range in the old file.
fn map_new_range_to_old(new_range: &Range<u32>, hunks: &[DiffHunk]) -> Result<Range<u32>> {
    let mut old_start: Option<u32> = None;
    let mut old_end: Option<u32> = None;

    let mut extend = |s: u32, e: u32| {
        if s < e {
            old_start = Some(old_start.map_or(s, |cur| cur.min(s)));
            old_end = Some(old_end.map_or(e, |cur| cur.max(e)));
        }
    };

    // offset = old_pos - new_pos for unchanged regions between hunks
    let mut prev_old_end: u32 = 0;
    let mut prev_new_end: u32 = 0;

    for hunk in hunks {
        // Unchanged region before this hunk: old[prev_old_end..hunk.old_range.start]
        // corresponds to new[prev_new_end..hunk.new_range.start]
        let offset: i64 = i64::from(prev_old_end) - i64::from(prev_new_end);
        let unchanged_new = prev_new_end..hunk.new_range.start;
        // Overlap of unchanged_new with new_range
        let overlap_start = unchanged_new.start.max(new_range.start);
        let overlap_end = unchanged_new.end.min(new_range.end);
        if overlap_start < overlap_end {
            let mapped_start = u32::try_from(i64::from(overlap_start) + offset)
                .context("Offset arithmetic overflow")?;
            let mapped_end = u32::try_from(i64::from(overlap_end) + offset)
                .context("Offset arithmetic overflow")?;
            extend(mapped_start, mapped_end);
        }

        // Hunk region: if new_range overlaps with hunk.new_range, include hunk.old_range
        let hunk_overlap_start = hunk.new_range.start.max(new_range.start);
        let hunk_overlap_end = hunk.new_range.end.min(new_range.end);
        if hunk_overlap_start < hunk_overlap_end
            || (hunk.new_range.is_empty()
                && hunk.new_range.start >= new_range.start
                && hunk.new_range.start < new_range.end)
        {
            extend(hunk.old_range.start, hunk.old_range.end);
        }

        prev_old_end = hunk.old_range.end;
        prev_new_end = hunk.new_range.end;
    }

    // Unchanged region after last hunk
    let offset: i64 = i64::from(prev_old_end) - i64::from(prev_new_end);
    let overlap_start = prev_new_end.max(new_range.start);
    let overlap_end = new_range.end; // no upper bound from hunks
    if overlap_start < overlap_end {
        let mapped_start = u32::try_from(i64::from(overlap_start) + offset)
            .context("Offset arithmetic overflow")?;
        let mapped_end =
            u32::try_from(i64::from(overlap_end) + offset).context("Offset arithmetic overflow")?;
        extend(mapped_start, mapped_end);
    }

    Ok(old_start.unwrap_or(new_range.start)..old_end.unwrap_or(new_range.start))
}

fn get_introduction_range(offset: i32, range: &Range<u32>) -> Result<LineRange> {
    let start = range
        .start
        .checked_add_signed(offset)
        .with_context(|| format!("Applying offset {offset} to range {range:?} overflows"))?;
    let end = range
        .end
        .checked_add_signed(offset)
        .with_context(|| format!("Applying offset {offset} to range {range:?} overflows"))?;
    Ok(LineRange(start..end))
}

/// Maps an introduction range back through diff hunks to produce the hunk content
/// before introduction.
fn build_hunk_before_introduction(
    introduction_range: &LineRange,
    parent_info: Option<&(FileContents, Box<[DiffHunk]>)>,
) -> Result<Option<HunkRecord>> {
    let Some((parent_contents, hunks)) = parent_info else {
        return Ok(None);
    };
    let old_range = map_new_range_to_old(&introduction_range.0, hunks)?;
    let hunk_text = old_range
        .clone()
        .map(|i| {
            String::from_utf8_lossy(
                parent_contents
                    .lines()
                    .expect("parent file was text and is now binary")
                    .get(i)
                    .expect("parent line index in range"),
            )
        })
        .join("");
    Ok(Some(HunkRecord {
        hunk: hunk_text,
        line_range: LineRange(old_range),
    }))
}

#[derive(thiserror::Error, Debug)]
pub enum SzzError {
    #[error("Could not run tree diff between {parent} and {child}: {source}")]
    TreeDiff {
        parent: SWHID,
        child: SWHID,
        #[source]
        source: anyhow::Error,
    },
    #[error("Could not annotate lines of {path} in {revision}: {source}")]
    Annotation {
        path: String,
        revision: SWHID,
        #[source]
        source: BlameError,
    },
    #[error("History is too deep")]
    MaxDepth,
    #[error("Revision makes too many changes")]
    TooManyChanges,
    #[error("Not implemented yet: {0}")]
    NotImplemented(&'static str),
    #[error("{0}")]
    Other(#[from] anyhow::Error),
}

#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct IntroductionFile {
    pub line_ranges: RapidHashSet<Range<u32>>,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct IntroductionFiles(pub HashMap<Box<[LabelNameId]>, IntroductionFile>);
#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct IntroductionFilesRecord(pub HashMap<String, IntroductionFile>);

#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct SzzOutputRecord {
    /// Filename of the vulnerability report in the OSV database
    pub vuln_filename: String,
    /// Vulnerabilities that are a fix according to the vulnerability report
    pub fix_revs: Vec<SWHID>,
    /// Union of tag names of each revision in `fix_revs_id`/`fix_revs_swhid`.
    ///
    /// For each revision in `fix_revs_swhid`, takes the name of any release that points
    /// **directly** to it, or the name of a `refs/tags/` branch that points **directly** to it.
    pub fix_revs_tag_names: Vec<String>,
    /// Whether the vulnerability report claims the vulnerability is present since the beginning of
    /// the project
    ///
    /// ie. the event `{ "introduced": "0" }` is present.
    ///
    /// This is usually a lie, and actually means the author of the vulnerability report does not
    /// known when the vulnerability was introduced.
    pub introduced_at_zero: bool,
    /// Vulnerabilities that introduced the vulnerability according to the vulnerability report
    pub known_introduction_revs: Vec<SWHID>,
    /// Union of tag names of each revision in
    /// `known_introduction_revs_id`/`known_introduction_revs_swhid`.
    pub known_introduction_revs_tag_names: Vec<String>,
    /// Vulnerabilities that introduced the vulnerability according to SZZ,
    /// and for each path, which lines in it are considered to be the introduction
    pub computed_introduction_revs: HashMap<SWHID, IntroductionFilesRecord>,
    /// Union of tag names of each revision in
    /// `computed_introduction_revs_id`/`computed_introduction_revs_swhid`.
    pub computed_introduction_revs_tag_names: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct HunkRecord {
    pub hunk: String,
    pub line_range: LineRange,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct RevisionRecord {
    pub swhid: SWHID,
    pub author_timestamp: Option<i64>,
    pub committer_timestamp: Option<i64>,
}

/// Provides line-level details on a SZZ run, complementary to [`SzzOutputRecord`]
#[derive(Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct SzzLineDetailsRecord {
    pub fix_rev: RevisionRecord,
    pub path: String,
    pub vulnerable_hunk_before_fix: HunkRecord,
    /// Introduction rev computed by SZZ
    pub introduction_rev: RevisionRecord,
    pub vulnerable_hunk_after_introduction: HunkRecord,
    pub hunk_before_introduction: Option<HunkRecord>,
    pub file_creation: RapidHashSet<RevisionRecord>,
    /// number of blame steps from the fix rev to find the introduction rev
    ///
    /// It is found through a DFS or BFS, so it is greater or equal to `intro_to_fix_rev_distance`.
    pub intro_to_fix_rev_num_blame_steps: u64,
    /// number of revisions between introduction_rev and fix_rev (shortest path)
    pub intro_to_fix_rev_distance: u64,
    /// number of revisions between file_creation_rev and introduction_rev (shortest path)
    pub creation_to_intro_rev_distance: u64,
    /// number of revisions between file_creation_rev and fix_rev (shortest path, usually equal
    /// to `intro_to_fix_rev_distance + creation_to_intro_rev_distance` but may be smaller)
    pub creation_to_fix_rev_distance: u64,
}

#[derive(Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct MiddleCommitRecord<'a> {
    /// parent revision of `predecessor_rev`
    pub rev: SWHID,
    pub rev_author_timestamp: Option<i64>,
    pub rev_committer_timestamp: Option<i64>,
    /// middle rev touching the line
    pub predecessor_rev: SWHID,
    pub predecessor_rev_author_timestamp: Option<i64>,
    pub predecessor_rev_committer_timestamp: Option<i64>,
    pub hunk_id: String,
    /// fix rev
    pub ancestor_rev: SWHID,
    pub ancestor_rev_author_timestamp: Option<i64>,
    pub ancestor_rev_committer_timestamp: Option<i64>,
    /// always "V-SZZ_middle_revision"
    pub tag: Cow<'a, str>,
    pub code_before_id: SWHID,
    pub code_after_id: SWHID,
    pub file_path: String,
    /// entire file at 'rev'
    pub code_before: Cow<'a, str>,
    /// entire file at 'predecessor_rev'
    pub code_after: Cow<'a, str>,
}

/// Extracts all unique revision NodeIds from SZZ output files in a directory.
///
/// Reads all ndjson files in the given directory (expected to contain `SzzOutputRecord` entries),
/// and collects all revision IDs mentioned as fix revisions, known introduction revisions,
/// or computed introduction revisions. Returns the union of all these revision IDs.
///
/// Files in the directory can be compressed; decompression is handled automatically.
pub fn list_revs_from_szz_output<G: SwhGraphWithProperties<Maps: properties::Maps> + Sync>(
    graph: &G,
    dir: &Path,
    middle_revisions_dir: Option<&Path>,
) -> Result<rapidhash::RapidHashSet<NodeId>> {
    let entries: Vec<_> = std::fs::read_dir(dir)
        .with_context(|| format!("Could not stat directory {}", &dir.display()))?
        .map(|entry| Ok(entry?))
        .collect::<Result<_>>()
        .with_context(|| format!("Could not stat {} entry", dir.display()))?;

    let middle_revisions_entries: Vec<_> = match middle_revisions_dir {
        Some(middle_revisions_dir) => std::fs::read_dir(middle_revisions_dir)
            .with_context(|| format!("Could not stat directory {}", &dir.display()))?
            .map(|entry| Ok(entry?))
            .collect::<Result<_>>()
            .with_context(|| format!("Could not stat {} entry", dir.display()))?,
        None => vec![],
    };

    let mut pl = dsi_progress_logger::concurrent_progress_logger!(
        display_memory = true,
        item_name = "record",
        local_speed = true,
    );
    pl.start("Listing revisions...");

    let szz_revs = entries
        .into_par_iter()
        .map_with(pl.clone(), |pl, entry| -> Result<_> {
            let path = entry.path();
            let file =
                File::open(&path).with_context(|| format!("Could not open {}", path.display()))?;
            let decoder = deko::AnyDecoder::new(BufReader::new(file));
            let mut revs = rapidhash::RapidHashSet::default();
            for record in serde_json::StreamDeserializer::new(serde_json::de::IoRead::new(decoder))
            {
                let SzzOutputRecord {
                    vuln_filename: _,
                    fix_revs,
                    fix_revs_tag_names: _,
                    introduced_at_zero: _,
                    known_introduction_revs,
                    known_introduction_revs_tag_names: _,
                    computed_introduction_revs,
                    computed_introduction_revs_tag_names: _,
                } = record.context("Could not deserialize ndjson record")?;
                revs.extend(
                    fix_revs
                        .into_iter()
                        .map(|rev| {
                            graph
                                .properties()
                                .node_id(rev)
                                .with_context(|| format!("Unknown fix rev {rev}"))
                        })
                        .collect::<Result<Vec<_>>>()?,
                );
                revs.extend(
                    known_introduction_revs
                        .into_iter()
                        .map(|rev| {
                            graph
                                .properties()
                                .node_id(rev)
                                .with_context(|| format!("Unknown known introduction rev {rev}"))
                        })
                        .collect::<Result<Vec<_>>>()?,
                );
                revs.extend(
                    computed_introduction_revs
                        .into_keys()
                        .map(|rev| {
                            graph
                                .properties()
                                .node_id(rev)
                                .with_context(|| format!("Unknown computed introduction rev {rev}"))
                        })
                        .collect::<Result<Vec<_>>>()?,
                );
                pl.light_update();
            }
            Ok(revs)
        });
    let middle_revisions_revs =
        middle_revisions_entries
            .into_par_iter()
            .map_with(pl.clone(), |pl, entry| -> Result<_> {
                let path = entry.path();
                let file = File::open(&path)
                    .with_context(|| format!("Could not open {}", path.display()))?;
                let decoder = deko::AnyDecoder::new(BufReader::new(file));
                let mut revs = rapidhash::RapidHashSet::default();
                for record in csv::Reader::from_reader(decoder).deserialize() {
                    let MiddleCommitRecord {
                        rev,
                        rev_author_timestamp: _,
                        rev_committer_timestamp: _,
                        predecessor_rev,
                        predecessor_rev_author_timestamp: _,
                        predecessor_rev_committer_timestamp: _,
                        hunk_id: _,
                        ancestor_rev: fix_rev,
                        ancestor_rev_author_timestamp: _,
                        ancestor_rev_committer_timestamp: _,
                        tag: _,
                        file_path: _,
                        code_before_id: _,
                        code_after_id: _,
                        code_before: _,
                        code_after: _,
                    } = record.context("Could not deserialize CSV record")?;
                    for node in [rev, predecessor_rev, fix_rev] {
                        revs.insert(
                            graph
                                .properties()
                                .node_id(node)
                                .with_context(|| format!("Unknown revision {node}"))?,
                        );
                    }
                    pl.light_update()
                }
                Ok(revs)
            });

    let revs = szz_revs.chain(middle_revisions_revs).try_reduce(
        rapidhash::RapidHashSet::default,
        |left, right| {
            let (mut larger, smaller) = if left.len() >= right.len() {
                (left, right)
            } else {
                (right, left)
            };
            larger.extend(smaller);
            Ok(larger)
        },
    )?;
    pl.done();

    Ok(revs)
}

#[derive(Debug, Default)]
pub struct SzzStats {
    pub ignored_because_unknown_commit: AtomicU64,
    pub ignored_because_last_affected: AtomicU64,
    pub ignored_because_limit: AtomicU64,
    pub ignored_because_more_than_one_fix: AtomicU64,
    pub ignored_because_unavailable_content: AtomicU64,
    pub num_successful_szz_runs: AtomicU64,
    pub total_known_introduction_revs: AtomicU64,
    pub total_computed_introduction_revs: AtomicU64,
}

/// Caches results of file creation rev lookups across calls that share ancestor graphs.
#[allow(clippy::type_complexity)]
pub struct FileCreationCache<G> {
    graph: G,
    cache: DashMap<(NodeId, Box<[LabelNameId]>), HashMap<NodeId, u64>, rapidhash::RapidBuildHasher>,
    /// only save a cache item for 1 revision out of N
    pub caching_interval: usize,
}

impl<G> FileCreationCache<G> {
    pub fn new(graph: G) -> Self {
        Self {
            graph,
            cache: DashMap::with_hasher(rapidhash::RapidBuildHasher::default()),
            caching_interval: 1_000,
        }
    }
}

impl<G> FileCreationCache<G>
where
    G: SwhLabeledForwardGraph
        + SwhGraphWithProperties<Maps: properties::Maps, LabelNames: properties::LabelNames>,
{
    /// Returns all revisions in the ancestry of `start_rev` (inclusive) that introduced the file
    /// at `path`, ie. revisions that have the file but none of their revision parents do.
    ///
    /// Results are memoized per (node, path) so repeated calls with overlapping ancestor graphs
    /// reuse previous work; as resolve_path_by_id is otherwise pretty expensive.
    pub fn find(&self, start_rev: NodeId, path: &[LabelNameId]) -> Result<HashMap<NodeId, u64>> {
        let graph = &self.graph;
        let path_key: Box<[LabelNameId]> = path.into();

        if let Some(result) = self.cache.get(&(start_rev, path_key.clone())) {
            return Ok(result.clone());
        }

        let mut path_resolver = swh_graph_stdlib::fs::CachedPathResolver::new(&self.graph, path);
        let mut rev_has_file = |rev: NodeId| -> bool {
            let Ok(Some(root_dir)) = find_root_dir(graph, rev) else {
                return false;
            };
            path_resolver
                .resolve_from(root_dir)
                .unwrap_or_else(|e| {
                    log::debug!(
                        "Cannot resolve path {} from {} (root dir of {}): {e:?}",
                        make_path(graph, path),
                        graph.properties().swhid(root_dir),
                        graph.properties().swhid(rev)
                    );
                    None
                })
                .is_some()
        };

        // Maps each visited node to the file-having revision parents we discovered for it.
        let mut file_having_parents: HashMap<NodeId, Vec<NodeId>> = HashMap::new();
        // Per-node creation-rev results, built bottom-up.
        let mut local_cache: HashMap<NodeId, HashMap<NodeId, u64>> = HashMap::new();
        let mut explored = RapidHashSet::default();
        let mut stack: Vec<NodeId> = vec![start_rev];
        let mut computed_count: usize = 0;
        let mut to_persist: Vec<NodeId> = Vec::new();

        while let Some(&rev) = stack.last() {
            if local_cache.contains_key(&rev) {
                stack.pop();
                continue;
            }

            if let Some(cached) = self.cache.get(&(rev, path_key.clone())) {
                local_cache.insert(rev, cached.clone());
                stack.pop();
                continue;
            }

            if explored.contains(&rev) {
                // Second visit: all parents are processed.
                let parents = file_having_parents.get(&rev);
                let result = match parents {
                    Some(parents) if !parents.is_empty() => {
                        let mut merged: HashMap<NodeId, u64> = HashMap::new();
                        for &parent in parents {
                            let parent_result = local_cache.get(&parent).expect(
                                "parent should be in local_cache after post-order traversal",
                            );
                            for (&creation_rev, &dist) in parent_result {
                                let entry = merged.entry(creation_rev).or_insert(u64::MAX);
                                *entry = (*entry).min(dist + 1);
                            }
                        }
                        merged
                    }
                    _ => {
                        // No parent has the file: this rev is a creation rev.
                        HashMap::from([(rev, 0)])
                    }
                };
                if computed_count % self.caching_interval == 0 {
                    to_persist.push(rev);
                }
                computed_count += 1;
                local_cache.insert(rev, result);
                stack.pop();
            } else {
                // First visit: discover file-having revision parents.
                explored.insert(rev);
                for parent in graph.successors(rev) {
                    if graph.properties().node_type(parent) != NodeType::Revision {
                        continue;
                    }
                    if rev_has_file(parent) {
                        file_having_parents.entry(rev).or_default().push(parent);
                        if !explored.contains(&parent) && !local_cache.contains_key(&parent) {
                            stack.push(parent);
                        }
                    }
                }
            }
        }

        for rev in to_persist {
            self.cache
                .entry((rev, path_key.clone()))
                .or_insert_with(|| local_cache[&rev].clone());
        }

        Ok(local_cache.remove(&start_rev).unwrap_or_default())
    }
}

/// Processes vulnerabilities using the SZZ algorithm to identify introduction commits.
pub struct SzzProcessor<
    G: SwhLabeledForwardGraph
        + SwhLabeledBackwardGraph
        + SwhGraphWithProperties<
            Maps: properties::Maps,
            LabelNames: properties::LabelNames,
            Strings: properties::Strings,
            Timestamps: properties::Timestamps,
        > + Sync
        + Send,
    CR: ContentRetriever + Send + Sync,
    SF: StrategyFactory,
    RS: RevisionSkipper + Clone,
    T: Tokenizer + Clone,
> {
    graph: Arc<G>,
    content_retriever: Arc<CR>,
    concurrent_downloads: usize,
    fix_rev_to_computed_introduction_revs: DashMap<NodeId, RapidHashMap<NodeId, IntroductionFiles>>,
    rev_distances: DashMap<(NodeId, NodeId), u64>,
    file_creation_cache: FileCreationCache<Arc<G>>,
    stats: SzzStats,
    max_tree_diff_items: Option<usize>,
    max_blame_steps_per_hunk: Option<u64>,
    max_hunks: Option<u64>,
    max_modified_files: Option<u64>,
    strategy_factory: SF,
    blamer: Blamer<Arc<G>, Arc<CR>, T, RS>,
    min_line_similarity_ratio: Option<f64>,
    max_line_similarity_distance: u32,
    max_similar_line_offset: usize,
    #[cfg(feature = "arrow")]
    middle_rev_writer: Option<ParallelDatasetWriter<CsvZstTableWriter<'static>>>,
    written_middle_revs: DashSet<(NodeId, NodeId, NodeId, Box<[LabelNameId]>)>,
}

impl<G, CR, SF, RS, T> SzzProcessor<G, CR, SF, RS, T>
where
    G: SwhLabeledForwardGraph
        + SwhLabeledBackwardGraph
        + SwhGraphWithProperties<
            Maps: properties::Maps,
            LabelNames: properties::LabelNames,
            Strings: properties::Strings,
            Timestamps: properties::Timestamps,
        > + Sync
        + Send,
    CR: ContentRetriever + Send + Sync,
    SF: StrategyFactory,
    RS: RevisionSkipper + Clone,
    T: Tokenizer + Clone,
{
    pub fn new(
        graph: Arc<G>,
        content_retriever: Arc<CR>,
        strategy_factory: SF,
        blamer: Blamer<Arc<G>, Arc<CR>, T, RS>,
    ) -> Self {
        let file_creation_cache = FileCreationCache::new(graph.clone());
        Self {
            graph,
            content_retriever,
            concurrent_downloads: 10,
            fix_rev_to_computed_introduction_revs: DashMap::default(),
            rev_distances: DashMap::default(),
            file_creation_cache,
            stats: SzzStats::default(),
            max_tree_diff_items: None,
            max_blame_steps_per_hunk: None,
            max_hunks: None,
            max_modified_files: None,
            strategy_factory,
            blamer,
            min_line_similarity_ratio: None,
            max_line_similarity_distance: DEFAULT_MAX_LINE_SIMILARITY_DISTANCE,
            max_similar_line_offset: usize::try_from(DEFAULT_MAX_SIMILAR_LINE_OFFSET)
                .expect("DEFAULT_MAX_SIMILAR_LINE_OFFSET overflows usize"),
            #[cfg(feature = "arrow")]
            middle_rev_writer: None,
            written_middle_revs: DashSet::default(),
        }
    }

    /// Defaults to 100
    pub fn with_concurrent_downloads_per_task(mut self, concurrent_downloads: usize) -> Self {
        self.concurrent_downloads = concurrent_downloads;
        self
    }

    pub fn with_max_tree_diff_items(mut self, max_tree_diff_items: Option<usize>) -> Self {
        self.max_tree_diff_items = max_tree_diff_items;
        self
    }

    pub fn with_max_blame_steps_per_hunk(mut self, max_blame_steps_per_hunk: Option<u64>) -> Self {
        self.max_blame_steps_per_hunk = max_blame_steps_per_hunk;
        self
    }

    pub fn with_max_hunks(mut self, max_hunks: Option<u64>) -> Self {
        self.max_hunks = max_hunks;
        self
    }

    pub fn with_max_modified_files(mut self, max_modified_files: Option<u64>) -> Self {
        self.max_modified_files = max_modified_files;
        self
    }

    /// Defaults to `None`.
    /// If `Some( < 1.)`, applies fuzzy-matching to lines while looking for an introducing commit
    ///
    /// If a line would be attributed to a revision while blaming but there is a similar line
    /// (levenshtein distance < this_ratio * line_len) in a parent revision, that line is
    /// assumed to be an earlier version of it.
    ///
    /// See [the V-SZZ algorithm](https://baolingfeng.github.io/papers/ICSE2022VSZZ.pdf#page=4)
    /// for details:
    ///
    /// > We compute the line similarity based on the edit distance
    /// > and set the similarity threshold is equal to 0.75. But this might
    /// > result in some incorrect mappings, which would be discussed in
    /// > the evaluation (see Section 5). If the line mapping algorithm can
    /// > identify a corresponding line in the previous commit or descen-
    /// > dants commits, we continue going back by running the git blame
    /// > command for the line. This process stops until the line mapping
    /// > algorithm identifies the line as a new line, the source code file of
    /// > the line is firstly added, or the commit is the initial commit of the
    /// > repository.
    pub fn with_min_line_similarity_ratio(
        mut self,
        min_line_similarity_ratio: Option<f64>,
    ) -> Self {
        self.min_line_similarity_ratio = min_line_similarity_ratio;
        self
    }

    /// Upper bound on the edit distance of two similar lines.
    ///
    /// This complements [`Self::with_min_line_similarity_ratio`] for large lines.
    ///
    /// Default value: [`DEFAULT_MAX_LINE_SIMILARITY_DISTANCE`]
    pub fn with_max_line_similarity_distance(mut self, max_line_similarity_distance: u32) -> Self {
        self.max_line_similarity_distance = max_line_similarity_distance;
        self
    }

    /// Defines the size of the window of old lines where to look for similar lines to a new line
    ///
    /// For example, if `max_line_similarity_distance = 1000`, only the 1k old lines
    /// before and after a new line's position will be checked for similarity.
    ///
    /// Larger values linearly impact performance on large files.
    ///
    /// Default value: [`DEFAULT_MAX_SIMILAR_LINE_OFFSET`]
    pub fn with_max_similar_line_offset(mut self, max_similar_line_offset: u32) -> Self {
        self.max_similar_line_offset =
            usize::try_from(max_similar_line_offset).unwrap_or(usize::MAX); // number of lines can't exceed usize::MAX anyway, because they need to fit in a vec.
        self
    }

    /// Write a CSV entry every time a similar line is found.
    ///
    /// Does not write anything unless [`Self::with_min_line_similarity_ratio`] is used
    pub fn with_middle_revisions_out_dir(
        mut self,
        middle_revisions_out_dir: &Path,
    ) -> Result<Self> {
        #[cfg(feature = "arrow")]
        {
            self.middle_rev_writer = Some(
                ParallelDatasetWriter::<CsvZstTableWriter>::new(
                    middle_revisions_out_dir.to_path_buf(),
                )
                .with_context(|| {
                    format!(
                        "Could not create directory {}",
                        middle_revisions_out_dir.display()
                    )
                })?,
            );
            Ok(self)
        }
        #[cfg(not(feature = "arrow"))]
        {
            let _ = (&mut self, middle_revisions_out_dir); // for clippy
            bail!(
                "middle_revisions_out_dir is not supported as swh-osv was compiled without the 'arrow' feature"
            )
        }
    }

    /// Returns the shortest number of revision hops from `from` to `to` through the ancestry graph.
    ///
    /// Returns an error if `from` is not an ancestor of `to`.
    fn rev_distance(&self, from: NodeId, to: NodeId) -> Result<u64> {
        if from == to {
            return Ok(0);
        }
        self.rev_distances
            .entry((from, to))
            .or_try_insert_with(|| {
                let graph = &self.graph;
                let mut queue = std::collections::VecDeque::new();
                let mut visited = RapidHashSet::default();
                let mut depth = 0u64;
                queue.push_back(to);
                visited.insert(to);
                queue.push_back(usize::MAX); // sentinel: end of depth 0
                while let Some(node) = queue.pop_front() {
                    if node == usize::MAX {
                        if queue.is_empty() {
                            break;
                        }
                        depth += 1;
                        queue.push_back(usize::MAX);
                        continue;
                    }
                    for succ in graph.successors(node) {
                        if graph.properties().node_type(succ) != NodeType::Revision {
                            continue;
                        }
                        if succ == from {
                            return Ok(depth + 1);
                        }
                        if visited.insert(succ) {
                            queue.push_back(succ);
                        }
                    }
                }
                bail!(
                    "{} is not an ancestor of {}",
                    graph.properties().swhid(from),
                    graph.properties().swhid(to)
                )
            })
            .map(|v| *v)
    }

    /// Fetches file contents at the introduction revision and inserts into cache
    /// if not already present.
    async fn fetch_intro_file_contents(
        &self,
        cache: &mut HashMap<NodeId, FileContents>,
        intro_rev: NodeId,
        path: &[LabelNameId],
    ) -> Result<()> {
        use std::collections::hash_map::Entry;

        if let Entry::Vacant(e) = cache.entry(intro_rev) {
            let graph = &self.graph;
            let root_dir = find_root_dir(graph.as_ref(), intro_rev)?.with_context(|| {
                format!(
                    "No root dir for introduction rev {}",
                    graph.properties().swhid(intro_rev)
                )
            })?;
            let cnt_id =
                resolve_path_by_id(graph.as_ref(), root_dir, path)?.with_context(|| {
                    format!(
                        "Could not resolve path in introduction rev {}",
                        graph.properties().swhid(intro_rev)
                    )
                })?;
            let file_contents = FileContents::fetch(
                graph.as_ref(),
                &self.content_retriever,
                &FileAtRev {
                    rev: intro_rev,
                    content: cnt_id,
                    path,
                    perm: None,
                },
            )
            .await
            .with_context(|| {
                format!(
                    "Could not fetch content at introduction rev {}",
                    graph.properties().swhid(intro_rev)
                )
            })?;
            e.insert(file_contents);
        }
        Ok(())
    }

    /// Finds the single parent of the introduction rev that has the file, diffs them,
    /// and caches the result. Caches `None` for file creation or merges.
    #[allow(clippy::type_complexity)]
    async fn fetch_intro_parent_diff(
        &self,
        cache: &mut HashMap<NodeId, Option<(FileContents, Box<[DiffHunk]>)>>,
        differ: &RevDiffer<Arc<G>, Arc<CR>, T>,
        intro_rev: NodeId,
        path: &[LabelNameId],
    ) -> Result<()> {
        use std::collections::hash_map::Entry;

        if let Entry::Vacant(e) = cache.entry(intro_rev) {
            let graph = &self.graph;

            let parents_with_file: Vec<(NodeId, NodeId)> =
                tokio::task::block_in_place(|| -> Result<_> {
                    let mut result = Vec::new();
                    for parent in graph.successors(intro_rev) {
                        if graph.properties().node_type(parent) != NodeType::Revision {
                            continue;
                        }
                        let Ok(Some(root_dir)) = find_root_dir(graph.as_ref(), parent) else {
                            continue;
                        };
                        if let Ok(Some(cnt_id)) = resolve_path_by_id(graph.as_ref(), root_dir, path)
                        {
                            result.push((parent, cnt_id));
                        }
                    }
                    Ok(result)
                })?;

            if parents_with_file.len() == 1 {
                let (parent, parent_cnt_id) = parents_with_file[0];
                let intro_cnt_id = tokio::task::block_in_place(|| {
                    let root_dir = find_root_dir(graph.as_ref(), intro_rev)?
                        .context("No root dir for intro rev")?;
                    resolve_path_by_id(graph.as_ref(), root_dir, path)?
                        .context("Could not resolve path in intro rev")
                })?;
                let file_diff = differ
                    .diff_file(
                        Some(&FileAtRev {
                            rev: parent,
                            content: parent_cnt_id,
                            path,
                            perm: None,
                        }),
                        &FileAtRev {
                            rev: intro_rev,
                            content: intro_cnt_id,
                            path,
                            perm: None,
                        },
                    )
                    .await
                    .with_context(|| {
                        format!(
                            "Could not diff parent and intro rev {}",
                            graph.properties().swhid(intro_rev)
                        )
                    })?;
                let value = match file_diff {
                    FileDiff::Modified {
                        old_file,
                        new_file: _,
                        hunks: Some(hunks),
                    } => Some((old_file, hunks)),
                    FileDiff::Modified { hunks: None, .. }
                    | FileDiff::Added { .. }
                    | FileDiff::Removed { .. } => None,
                };
                e.insert(value);
            } else {
                // No parent with file (file creation/root) or multiple parents (merge)
                e.insert(None);
            }
        }
        Ok(())
    }

    /// Writes detailed line-level SZZ records for every vulnerable line.
    #[allow(clippy::too_many_arguments)]
    async fn write_line_details(
        &self,
        writer: &mut (impl Write + ?Sized),
        differ: &RevDiffer<Arc<G>, Arc<CR>, T>,
        fix_rev: NodeId,
        parent_rev: NodeId,
        old_path: &[LabelNameId],
        lossy_decoded_old_path: &str,
        blame: &Blame<Arc<G>, Arc<CR>, T, RS>,
        blame_steps: u64,
    ) -> Result<()> {
        let graph = &self.graph;
        let fix_rev_swhid = graph.properties().swhid(fix_rev);

        let file_creation_revs = match self.file_creation_cache.find(parent_rev, old_path) {
            Ok(file_creation_revs) => file_creation_revs,
            Err(e) => {
                log::error!(
                    "Cannot find which revision created {} of {}: {e:?}",
                    make_path(&self.graph, old_path),
                    graph.properties().swhid(parent_rev)
                );
                return Ok(());
            }
        };
        let final_contents: Vec<_> = blame
            .final_contents()
            .lines()
            .context("Could not get lines in final content")?
            .iter()
            .collect();
        let mut intro_file_contents: HashMap<NodeId, FileContents> = HashMap::new();
        #[allow(clippy::type_complexity)]
        let mut intro_parent_info: HashMap<
            NodeId,
            Option<(FileContents, Box<[DiffHunk]>)>,
        > = HashMap::new();
        let mut rev_records = HashMap::new();

        for (range, provenance) in blame.ranges().iter() {
            self.fetch_intro_file_contents(&mut intro_file_contents, provenance.rev, old_path)
                .await?;

            self.fetch_intro_parent_diff(&mut intro_parent_info, differ, provenance.rev, old_path)
                .await?;

            let introduction_range = get_introduction_range(provenance.offset, &range)?;

            let hunk_before_introduction = build_hunk_before_introduction(
                &introduction_range,
                intro_parent_info
                    .get(&provenance.rev)
                    .and_then(|v| v.as_ref()),
            )?;

            let mut rev_record = |rev| {
                rev_records
                    .entry(rev)
                    .or_insert_with(|| RevisionRecord {
                        swhid: graph.properties().swhid(rev),
                        author_timestamp: graph.properties().author_timestamp(rev),
                        committer_timestamp: graph.properties().committer_timestamp(rev),
                    })
                    .clone()
            };

            let record = SzzLineDetailsRecord {
                fix_rev: rev_record(fix_rev),
                path: lossy_decoded_old_path.to_owned(),
                vulnerable_hunk_before_fix: HunkRecord {
                    hunk: final_contents[range.start as usize..range.end as usize]
                        .iter()
                        .map(|line| String::from_utf8_lossy(line))
                        .join(""),
                    line_range: LineRange(range),
                },
                introduction_rev: rev_record(provenance.rev),
                vulnerable_hunk_after_introduction: HunkRecord {
                    hunk: introduction_range
                        .0
                        .clone()
                        .map(|i| {
                            String::from_utf8_lossy(
                                intro_file_contents
                                    .get(&provenance.rev)
                                    .expect("introduction file should be cached, but is missing")
                                    .lines()
                                    .expect("introduction file was text and is now binary")
                                    .get(i)
                                    .expect("introduction line index in range"),
                            )
                        })
                        .join(""),
                    line_range: introduction_range,
                },
                intro_to_fix_rev_num_blame_steps: blame_steps,
                intro_to_fix_rev_distance: self.rev_distance(provenance.rev, fix_rev)?,
                creation_to_intro_rev_distance: {
                    let intro_creation_revs =
                        self.file_creation_cache.find(provenance.rev, old_path)?;
                    intro_creation_revs.values().copied().min().ok_or_else(|| {
                        anyhow!(
                            "No file creation rev is an ancestor of introduction rev {}",
                            graph.properties().swhid(provenance.rev)
                        )
                    })?
                },
                creation_to_fix_rev_distance: file_creation_revs
                    .values()
                    .copied()
                    .min()
                    .map(|d| d + 1)
                    .ok_or_else(|| {
                        anyhow!("No file creation rev is an ancestor of fix rev {fix_rev_swhid}")
                    })?,
                file_creation: file_creation_revs
                    .keys()
                    .map(|&cr| rev_record(cr))
                    .collect(),
                hunk_before_introduction,
            };
            serde_json::to_writer(&mut *writer, &record)
                .with_context(|| format!("Could not write {record:?}"))?;
            writer.write_all(b"\n").context("Could not end line")?;
        }
        Ok(())
    }

    /// Given a revision that fixes a vulnerability, returns a set of commits that probably
    /// introduced the vulnerability.
    pub async fn szz<W>(
        &self,
        differ: &RevDiffer<Arc<G>, Arc<CR>, T>,
        strategy: &mut SF::Strategy<'_>,
        fix_rev: NodeId,
        line_details_writer: &mut Option<W>,
    ) -> Result<rapidhash::RapidHashMap<NodeId, IntroductionFiles>, SzzError>
    where
        W: std::ops::DerefMut,
        W::Target: Write,
    {
        let graph = &self.graph;
        let fix_rev_swhid = graph.properties().swhid(fix_rev);
        if graph.properties().node_type(fix_rev) != NodeType::Revision {
            Err(anyhow!("{fix_rev} is not a revision"))?;
        }
        let res = tokio::task::block_in_place(|| {
            let mut parent_revs: Vec<_> = graph
                .successors(fix_rev)
                .into_iter()
                .filter(|&node| graph.properties().node_type(node) == NodeType::Revision)
                .collect();
            if parent_revs.is_empty() {
                return Ok(None);
            }
            if parent_revs.len() > 1 {
                return Err(SzzError::NotImplemented("merge commits"));
            }

            let parent_rev = parent_revs.pop().unwrap();
            let parent_rev_swhid = graph.properties().swhid(parent_rev);
            Ok(Some((parent_rev, parent_rev_swhid)))
        })?;

        let Some((parent_rev, parent_rev_swhid)) = res else {
            // fix_rev is a root commit, so there is no introduction revision
            log::debug!("fix_rev {} is a root commit", fix_rev_swhid);
            return Ok(Default::default());
        };

        // Compute the diff between the fix revision and its parent (ie. compute the patch)
        let diff = differ
            .diff_revisions(Some(parent_rev), fix_rev, self.max_tree_diff_items)
            .await
            .map_err(|source| SzzError::TreeDiff {
                parent: graph.properties().swhid(parent_rev),
                child: graph.properties().swhid(fix_rev),
                source,
            })?;

        log::debug!(
            "fix_rev={fix_rev} {fix_rev_swhid}, parent_rev={parent_rev} {parent_rev_swhid}, diff:\n{diff}",
        );

        // count the number of lines changed/removed by the fix that were added by a given rev
        let mut introduction_files_by_rev = RapidHashMap::<NodeId, IntroductionFiles>::default();

        let changes: Vec<_> = tokio::task::block_in_place(|| diff.changes().collect());
        self.check_limits(&changes)?;

        for res in strategy.vulnerable_hunks(fix_rev, &changes)? {
            let VulnerableFileLines {
                path: old_path,
                lines: mut hunks,
            } = res?;

            // remove empty ranges
            hunks.retain(|LineRange(old_range)| old_range.start != old_range.end);

            let lossy_decoded_old_path = make_path(&graph, &old_path);

            if hunks.is_empty() {
                // annoying case to handle below
                if let Some(writer) = line_details_writer {
                    let blame = self
                        .blamer
                        .blame(parent_rev, old_path.clone(), Some(0..0))
                        .await
                        .map_err(|source| SzzError::Annotation {
                            path: lossy_decoded_old_path.clone(),
                            revision: parent_rev_swhid,
                            source,
                        })?;
                    self.write_line_details(
                        &mut **writer,
                        differ,
                        fix_rev,
                        parent_rev,
                        &old_path,
                        &lossy_decoded_old_path,
                        &blame,
                        0,
                    )
                    .await?;
                }

                continue;
            }

            // Compute (a superset of) what lines of the file were change, as a range
            let blame_range = hunks
                .iter()
                .map(|LineRange(old_range)| old_range.start)
                .min()
                .expect("empty hunks")
                ..hunks
                    .iter()
                    .map(|LineRange(old_range)| old_range.end)
                    .max()
                    .expect("empty hunks");

            // Find what revision last changed each line in that range
            let mut blame = self
                .blamer
                .blame(
                    parent_rev,
                    old_path.clone(), // FIXME: don't clone
                    Some(blame_range.clone()),
                )
                .await
                .map_err(|source| SzzError::Annotation {
                    path: lossy_decoded_old_path.clone(),
                    revision: parent_rev_swhid,
                    source,
                })?;

            // Compute the exact set of lines changed/removed by the patch
            let ranges_to_resolve = provenance_ranges_from_hunks(&blame, &hunks)?;
            blame
                .set_ranges(ranges_to_resolve)
                .context("Could not initialize blamer ranges")?;

            let mut resolved_provenances = vec![];
            let mut levenshtein_cache = utils::CachedLevenshtein::default();

            // As long as some lines are not resolved to the revision introducing them...
            let mut blame_steps = 0;
            while blame.ranges().iter().next().is_some() {
                if blame_steps >= self.max_blame_steps_per_hunk.unwrap_or(u64::MAX) {
                    return Err(SzzError::MaxDepth);
                }
                blame_steps += 1;

                // Make the Blame recurse a step more
                blame.step().await.map_err(|source| SzzError::Annotation {
                    path: make_path(&graph, &old_path),
                    revision: parent_rev_swhid,
                    source,
                })?;

                // For each line whose originating revision is found, add them to
                // introduction_files_by_rev, and queue the others to be handled in the next loop
                // iteration.
                let mut new_ranges = Vec::<(Range<u32>, LineProvenance)>::new();

                for (range, provenance) in blame.ranges().iter() {
                    if provenance.added_in_this_rev {
                        // the blamer already fetched this content, so this shouldn't fail
                        //
                        // FIXME: Don't *also* fetch this on every iteration of the loop
                        let file_contents = self
                            .get_file_contents_from_rev(provenance.rev, &old_path)
                            .await
                            .context("Could not resolve content we just accessed")?
                            .context("Content we just accessed is missing")?;

                        // get contents in parent revisions
                        let mut parent_file_contents = vec![];
                        for parent_rev in self.graph.successors(provenance.rev) {
                            if graph.properties().node_type(parent_rev) == NodeType::Revision {
                                // the blamer already fetched this content, so this shouldn't fail
                                //
                                // FIXME: Don't *also* fetch this on every iteration of the outer loop
                                if let Ok(Some(parent_file_content)) =
                                    self.get_file_contents_from_rev(parent_rev, &old_path).await
                                {
                                    parent_file_contents.push((parent_rev, parent_file_content))
                                }
                            }
                        }

                        // The blamer says every line in the range originated from provenance.rev;
                        // which means that no parent of provenance.rev has the same lines
                        // (up to tokenization).
                        let mut lines_added_in_this_rev = vec![];
                        tokio::task::block_in_place(|| -> Result<_> {
                            #[cfg(feature = "arrow")]
                            let mut middle_rev_writer =
                                self.middle_rev_writer.as_ref().map(|writer| {
                                    writer
                                        .get_thread_writer()
                                        .expect("Could not get middle_rev_writer")
                                });
                            #[cfg(not(feature = "arrow"))]
                            let middle_rev_writer = &mut None::<&mut ()>;
                            for line in range {
                                let alternative_provenance = self.find_alternative_line_provenance(
                                    line,
                                    provenance,
                                    &file_contents,
                                    &parent_file_contents,
                                    &mut levenshtein_cache,
                                    fix_rev,
                                    middle_rev_writer.as_deref_mut(),
                                ).with_context(|| format!(
                                    "Could not lookup alternative provenance for line {line} (offsetted by {}) of {}",
                                    provenance.offset,
                                    graph.properties().swhid(provenance.rev)
                                ))?;
                                match alternative_provenance {
                                    // found a similar line in a parent of provenance.rev; so we will
                                    // continue blaming it
                                    Some(alt_provenance) => {
                                        assert!(
                                            provenance.added_in_this_rev
                                                || provenance.rev != alt_provenance.rev,
                                            "Stripped added_in_this_rev flag, but kept revision the same"
                                        );
                                        new_ranges.push((line..line + 1, alt_provenance))
                                    }
                                    // didn't, so we found the final provenance of the line
                                    None => lines_added_in_this_rev.push(line),
                                }
                            }

                            Ok(())
                        })?;
                        for range in utils::RangesFromItemsIter::new(lines_added_in_this_rev) {
                            let IntroductionFiles(introduction_files) =
                                introduction_files_by_rev.entry(provenance.rev).or_default();
                            let IntroductionFile { line_ranges } =
                                introduction_files.entry(old_path.clone()).or_default();
                            let introduction_range =
                                get_introduction_range(provenance.offset, &range)?;
                            line_ranges.insert(introduction_range.0);
                            resolved_provenances.push((range.clone(), provenance.clone()));
                        }
                    } else {
                        // the line was added by a parent of provenance.rev,
                        // and we need to run blame.step() again to get it.

                        new_ranges.push((range, provenance.clone()))
                    }
                }

                // trim ranges that we are not interested in blaming anymore
                let new_ranges = RangeMap::new(new_ranges)
                    .context("Filtered RangeMap is not a valid RangeMap")?;
                blame
                    .set_ranges_with_new_bounds(new_ranges)
                    .context("Could not replace blame ranges")?;
                // XXX TODO: set blame.revisions_to_explore accordingly
            }

            if let Some(writer) = line_details_writer {
                // Restore the blame state to what it would be if we did not mess with it above
                resolved_provenances.par_sort_by_key(|(range, _prov)| range.start);
                let ranges = RangeMap::new(resolved_provenances)
                    .context("Resolved provenances are an invalid RangeMap")?;
                blame
                    .set_ranges_with_new_bounds(ranges)
                    .context("Could not finalize blame ranges")?;

                self.write_line_details(
                    &mut **writer,
                    differ,
                    fix_rev,
                    parent_rev,
                    &old_path,
                    &lossy_decoded_old_path,
                    &blame,
                    blame_steps,
                )
                .await?;
            }
        }

        // Can we do something smarter than returning all revisions? (eg. only the ones with the most lines)
        Ok(introduction_files_by_rev)
    }

    /// Process a single git range from a vulnerability.
    ///
    /// This method should be called from within the closure passed to db::for_each_git_range.
    pub async fn process_range<W>(
        &self,
        vuln_writer: &mut Option<W>,
        line_details_writer: &mut Option<W>,
        filename: &str,
        range: &osv::schema::Range,
    ) -> Result<()>
    where
        W: std::ops::DerefMut,
        W::Target: Write,
    {
        let graph = Arc::clone(&self.graph);
        let content_retriever = Arc::clone(&self.content_retriever);
        let concurrent_downloads = self.concurrent_downloads;
        let stats = &self.stats;
        let fix_rev_to_computed_introduction_revs = &self.fix_rev_to_computed_introduction_revs;

        let differ = RevDiffer::new(Arc::clone(&graph), Arc::clone(&content_retriever))
            .with_tokenizer(self.blamer.tokenizer().clone())
            .with_concurrent_downloads(concurrent_downloads);

        let mut strategy = self
            .strategy_factory
            .get_instance(filename, stats)
            .context("Could not instantiate strategy")?;

        let Some(fix_revs) = strategy
            .get_fix_revs(range)
            .context("Could not get fix revs")?
        else {
            log::debug!("No fix rev found");
            return Ok(());
        };
        if fix_revs.len() != 1 {
            stats
                .ignored_because_more_than_one_fix
                .fetch_add(1, Ordering::Relaxed);
            return Ok(());
        }
        let fix_rev = fix_revs.iter().next().unwrap(); // TODO: add support for multiple revs

        let Some((introduced_at_zero, known_introduction_revs)) = strategy
            .get_introduction_revs(range)
            .context("Could not get introduction revs")?
        else {
            log::debug!("No introduction rev found");
            return Ok(());
        };

        // not using fix_rev_to_computed_introduction_revs.entry() to avoid holding a DashMap shard
        // across an await point.
        // it may cause duplicate work, but avoids locking other threads.
        let computed_introduction_revs =
            if let Some(revs) = fix_rev_to_computed_introduction_revs.get(&fix_rev) {
                revs.clone()
            } else {
                let revs = match self
                    .szz(&differ, &mut strategy, fix_rev, line_details_writer)
                    .await
                {
                    Ok(revs) => revs,
                    Err(SzzError::MaxDepth) => {
                        log::warn!(
                            "Reached maximum history depth for {}, giving up",
                            graph.properties().swhid(fix_rev)
                        );
                        return Ok(());
                    }
                    Err(SzzError::TooManyChanges) => {
                        log::warn!(
                            "{} makes too many changes, skipping",
                            graph.properties().swhid(fix_rev)
                        );
                        return Ok(());
                    }
                    Err(e @ (SzzError::TreeDiff { .. } | SzzError::Annotation { .. })) => {
                        log::debug!(
                            "Could not run SZZ on {}: {e:#}",
                            graph.properties().swhid(fix_rev)
                        );
                        stats
                            .ignored_because_unavailable_content
                            .fetch_add(1, Ordering::Relaxed);
                        return Ok(());
                    }
                    Err(SzzError::NotImplemented(e)) => {
                        log::debug!("{e:#}");
                        return Ok(());
                    }
                    Err(SzzError::Other(e)) => {
                        log::error!("Unexpected SZZ error: {e:#?}");
                        return Err(e);
                    }
                };
                fix_rev_to_computed_introduction_revs
                    .entry(fix_rev)
                    .or_insert(revs.clone());
                revs
            };

        stats
            .num_successful_szz_runs
            .fetch_add(1, Ordering::Relaxed);
        stats.total_known_introduction_revs.fetch_add(
            known_introduction_revs
                .len()
                .try_into()
                .expect("known_introduction_revs.len() overflows usize"),
            Ordering::Relaxed,
        );
        stats.total_computed_introduction_revs.fetch_add(
            computed_introduction_revs
                .len()
                .try_into()
                .expect("computed_introduction_revs.len() overflows usize"),
            Ordering::Relaxed,
        );

        if let Some(writer) = vuln_writer {
            let record = SzzOutputRecord {
                vuln_filename: filename.to_string(),
                fix_revs: fix_revs
                    .iter()
                    .map(|node| graph.properties().swhid(node))
                    .collect(),
                fix_revs_tag_names: git::get_all_tag_names(&graph, &fix_revs),
                introduced_at_zero,
                known_introduction_revs: known_introduction_revs
                    .iter()
                    .map(|node| graph.properties().swhid(node))
                    .collect(),
                known_introduction_revs_tag_names: git::get_all_tag_names(
                    &graph,
                    &known_introduction_revs,
                ),
                computed_introduction_revs: computed_introduction_revs
                    .iter()
                    .map(|(&node, IntroductionFiles(introduction_files))| -> (SWHID, IntroductionFilesRecord) {
                        (
                            graph.properties().swhid(node),
                            IntroductionFilesRecord(introduction_files.iter().map(|(path_parts, lines)| {
                                (make_path(&graph, path_parts), lines.clone())
                            }).collect()),
                        )
                    })
                    .collect(),
                computed_introduction_revs_tag_names: git::get_all_tag_names(
                    &graph,
                    computed_introduction_revs.into_keys(),
                ),
            };
            let writer = &mut **writer;
            serde_json::to_writer(&mut *writer, &record)
                .with_context(|| format!("Could not write {record:?}"))?;
            writer.write_all(b"\n").context("Could not end line")?;
            writer.flush().context("Could not flush line")?;
        }
        if let Some(writer) = line_details_writer {
            let writer = &mut **writer;
            writer
                .flush()
                .context("Could not flush line details writer")?;
        }

        Ok(())
    }

    /// Returns the collected statistics.
    pub fn stats(self) -> SzzStats {
        self.stats
    }

    pub fn flush(&mut self) -> Result<()> {
        #[cfg(feature = "arrow")]
        if let Some(writer) = self.middle_rev_writer.as_mut() {
            writer
                .flush()
                .context("Could not flush similar lines writer")?;
        }

        Ok(())
    }

    fn check_limits(&self, changes: &[&FileDiff]) -> Result<(), SzzError> {
        if let Some(max_hunks) = self.max_hunks {
            if changes
                .iter()
                .map(|change| match change {
                    FileDiff::Added { .. } => 0,
                    FileDiff::Removed { .. } => 0,
                    FileDiff::Modified {
                        hunks: Some(hunks), ..
                    } => hunks.len() as u64,
                    FileDiff::Modified { hunks: None, .. } => 0,
                })
                .sum::<u64>()
                >= max_hunks
            {
                return Err(SzzError::TooManyChanges);
            }
        }
        if let Some(max_modified_files) = self.max_modified_files {
            if changes
                .iter()
                .filter(|change| matches!(change, FileDiff::Modified { .. }))
                .count() as u64
                >= max_modified_files
            {
                return Err(SzzError::TooManyChanges);
            }
        }

        Ok(())
    }

    async fn get_file_contents_from_rev(
        &self,
        rev: NodeId,
        path: &[LabelNameId],
    ) -> Result<Option<FileContents>> {
        let root_dir = find_root_dir(&self.graph, rev)
            .context("Could not find root dir")?
            .context("Revision has no root dir")?;
        let Some(content) = resolve_path_by_id(&self.graph, root_dir, path)
            .context("Could not resolve path we just accessed")?
        else {
            return Ok(None);
        };
        let file_at_rev = FileAtRev {
            rev,
            content,
            path,
            perm: None,
        };
        FileContents::fetch(&self.graph, &self.content_retriever, &file_at_rev)
            .await
            .context("Could not fetch content we just accessed")
            .map(Some)
    }

    #[allow(clippy::too_many_arguments)]
    /// Called when the blamer found the first occurrence of the line.
    /// Returns an earlier occurrence of a similar line.
    fn find_alternative_line_provenance(
        &self,
        lineno: u32, // in the original revision, so without provenance.offset applied
        provenance: &LineProvenance,
        file_contents: &FileContents,
        parent_file_contents: &[(NodeId, FileContents)], // (rev, content in rev)
        levenshtein_cache: &mut utils::CachedLevenshtein,
        fix_rev: NodeId,
        middle_rev_writer: Option<&mut MiddleRevWriter>,
    ) -> Result<Option<LineProvenance>> {
        let Some(min_line_similarity_ratio) = self.min_line_similarity_ratio else {
            // V-SZZ extension disabled
            return Ok(None);
        };

        let lineno = lineno
            .checked_add_signed(provenance.offset)
            .context("Line number overflowed")?;

        let Some(lines) = file_contents.lines() else {
            // Binary file
            return Ok(None);
        };
        let line: &[u8] = lines.get(lineno).with_context(|| {
            format!(
                "Line {lineno} does not exist in {} (at {}) of {})",
                file_contents.swhid,
                make_path(&self.graph, &file_contents.path),
                self.graph.properties().swhid(provenance.rev),
            )
        })?;

        // TODO: check for exact matches in the parent to avoid computing any Levenshtein in that
        // case

        let mut best_match = None;
        let mut best_distance = ((line.len() as f64) * (1. - min_line_similarity_ratio)) as u32;

        if best_distance > self.max_line_similarity_distance {
            if log::log_enabled!(log::Level::Debug) {
                log::debug!(
                    "Looking for lines similar to line {lineno} of {} at {} of {}, but it is {} bytes long, causing the acceptable distance to be {best_distance}. This causes unreasonably bad performance due to Levenshtein's pseudo-quadratic time. Clamping it to {}.",
                    file_contents.swhid,
                    make_path(&self.graph, &file_contents.path),
                    self.graph.properties().swhid(provenance.rev),
                    line.len(),
                    self.max_line_similarity_distance
                );
            }
            best_distance = self.max_line_similarity_distance;
        }

        let start_time = SystemTime::now();
        for (parent_rev, parent_file_content) in parent_file_contents {
            let Some(parent_file_lines) = parent_file_content.lines() else {
                // Binary file
                continue;
            };
            let parent_file_lines = parent_file_lines
                .iter()
                .collect::<Vec<_>>()
                .into_iter()
                .enumerate();

            let parent_lines_before = parent_file_lines
                .clone()
                .take(lineno as usize)
                .rev()
                .take(self.max_similar_line_offset);
            let parent_lines_after = parent_file_lines
                .skip(lineno as usize)
                .take(self.max_similar_line_offset);

            // reorder lines so that we visit the closest ones (by position) first. This has two benefits:
            // 1. if two lines have the same Levenshtein distance, ties are broken by first
            //    visited, which is good because they are probably a better match
            // 2. close lines (by position) are likely to be the most similar ones (by Levenshtein
            //    distance), which means we can abort computing Levenshtein distances of less
            //    similar lines sooner, which is faster
            let parent_file_lines = parent_lines_before.interleave(parent_lines_after);

            for (other_lineno, other_line) in parent_file_lines {
                if let Some(distance) =
                    levenshtein_cache.distance(line, other_line, best_distance.saturating_sub(1))
                {
                    best_distance = distance;
                    best_match = Some((other_lineno, parent_rev, other_line));
                    if best_distance == 0 {
                        break;
                    }
                }
            }
            if best_distance == 0 {
                break;
            }
        }

        match start_time.elapsed() {
            Ok(total_time) => {
                if total_time > Duration::from_secs(120) {
                    log::warn!(
                        "Finding lines similar to line {lineno} of {} at {} of {} took {total_time:?}. It is {} bytes long.",
                        file_contents.swhid,
                        make_path(&self.graph, &file_contents.path),
                        self.graph.properties().swhid(provenance.rev),
                        line.len(),
                    );
                }
            }
            Err(_) => log::warn!("Time ran backward from {start_time:?} to now"),
        }

        if let Some((best_lineno, &best_parent_rev, best_parent_line)) = best_match {
            log::debug!(
                "{:?} is similar to {:?} in its parent {}",
                String::from_utf8_lossy(line),
                String::from_utf8_lossy(best_parent_line),
                self.graph.properties().swhid(best_parent_rev)
            );
            #[cfg(not(feature = "arrow"))]
            let _ = (fix_rev, middle_rev_writer, &self.written_middle_revs); // for Clippy
            #[cfg(feature = "arrow")]
            if let Some(writer) = middle_rev_writer {
                let (_parent_rev, parent_file_content) = parent_file_contents
                    .iter()
                    .find(|(parent_rev, _)| *parent_rev == best_parent_rev)
                    .expect("best_parent_rev disappeared from parent_file_contents");
                if self.written_middle_revs.insert((
                    best_parent_rev,
                    provenance.rev,
                    fix_rev,
                    file_contents.path.clone(),
                )) {
                    // only write it if it was not already

                    writer
                        .serialize(MiddleCommitRecord {
                            rev: self.graph.properties().swhid(best_parent_rev),
                            rev_author_timestamp: self
                                .graph
                                .properties()
                                .author_timestamp(best_parent_rev),
                            rev_committer_timestamp: self
                                .graph
                                .properties()
                                .committer_timestamp(best_parent_rev),
                            predecessor_rev: self.graph.properties().swhid(provenance.rev),
                            predecessor_rev_author_timestamp: self
                                .graph
                                .properties()
                                .author_timestamp(provenance.rev),
                            predecessor_rev_committer_timestamp: self
                                .graph
                                .properties()
                                .committer_timestamp(provenance.rev),
                            hunk_id: "".to_string(),
                            ancestor_rev: self.graph.properties().swhid(fix_rev),
                            ancestor_rev_author_timestamp: self
                                .graph
                                .properties()
                                .author_timestamp(fix_rev),
                            ancestor_rev_committer_timestamp: self
                                .graph
                                .properties()
                                .committer_timestamp(fix_rev),
                            tag: Cow::Borrowed("V-SZZ_middle_revision"),
                            file_path: make_path(&self.graph, &file_contents.path),
                            code_before_id: parent_file_content.swhid,
                            code_after_id: file_contents.swhid,
                            code_before: String::from_utf8_lossy(&parent_file_content.contents),
                            code_after: String::from_utf8_lossy(&file_contents.contents),
                        })
                        .context("Could not write MiddleCommitRecord")?;
                }
            }
            Ok(Some(LineProvenance {
                rev: best_parent_rev,
                offset: i32::try_from(
                    i64::from(provenance.offset)
                        + (u64::try_from(best_lineno).expect("Line number overflowed u64") as i64)
                        - (u64::from(lineno) as i64),
                )
                .context("offset overflow")?,
                added_in_this_rev: false, // as far as we know. The blamer may update it later
            }))
        } else {
            Ok(None)
        }
    }
}

fn provenance_ranges_from_hunks<
    G: SwhLabeledForwardGraph
        + SwhGraphWithProperties<Maps: properties::Maps, LabelNames: properties::LabelNames>,
    CR: std::ops::Deref<Target: ContentRetriever> + Clone,
    T: Tokenizer + Clone,
    RS: RevisionSkipper,
>(
    blame: &Blame<Arc<G>, CR, T, RS>,
    hunks: &[LineRange],
) -> Result<RangeMap<LineProvenance>> {
    let (first_range, first_prov) = blame
        .ranges()
        .iter()
        .next()
        .expect("Blame has empty set of ranges");
    let first_prov = first_prov.clone();
    assert!(
        first_range.contains(&first_range.start),
        "First range does not contain first line"
    );
    let parent_rev = first_prov.rev;
    assert_eq!(
        first_prov,
        LineProvenance {
            rev: parent_rev,
            offset: 0,
            added_in_this_rev: false
        },
        "Unexpected provenance initialization: {first_prov:?}"
    );

    RangeMap::new(
        hunks
            .iter()
            .map(|LineRange(old_range)| (old_range.clone(), first_prov.clone())),
    )
    .context("Could not build RangeMap from diff hunks")
}

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

    #[test]
    fn map_new_range_to_old_no_hunks() {
        assert_eq!(map_new_range_to_old(&(3..7), &[]).unwrap(), 3..7);
    }

    #[test]
    fn map_new_range_to_old_unchanged_before_hunk() {
        let hunks = [DiffHunk {
            old_range: 10..12,
            new_range: 10..13,
        }];
        assert_eq!(map_new_range_to_old(&(2..5), &hunks).unwrap(), 2..5);
    }

    #[test]
    fn map_new_range_to_old_unchanged_between_hunks() {
        let hunks = [
            DiffHunk {
                old_range: 2..3,
                new_range: 2..4, // +1 line
            },
            DiffHunk {
                old_range: 10..11,
                new_range: 11..12,
            },
        ];
        assert_eq!(map_new_range_to_old(&(5..8), &hunks).unwrap(), 4..7);
    }

    #[test]
    fn map_new_range_to_old_unchanged_after_last_hunk() {
        let hunks = [DiffHunk {
            old_range: 2..4,
            new_range: 2..3, // -1 line
        }];
        assert_eq!(map_new_range_to_old(&(5..8), &hunks).unwrap(), 6..9);
    }

    #[test]
    fn map_new_range_to_old_equal_to_hunk() {
        // 1:1 replacement
        let hunks = [DiffHunk {
            old_range: 5..8,
            new_range: 5..8,
        }];
        assert_eq!(map_new_range_to_old(&(5..8), &hunks).unwrap(), 5..8);

        let hunks = [DiffHunk {
            old_range: 5..10,
            new_range: 5..7, // -3 lines
        }];
        assert_eq!(map_new_range_to_old(&(5..7), &hunks).unwrap(), 5..10);

        let hunks = [DiffHunk {
            old_range: 5..6,
            new_range: 5..9, // + 3 lines
        }];
        assert_eq!(map_new_range_to_old(&(5..9), &hunks).unwrap(), 5..6);
    }

    #[test]
    fn map_new_range_to_old_before_and_overlap() {
        let hunks = [DiffHunk {
            old_range: 5..8,
            new_range: 5..8,
        }];
        assert_eq!(map_new_range_to_old(&(3..7), &hunks).unwrap(), 3..8);

        let hunks = [DiffHunk {
            old_range: 5..7,
            new_range: 5..8,
        }];
        assert_eq!(map_new_range_to_old(&(3..7), &hunks).unwrap(), 3..7);
    }

    #[test]
    fn map_new_range_to_old_after_and_overlap() {
        let hunks = [DiffHunk {
            old_range: 5..8,
            new_range: 5..8,
        }];
        assert_eq!(map_new_range_to_old(&(6..10), &hunks).unwrap(), 5..10);

        let hunks = [DiffHunk {
            old_range: 5..7,
            new_range: 5..8,
        }];
        assert_eq!(map_new_range_to_old(&(6..10), &hunks).unwrap(), 5..9);
    }

    #[test]
    fn map_new_range_to_old_contains_pure_deletion() {
        let hunks = [DiffHunk {
            old_range: 5..8,
            new_range: 5..5,
        }];
        assert_eq!(map_new_range_to_old(&(3..6), &hunks).unwrap(), 3..9);
    }

    #[test]
    fn map_new_range_to_old_pure_deletion_at_range_boundary() {
        let hunks = [DiffHunk {
            old_range: 5..8,
            new_range: 5..5,
        }];
        assert_eq!(map_new_range_to_old(&(3..5), &hunks).unwrap(), 3..5);
    }

    #[test]
    fn map_new_range_to_old_pure_insertion() {
        let hunks = [DiffHunk {
            old_range: 5..5,
            new_range: 5..8,
        }];
        assert_eq!(map_new_range_to_old(&(5..8), &hunks).unwrap(), 5..5);
    }
}
