// Copyright 2026 PingCAP, Inc.

use std::{path::PathBuf, sync::Arc};

use bytes::Bytes;
use clap::{ArgEnum, Args};
use kvengine::{
    dfs::{DFSConfig, FileType},
    table::{
        file::{File, InMemFile},
        fts::{
            CommonPk, DedicatedFile, EDedicatedFile, EPackedFileLp, FtsCache, IntPk,
            OrderedPkIterator, PackedFile, PkReader, PkType,
        },
    },
};
use tokio::runtime::{Builder, Runtime};

use crate::{common::get_file_data_from_dfs, stats::format_size};

#[derive(Args)]
pub struct ShowFtsArgs {
    /// The path of the config file.
    #[clap(long, default_value = "")]
    pub config: PathBuf,
    /// The id of the FTS file.
    #[clap(long, default_value_t = 0)]
    pub id: u64,
    /// The path of local FTS file.
    ///
    /// If specified, the FTS file will be retrieved from the path instead of
    /// DFS.
    #[clap(long)]
    pub local: Option<PathBuf>,
    /// The FTS file type.
    #[clap(long, arg_enum, default_value = "packed")]
    pub file_type: FtsFileKind,
    /// Dump all handles and versions for debugging.
    #[clap(long)]
    pub dump_handles: bool,
}

#[derive(Copy, Clone, Debug, ArgEnum)]
pub enum FtsFileKind {
    Packed,
    Dedicated,
}

#[derive(Clone, Serialize, Deserialize, PartialEq, Debug, Default)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct ShowFtsConfig {
    pub dfs: DFSConfig,
}

pub fn execute_show_fts(args: ShowFtsArgs) {
    let mut config = ShowFtsConfig::default();
    if args.config.exists() {
        let data = std::fs::read(args.config.clone()).expect("failed to read config file");
        config = toml::from_slice(&data).unwrap();
    }
    config.dfs.override_from_env();

    let data = match args.local {
        Some(local) => crate::sst::get_file_data_from_local(&local),
        None => get_file_data_from_dfs(args.id, file_type(args.file_type), &config.dfs),
    };
    let file: Arc<dyn File> = Arc::new(InMemFile::new(args.id, data));

    let runtime = Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("failed to build runtime");

    match args.file_type {
        FtsFileKind::Packed => show_packed_file(file, &runtime, args.dump_handles),
        FtsFileKind::Dedicated => show_dedicated_file(file, &runtime, args.dump_handles),
    }
}

fn show_packed_file(file: Arc<dyn File>, runtime: &Runtime, dump_handles: bool) {
    let packed = PackedFile::new(file, FtsCache::disabled()).expect("failed to open packed file");
    print_packed_summary(&packed);
    runtime.block_on(print_packed_lps(&packed, dump_handles));
}

fn print_packed_summary(packed: &PackedFile) {
    let props = packed.props();
    let info = packed.build_info();
    println!("[FTS packed {}]", packed.id());
    println!("  size: {}", format_size(packed.file().size()));
    println!("  meta_offset: {}", info.meta_offset);
    println!("  snap_version: {}", info.snap_version);
    println!("  total_lps: {}", props.get_total_lps());
    println!(
        "  total_index_bytes: {}",
        format_size(props.get_total_index_bytes() as u64)
    );
    println!(
        "  total_keys_bytes: {}",
        format_size(props.get_total_keys_bytes() as u64)
    );
    println!("  pk_total: {}", props.get_pk_total());
    println!("  pk_live_unique: {}", props.get_pk_live_unique());
    println!(
        "  smallest_lp_key: {}",
        log_wrappers::hex_encode_upper(props.get_smallest_lp_key())
    );
    println!(
        "  largest_lp_key: {}",
        log_wrappers::hex_encode_upper(props.get_largest_lp_key())
    );
    println!(
        "  smallest_key: {}",
        log_wrappers::hex_encode_upper(props.get_smallest_key())
    );
    println!(
        "  biggest_key: {}",
        log_wrappers::hex_encode_upper(props.get_biggest_key())
    );
    println!(
        "  table_ids: {}",
        if props.table_ids.is_empty() {
            "[]".to_owned()
        } else {
            props
                .table_ids
                .iter()
                .map(|t| t.to_string())
                .collect::<Vec<_>>()
                .join(", ")
        }
    );
    let table_range = format_table_index_range(props);
    println!("  table_index_range: {}", table_range);
    let n_data_blocks = packed.n_data_blocks().expect("failed to load index block");
    println!("  data_blocks: {}", n_data_blocks);
}

async fn print_packed_lps(packed: &PackedFile, dump_handles: bool) {
    let mut lp_iter = packed.lp_iter().expect("failed to create lp iterator");
    let mut idx = 0;
    while let Some(lp) = lp_iter.next_lp().await.expect("failed to load lp") {
        print_lp_summary(idx, &lp);
        if dump_handles {
            match lp {
                EPackedFileLp::Int(ref lp) => dump_handles_for_reader::<IntPk, _>(lp, 6).await,
                EPackedFileLp::Common(ref lp) => {
                    dump_handles_for_reader::<CommonPk, _>(lp, 6).await
                }
            }
        }
        idx += 1;
    }
}

fn print_lp_summary(idx: usize, lp: &EPackedFileLp) {
    let props = lp.props();
    println!("  [LP {}]", idx);
    println!(
        "    lp_key: {}",
        log_wrappers::hex_encode_upper(lp.lp_key())
    );
    println!("    table_id: {}", props.get_table_id());
    println!("    index_id: {}", props.get_index_id());
    println!("    is_int_handle: {}", props.get_is_int_handle());
    println!("    n_pk: {}", props.get_n_pk());
}

fn show_dedicated_file(file: Arc<dyn File>, runtime: &Runtime, dump_handles: bool) {
    let file =
        EDedicatedFile::new(file, FtsCache::disabled()).expect("failed to open dedicated file");
    print_dedicated_summary(&file);
    if dump_handles {
        match file {
            EDedicatedFile::Int(ref file) => runtime
                .block_on(dump_handles_for_reader::<IntPk, DedicatedFile<IntPk>>(
                    file, 2,
                )),
            EDedicatedFile::Common(ref file) => runtime
                .block_on(dump_handles_for_reader::<CommonPk, DedicatedFile<CommonPk>>(file, 2)),
        }
    }
}

fn print_dedicated_summary(file: &EDedicatedFile) {
    let props = file.props();
    println!("[FTS dedicated {}]", file.id());
    println!("  size: {}", format_size(file.file().size()));
    let footer = match file {
        EDedicatedFile::Int(f) => f.footer(),
        EDedicatedFile::Common(f) => f.footer(),
    };
    println!("  meta_offset: {}", footer.metadata_offset());
    println!("  data_block_offset: {}", footer.data_block_offset);
    println!("  handle_index_offset: {}", footer.iblock_offset);
    println!(
        "  pk_filter_block_offset: {}",
        footer.pk_filter_block_offset
    );
    println!("  prop_offset: {}", footer.prop_offset);
    println!("  is_int_handle: {}", file.is_int_handle());
    println!("  table_id: {}", props.get_table_id());
    println!("  index_id: {}", props.get_index_id());
    println!("  pk_total: {}", props.get_pk_total());
    println!("  pk_live_unique: {}", props.get_pk_live_unique());
    println!(
        "  lp_key: {}",
        log_wrappers::hex_encode_upper(props.get_lp_key())
    );
    println!(
        "  smallest_key: {}",
        log_wrappers::hex_encode_upper(props.get_smallest_key())
    );
    println!(
        "  biggest_key: {}",
        log_wrappers::hex_encode_upper(props.get_biggest_key())
    );
}

async fn dump_handles_for_reader<Pk, R>(reader: &R, indent: usize)
where
    Pk: PkType,
    R: PkReader<Pk>,
{
    let mut iter = reader.pk_iter().expect("failed to create pk iterator");
    let indent = " ".repeat(indent);
    while let Some((doc_id, pk, version, is_deleted)) =
        iter.next().await.expect("failed to read pk")
    {
        let handle = format_handle::<Pk>(pk);
        println!(
            "{}{}: handle={}, version={}, deleted={}",
            indent,
            doc_id,
            handle,
            version,
            is_deleted != 0
        );
    }
}

fn format_table_index_range(props: &kvenginepb::fts::PackedFilePropBlock) -> String {
    let smallest = props.get_smallest_table_index();
    let largest = props.get_largest_table_index();
    if !props.has_smallest_table_index() || !props.has_largest_table_index() {
        return "N/A".to_owned();
    }
    format!(
        "({},{})..=({},{})",
        smallest.get_table_id(),
        smallest.get_index_id(),
        largest.get_table_id(),
        largest.get_index_id()
    )
}

fn format_handle<Pk: PkType>(pk: Bytes) -> String {
    if Pk::IS_INT {
        match IntPk::decode(pk.as_ref()) {
            Ok(v) => format!(
                "{} (raw {})",
                v,
                log_wrappers::hex_encode_upper(pk.as_ref())
            ),
            Err(_) => format!("raw {}", log_wrappers::hex_encode_upper(pk.as_ref())),
        }
    } else {
        let utf8 = String::from_utf8_lossy(pk.as_ref());
        format!(
            "{} (raw {})",
            utf8,
            log_wrappers::hex_encode_upper(pk.as_ref())
        )
    }
}

fn file_type(kind: FtsFileKind) -> FileType {
    match kind {
        FtsFileKind::Packed => FileType::FtsPackedFile,
        FtsFileKind::Dedicated => FileType::FtsDedicatedFile,
    }
}