// Copyright (C) 2025  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::sync::Arc;

use anyhow::{Result, bail};
use dashmap::DashSet;
use dsi_progress_logger::{ProgressLog, concurrent_progress_logger};
use rayon::prelude::*;
use swh_graph::NodeType;
use swh_graph::graph::{NodeId, SwhForwardGraph, SwhGraphWithProperties};
use swh_graph::properties;
use swh_graph_stdlib::collections::{AdaptiveNodeSet, NodeSet, ReadNodeSet};
use swh_graph_stdlib::connectivity::SubgraphWccs;
use value_traits::slices::SliceByValue;

pub fn for_each_root<G: SwhForwardGraph, E>(
    graph: &G,
    src: Vec<NodeId>,
    mut f: impl FnMut(NodeId) -> Result<(), E>,
) -> Result<(), E> {
    let mut todo: Vec<_> = src;
    let mut seen = AdaptiveNodeSet::new(graph.num_nodes());
    while let Some(node) = todo.pop() {
        let mut is_root = true;
        for succ in graph.successors(node) {
            is_root = false;
            if !seen.contains(succ) {
                todo.push(succ);
                seen.insert(succ);
            }
        }
        if is_root {
            f(node)?;
        }
    }
    Ok(())
}

/// Lists all content nodes reachable from revisions in a SubgraphWccs.
///
/// For each revision in the SubgraphWccs, performs a depth-first traversal to find
/// all reachable content nodes.
pub fn list_reachable_contents_from_subgraph<
    G: SwhForwardGraph + SwhGraphWithProperties<Maps: properties::Maps> + Sync + Send,
    V: SliceByValue<Value = usize> + Sync,
>(
    graph: Arc<G>,
    subgraphwccs: &SubgraphWccs<impl AsRef<[usize]> + Sync, V>,
) -> Result<DashSet<NodeId, rapidhash::RapidBuildHasher>> {
    let num_revs = subgraphwccs.num_nodes();

    let t = std::thread::spawn(move || {
        DashSet::with_capacity_and_hasher(num_revs * 10, rapidhash::RapidBuildHasher::default())
    });
    let seen_directories =
        DashSet::with_capacity_and_hasher(num_revs * 10, rapidhash::RapidBuildHasher::default());
    let seen_contents = t.join().expect("Could not join thread");

    let mut pl = concurrent_progress_logger!(
        display_memory = true,
        item_name = "revision",
        local_speed = true,
        expected_updates = Some(num_revs),
    );
    pl.threshold(1000);
    pl.start("Listing reachable contents...");

    (0..num_revs)
        .into_par_iter()
        .try_for_each_with(pl.clone(), |pl, rev_subgraphnodeid| {
            let rev = subgraphwccs
                .contraction()
                .underlying_node_id(rev_subgraphnodeid);

            let mut todo = vec![rev];
            while let Some(node) = todo.pop() {
                for succ in graph.successors(node) {
                    match graph.properties().node_type(succ) {
                        NodeType::Content => {
                            seen_contents.insert(succ);
                        }
                        NodeType::Directory => {
                            let new = seen_directories.insert(succ);
                            if new {
                                todo.push(succ);
                            }
                        }
                        NodeType::Revision => {
                            // ignore, will be processed later (if successor of revision)
                            // or is irrelevant (if git submodule)
                        }
                        _ => bail!(
                            "{} has unexpected successor type: {}",
                            graph.properties().swhid(node),
                            graph.properties().swhid(succ)
                        ),
                    }
                }
            }

            pl.light_update();
            Ok(())
        })?;

    pl.done();

    Ok(seen_contents)
}
