// Copyright 2026 PingCAP, Inc.

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

use clap::Args;
use kvengine::{
    dfs::{DFSConfig, Dfs, FileType, Options, S3Fs},
    table::{
        columnar::{Block, ColumnarConcatReader, ColumnarFile, ColumnarMetaCache, ColumnarReader},
        file::{File, InMemFile},
        schema_file::{Schema, SchemaFile},
    },
};
use tokio::runtime::Builder;

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

#[derive(Args)]
pub struct ShowColumnarArgs {
    /// The path of the config file.
    #[clap(long, default_value = "")]
    pub config: PathBuf,
    /// The id of the Columnar file.
    #[clap(long, default_value_t = 0)]
    pub id: u64,
    /// The path of local Columnar file.
    ///
    /// If specified, the Columnar file will be retrieved from the path instead
    /// of DFS.
    #[clap(long)]
    pub local: Option<PathBuf>,

    /// Optional schema file id (for using full schema; otherwise infer minimal
    /// schema from the file).
    #[clap(long)]
    pub schema_id: Option<u64>,
    /// Optional local schema file path.
    #[clap(long)]
    pub schema_local: Option<PathBuf>,

    /// Restrict to a single table id (default: dump all tables found in the
    /// file).
    #[clap(long)]
    pub table_id: Option<i64>,

    /// Dump all (handle, version, delete) entries for debugging.
    #[clap(long)]
    pub dump_handles: bool,

    /// Limit the number of dumped rows per table (only applies with
    /// --dump-handles).
    #[clap(long)]
    pub head: Option<usize>,
}

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

pub fn execute_show_columnar(args: ShowColumnarArgs) {
    let mut config = ShowColumnarConfig::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.as_ref() {
        Some(local) => crate::sst::get_file_data_from_local(local),
        None => get_file_data_from_dfs(args.id, FileType::Columnar, &config.dfs),
    };
    let file: Arc<dyn File> = Arc::new(InMemFile::new(args.id, data));

    let columnar = ColumnarFile::open(file, None, ColumnarMetaCache::default())
        .expect("failed to open columnar file");

    print_columnar_summary(&columnar);

    if !args.dump_handles {
        return;
    }

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

    let schema_file = load_schema_file_if_any(&args, &config);
    let mut table_ids = columnar.get_table_ids();
    table_ids.sort_unstable();
    if let Some(table_id) = args.table_id {
        table_ids.retain(|t| *t == table_id);
    }

    if table_ids.is_empty() {
        println!("No table found in this columnar file.");
        return;
    }

    for table_id in table_ids {
        let schema = schema_file
            .as_ref()
            .and_then(|sf| sf.export_schemas().get(&table_id).cloned())
            .unwrap_or_else(|| {
                columnar
                    .infer_table_schema(table_id)
                    .expect("failed to infer table schema from columnar file")
            });

        println!("[Table {}]", table_id);
        println!("  is_common_handle: {}", schema.is_common_handle());
        if let Some(last) = columnar.get_table_last_handle(table_id) {
            println!(
                "  last_handle_raw: {}",
                log_wrappers::hex_encode_upper(last.as_slice())
            );
        }
        let head = args.head;
        runtime.block_on(dump_table_handles(&columnar, schema, head));
    }
}

fn load_schema_file_if_any(
    args: &ShowColumnarArgs,
    config: &ShowColumnarConfig,
) -> Option<SchemaFile> {
    let data = match (&args.schema_id, &args.schema_local) {
        (None, None) => return None,
        (Some(_), Some(_)) => {
            panic!("please specify only one of --schema-id or --schema-local")
        }
        (Some(schema_id), None) => {
            let s3fs = S3Fs::new_from_config(config.dfs.clone());
            let runtime = s3fs.get_runtime();
            runtime
                .block_on(
                    s3fs.read_file(*schema_id, Options::default().with_type(FileType::Schema)),
                )
                .expect("failed to read schema file from dfs")
        }
        (None, Some(schema_local)) => crate::sst::get_file_data_from_local(schema_local),
    };
    let schema_id = args.schema_id.unwrap_or(0);
    let file = Arc::new(InMemFile::new(schema_id, data));
    Some(SchemaFile::open(file).expect("failed to open schema file"))
}

fn print_columnar_summary(file: &ColumnarFile) {
    println!("[Columnar {}]", file.id());
    println!("  size: {}", format_size(file.size()));
    println!("  meta_offset: {}", file.get_meta_offset());
    println!("  table_count: {}", file.table_count());
    println!("  max_version: {}", file.get_max_version());
    println!(
        "  snap_version: {}",
        file.get_snap_version()
            .map(|v| v.to_string())
            .unwrap_or_else(|| "N/A".to_owned())
    );
    println!("  encryption_ver: {}", file.get_encryption_ver());
    println!(
        "  estimated_kv_size: {}",
        format_size(file.get_estimated_kv_size() as u64)
    );
    println!(
        "  smallest_key: {}",
        log_wrappers::hex_encode_upper(file.get_smallest().deref())
    );
    println!(
        "  biggest_key: {}",
        log_wrappers::hex_encode_upper(file.get_biggest().deref())
    );
    let mut table_ids = file.get_table_ids();
    table_ids.sort_unstable();
    println!(
        "  table_ids: {}",
        if table_ids.is_empty() {
            "[]".to_owned()
        } else {
            table_ids
                .iter()
                .map(|t| t.to_string())
                .collect::<Vec<_>>()
                .join(", ")
        }
    );
}

async fn dump_table_handles(columnar: &ColumnarFile, schema: Schema, head: Option<usize>) {
    let mut reader =
        ColumnarConcatReader::new(std::slice::from_ref(columnar), schema.clone(), None, None);
    if schema.is_common_handle() {
        reader.seek(&[]).await.expect("seek failed");
    } else {
        reader.seek(&0i64.to_le_bytes()).await.expect("seek failed");
    }

    let mut block = Block::new(&schema);
    let mut printed = 0usize;
    loop {
        block.reset();
        let read_rows = reader.read(&mut block, 8192).await.expect("read failed");
        if read_rows == 0 {
            break;
        }
        let handles = block.get_handle_buf();
        let versions = block.get_version_buf();
        for i in 0..block.length() {
            if let Some(limit) = head {
                if printed >= limit {
                    return;
                }
            }
            let (handle_display, handle_raw) = if handles.get_fixed_size() > 0 {
                let v = handles.get_int_handle_value(i);
                (
                    v.to_string(),
                    log_wrappers::hex_encode_upper(handles.get_not_null_value(i)),
                )
            } else {
                (
                    format!(
                        "raw {}",
                        log_wrappers::hex_encode_upper(handles.get_not_null_value(i))
                    ),
                    log_wrappers::hex_encode_upper(handles.get_not_null_value(i)),
                )
            };
            let version = versions.get_version(i);
            let deleted = versions.is_nullable() && versions.is_null(i);
            println!(
                "  {}: handle={}, handle_raw={}, version={}, deleted={}",
                printed, handle_display, handle_raw, version, deleted
            );
            printed += 1;
        }
    }
}