// 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::path::PathBuf;
use std::sync::Arc;

use anyhow::{Context, Result};
use clap::Parser;
use dataset_writer::{CsvZstTableWriter, ParallelDatasetWriter};
use epserde::deser::Deserialize;
use epserde::prelude::Flags;
use mimalloc::MiMalloc;
use rayon::prelude::*;
use serde::Serialize;
use swh_graph::SWHID;
use swh_graph::graph::{SwhGraphWithProperties, SwhUnidirectionalGraph};
use swh_graph::mph::DynMphf;
use swh_graph_stdlib::connectivity::SubgraphWccs;

#[global_allocator]
static GLOBAL: MiMalloc = MiMalloc;

#[derive(Parser, Debug)]
#[command()]
/// Writes to stdout all contents changed by a fix commit, and all their previous versions
struct Args {
    #[arg(long)]
    /// Path to swh-graph's files
    graph: PathBuf,
    #[arg(long)]
    /// Path to the epserde-serialized `SubgraphWccs` written by `compute-connected-components`
    subgraphwccs: PathBuf,
    #[arg(long)]
    /// Where to write the .csv.zst files with content SWHIDs
    dir_out: PathBuf,
}

pub fn main() -> Result<()> {
    let args = Args::parse();

    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::from_default_env()
                .add_directive(tracing::Level::INFO.into()),
        )
        .init();

    tracing::info!("Loading graph...");
    let graph = SwhUnidirectionalGraph::new(&args.graph)
        .with_context(|| format!("Could not load graph from {}", args.graph.display()))?
        .load_labels()
        .context("Could not load labels")?
        .init_properties()
        .load_properties(|properties| properties.load_maps::<DynMphf>())
        .context("Could not load maps")?;
    let graph = Arc::new(graph);

    tracing::info!("Loading SubgraphWccs...");
    let subgraphwccs = unsafe { <SubgraphWccs>::mmap(&args.subgraphwccs, Flags::RANDOM_ACCESS) }
        .with_context(|| {
            format!(
                "Could not mmap connected components from {}",
                args.subgraphwccs.display()
            )
        })?;
    let subgraphwccs = subgraphwccs.uncase();

    #[allow(non_snake_case)]
    #[derive(Debug, Serialize)]
    struct Row {
        SWHID: SWHID,
    }

    tracing::info!("Listing reachable contents...");
    let seen_contents = swh_osv::traversals::list_reachable_contents_from_subgraph(
        Arc::clone(&graph),
        subgraphwccs,
    )?;

    tracing::info!("Writing results...");
    let dataset_writer = ParallelDatasetWriter::<CsvZstTableWriter>::new(args.dir_out.clone())
        .with_context(|| format!("Could not create directory {}", args.dir_out.display()))?;

    seen_contents.par_iter().try_for_each_init(
        || dataset_writer.get_thread_writer().unwrap(),
        |writer, content_node| {
            writer
                .serialize(Row {
                    SWHID: graph.properties().swhid(*content_node),
                })
                .context("Could not write SWHID")
        },
    )?;

    Ok(())
}
