zilliz 1.3.0

use std::collections::HashMap;
use std::io::{self, BufRead, IsTerminal, Write};

use anyhow::{bail, Context, Result};
use serde_json::Value;

use crate::api::client::ApiClient;
use crate::api::error::ApiError;
use crate::config::credentials::resolve_api_key;
use crate::config::manager::ConfigManager;
use crate::model::loader::Models;
use crate::model::types::{Operation, OutputConfig, Param};
use crate::service::executor::OperationExecutor;

use super::formatter;

/// Options for output formatting and post-processing.
pub struct OutputOpts<'a> {
    pub format: &'a str,
    pub query: Option<&'a str>,
    pub no_header: bool,
    pub wait: bool,
    pub api_key: Option<&'a str>,
}

impl<'a> OutputOpts<'a> {
    pub fn new(format: &'a str) -> Self {
        Self {
            format,
            query: None,
            no_header: false,
            wait: false,
            api_key: None,
        }
    }
}

const DANGEROUS_OPERATIONS: &[&str] = &["delete", "drop", "suspend", "release", "restore", "clear"];

/// Run a CLI command by resource name and operation name.
pub async fn run(
    models: &Models,
    config_mgr: &ConfigManager,
    resource: &str,
    operation: &str,
    raw_args: &[String],
    output_opts: &OutputOpts<'_>,
    fetch_all: bool,
) -> Result<()> {
    // Find the operation in control-plane or data-plane models
    let (op, is_data_plane) = find_operation(models, resource, operation)?;

    // Dangerous operation confirmation
    if DANGEROUS_OPERATIONS.contains(&operation) {
        let has_yes = raw_args.iter().any(|a| a == "--yes" || a == "-y");
        if !has_yes {
            print!(
                "Are you sure you want to {} {}? [y/N] ",
                operation, resource
            );
            io::stdout().flush()?;
            let mut input = String::new();
            io::stdin().lock().read_line(&mut input)?;
            if !input.trim().eq_ignore_ascii_case("y") {
                println!("Aborted.");
                return Ok(());
            }
        }
    }

    // Strip --yes/-y from args before parsing
    let raw_args: Vec<String> = raw_args
        .iter()
        .filter(|a| *a != "--yes" && *a != "-y")
        .cloned()
        .collect();

    // Check dedicated-only operations
    if op.dedicated_only {
        let ctx = config_mgr.get_context();
        let is_dedicated = ctx
            .plan
            .as_deref()
            .map(|p| p.eq_ignore_ascii_case("dedicated"))
            .unwrap_or(false);
        if !is_dedicated {
            bail!(
                "Operation '{} {}' is only available on Dedicated clusters.\n\
                 Your current context plan: {}.\n\
                 Set a Dedicated cluster context: zilliz context set --cluster-id <id>",
                resource,
                operation,
                ctx.plan.as_deref().unwrap_or("unknown")
            );
        }
    }

    // Resolve API key
    let api_key =
        resolve_api_key(output_opts.api_key, config_mgr).ok_or_else(|| ApiError::NoApiKey)?;

    // Resolve base URLs
    let cp_base_url =
        super::endpoint::resolve_control_plane_url(config_mgr, &models.control_plane, None);
    let base_url = if is_data_plane {
        let ctx = config_mgr.get_context();
        ctx.endpoint
            .context("No cluster context set. Run: zilliz context set --cluster-id <id>")?
    } else {
        cp_base_url.clone()
    };

    // Create a control-plane client (needed for entity selection prompts)
    let cp_client = ApiClient::new(api_key.clone(), cp_base_url);

    // Parse --flag value args into param map (same format as zilliz-cli)
    let mut param_values = parse_args(&raw_args, &op)?;

    // Validate required params -- prompt interactively if TTY, error otherwise
    let missing_params: Vec<&Param> = op
        .params
        .iter()
        .filter(|p| p.required && !param_values.contains_key(&p.name))
        .collect();
    if !missing_params.is_empty() {
        // Split into promptable (simple types) and non-promptable (array/object)
        let (promptable, complex): (Vec<&Param>, Vec<&Param>) = missing_params
            .into_iter()
            .partition(|p| !matches!(p.param_type.as_str(), "array" | "object"));

        if std::io::stdin().is_terminal() && !promptable.is_empty() {
            // Prompt for simple types first
            let prompted = prompt_missing_params(&promptable, &cp_client).await?;
            for (name, value) in prompted {
                param_values.insert(name, value);
            }
        } else if !promptable.is_empty() {
            // Non-TTY: error for all missing params at once
            let flags: Vec<String> = promptable
                .iter()
                .chain(complex.iter())
                .map(|p| p.cli_flag())
                .collect();
            bail!(
                "Missing required option{}: {}",
                if flags.len() > 1 { "s" } else { "" },
                flags.join(", ")
            );
        }

        // Error for complex types that can't be prompted (even in TTY mode)
        if !complex.is_empty() {
            let flags: Vec<String> = complex.iter().map(|p| p.cli_flag()).collect();
            bail!(
                "Missing required option{}: {}",
                if flags.len() > 1 { "s" } else { "" },
                flags.join(", ")
            );
        }
    }

    // Execute (reuse cp_client for control-plane, create new for data-plane)
    let client = if is_data_plane {
        ApiClient::new(api_key, base_url)
    } else {
        cp_client
    };
    let executor = OperationExecutor::new(&client);

    let result = if fetch_all && op.pagination.is_some() {
        let items = executor.execute_all_pages(&op, &param_values).await?;
        Value::Array(items)
    } else {
        executor.execute(&op, &param_values).await?
    };

    // --wait: if response contains jobId, poll until terminal state
    if output_opts.wait {
        if let Some(job_id) = result.get("jobId").and_then(|v| v.as_str()) {
            let wait_result = super::job_waiter::wait_for_job(&client, job_id, 1800, 5).await?;
            print_output_with_op_name(&wait_result, output_opts, Some(&op), Some(operation));
            return Ok(());
        }
    }

    // Custom table for backup list
    if resource == "backup"
        && operation == "list"
        && output_opts.format == "table"
        && output_opts.query.is_none()
    {
        print_backup_table(&result, output_opts.no_header);
        return Ok(());
    }

    print_output_with_op_name(&result, output_opts, Some(&op), Some(operation));

    Ok(())
}

/// Format and print the result based on output options.
pub fn print_output_with_opts(result: &Value, opts: &OutputOpts<'_>, op: Option<&Operation>) {
    print_output_with_op_name(result, opts, op, None)
}

/// Format and print the result, with an optional operation name for empty-result messaging.
pub fn print_output_with_op_name(
    result: &Value,
    opts: &OutputOpts<'_>,
    op: Option<&Operation>,
    op_name: Option<&str>,
) {
    let output_config = op.map(|o| &o.output);
    let pagination_data_field = op
        .and_then(|o| o.pagination.as_ref())
        .map(|p| p.data_field.as_str());
    print_output_impl(result, opts, output_config, pagination_data_field, op_name)
}

fn print_output_impl(
    result: &Value,
    opts: &OutputOpts<'_>,
    output_config: Option<&OutputConfig>,
    pagination_data_field: Option<&str>,
    op_name: Option<&str>,
) {
    // Apply JMESPath query filter if specified
    let filtered = if let Some(query) = opts.query {
        match formatter::apply_query(result, query) {
            Ok(v) => v,
            Err(e) => {
                eprintln!("Error: {}", e);
                return;
            }
        }
    } else {
        result.clone()
    };

    match opts.format {
        "json" => {
            println!("{}", formatter::format_json(&filtered));
        }
        "text" => {
            println!("{}", formatter::format_text(&filtered));
        }
        "yaml" => {
            println!("{}", formatter::format_yaml(&filtered));
        }
        "csv" => {
            println!("{}", formatter::format_csv(&filtered, opts.no_header));
        }
        _ => {
            // table format: try to extract data array and auto-detect columns
            // Normalize protobuf-wrapped array fields before display
            let mut normalized = filtered.clone();
            formatter::normalize_array_fields(&mut normalized);

            // Check output config first, then fall back to pagination data_field
            let data_field = output_config
                .and_then(|o| o.data_field.as_deref())
                .or(pagination_data_field);
            let items = extract_items(&normalized, data_field);

            // Empty result: show "Empty" for list operations (no data to display),
            // "Success" for mutations (create/drop/delete completed).
            // Check after extract_items so wrapped responses like
            // {"data": [], "total": 0} are correctly detected as empty.
            if is_empty_result(&filtered) || items.is_empty() {
                let is_list = op_name.map(|n| n == "list").unwrap_or(false);
                let label = if is_list { "Empty" } else { "Success" };
                if std::io::stdout().is_terminal() {
                    let icon = if is_list {
                        "\x1b[33m!\x1b[0m" // yellow ! for empty
                    } else {
                        "\x1b[32m\u{2714}\x1b[0m" // green checkmark for success
                    };
                    println!("{} {}", icon, label);
                } else {
                    println!("{}", label);
                }
                return;
            }

            if items.iter().any(|v| !v.is_object()) {
                // Non-object items (e.g. array of strings from database list):
                // render as a single-column table
                let col_name = output_config
                    .and_then(|o| o.columns.as_ref())
                    .and_then(|c| c.first())
                    .map(|s| s.as_str())
                    .unwrap_or("Name");
                let mut table = formatter::create_table(&[col_name], opts.no_header);
                for item in &items {
                    let val = match item {
                        Value::String(s) => s.clone(),
                        Value::Null => String::new(),
                        other => other.to_string(),
                    };
                    table.add_row([val]);
                }
                println!("{}", table);
            } else if items.len() == 1 {
                // Single object: vertical key-value layout
                println!("{}", formatter::format_kv_table(items[0]));
            } else {
                // Use explicit columns from operation config, or auto-detect
                let auto_cols;
                let col_refs = match output_config.and_then(|o| o.columns.as_ref()) {
                    Some(cols) => cols.iter().map(|s| s.as_str()).collect::<Vec<&str>>(),
                    None => {
                        auto_cols = formatter::auto_columns(&items);
                        auto_cols.iter().map(|s| s.as_str()).collect::<Vec<&str>>()
                    }
                };
                let color_map = output_config.and_then(|o| o.color_map.as_ref());
                println!(
                    "{}",
                    formatter::format_table_with_opts(&items, &col_refs, opts.no_header, color_map)
                );
            }
        }
    }
}

/// Extract array items from a result, trying data_field first, then common patterns.
/// Check if a result is "empty" — null, empty object, or empty array.
fn is_empty_result(value: &Value) -> bool {
    match value {
        Value::Null => true,
        Value::Object(map) => map.is_empty(),
        Value::Array(arr) => arr.is_empty(),
        _ => false,
    }
}

fn extract_items<'a>(result: &'a Value, data_field: Option<&str>) -> Vec<&'a Value> {
    // Try explicit data_field from operation output config
    if let Some(field) = data_field {
        if let Some(arr) = result.get(field).and_then(|v| v.as_array()) {
            return arr.iter().collect();
        }
    }

    // If result is already an array
    if let Some(arr) = result.as_array() {
        return arr.iter().collect();
    }

    // Try common field names
    for field in &[
        "data",
        "results",
        "items",
        "clusters",
        "collections",
        "backups",
        "volumes",
        "invoices",
        "records",
        "jobs",
    ] {
        if let Some(arr) = result.get(*field).and_then(|v| v.as_array()) {
            return arr.iter().collect();
        }
    }

    // Single object: wrap as one-item list for table display
    if result.is_object() {
        return vec![result];
    }

    vec![]
}

/// Prompt the user interactively for missing required parameters.
///
/// When a parameter has an entity selector (e.g. `clusterId`), the prompt
/// fetches available entities via the API and presents a numbered selection
/// list instead of a free-text input.
async fn prompt_missing_params(
    params: &[&Param],
    client: &ApiClient,
) -> Result<HashMap<String, Value>> {
    let mut values = HashMap::new();
    let stdin = io::stdin();
    let stderr = io::stderr();

    for param in params {
        let description = param.description.as_deref().unwrap_or("");
        let label = if description.is_empty() {
            param.name.clone()
        } else {
            format!("{} ({})", param.name, description)
        };

        let value = match param.param_type.as_str() {
            "boolean" => prompt_boolean(&stdin, &stderr, &label)?,
            "integer" => prompt_integer(&stdin, &stderr, &label)?,
            _ => {
                if param.name == "clusterId" || param.name == "destClusterId" {
                    prompt_cluster_select(&stdin, &stderr, client).await?
                } else if param.name == "projectId" {
                    prompt_project_select(&stdin, &stderr, client).await?
                } else if param.name == "backupId" {
                    let cluster_id: Option<String> = values
                        .get("clusterId")
                        .and_then(|v: &Value| v.as_str())
                        .map(|s: &str| s.to_string());
                    if let Some(ref cid) = cluster_id {
                        prompt_backup_select(&stdin, &stderr, client, cid).await?
                    } else {
                        prompt_string(&stdin, &stderr, &label)?
                    }
                } else if param.name == "volumeName" && param.position.as_deref() == Some("path") {
                    let project_id: Option<String> = values
                        .get("projectId")
                        .and_then(|v: &Value| v.as_str())
                        .map(|s: &str| s.to_string());
                    let pid = match project_id {
                        Some(pid) => pid,
                        None => {
                            let v = prompt_project_select(&stdin, &stderr, client).await?;
                            v.as_str().unwrap_or("").to_string()
                        }
                    };
                    if !pid.is_empty() {
                        prompt_volume_select(&stdin, &stderr, client, &pid).await?
                    } else {
                        prompt_string(&stdin, &stderr, &label)?
                    }
                } else if param.name == "jobId" {
                    let cluster_id: Option<String> = values
                        .get("clusterId")
                        .and_then(|v: &Value| v.as_str())
                        .map(|s: &str| s.to_string());
                    if let Some(ref cid) = cluster_id {
                        prompt_import_job_select(&stdin, &stderr, client, cid).await?
                    } else {
                        prompt_string(&stdin, &stderr, &label)?
                    }
                } else if let Some(choices) = &param.choices {
                    prompt_choices(&stdin, &stderr, &label, choices)?
                } else {
                    prompt_string(&stdin, &stderr, &label)?
                }
            }
        };

        values.insert(param.name.clone(), value);
    }

    Ok(values)
}

/// Fetch clusters from the API and present a numbered selection list.
async fn prompt_cluster_select(
    stdin: &io::Stdin,
    stderr: &io::Stderr,
    client: &ApiClient,
) -> Result<Value> {
    let resp = client
        .call("GET", "/v2/clusters", None, None)
        .await
        .context("Failed to fetch cluster list")?;

    let clusters: Vec<&Value> = resp
        .get("clusters")
        .or_else(|| resp.get("data").and_then(|d| d.get("clusters")))
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().collect())
        .unwrap_or_default();

    if clusters.is_empty() {
        bail!("No clusters found. Create a cluster first with: zilliz cluster create");
    }

    // Single cluster: auto-select
    if clusters.len() == 1 {
        let name = clusters[0]
            .get("clusterName")
            .and_then(|v| v.as_str())
            .unwrap_or("?");
        let id = clusters[0]
            .get("clusterId")
            .and_then(|v| v.as_str())
            .context("Cluster has no ID")?;
        eprintln!("Using cluster: {} ({})", name, id);
        return Ok(Value::String(id.to_string()));
    }

    // Multiple clusters: numbered selection
    eprintln!("Select cluster:");
    for (i, c) in clusters.iter().enumerate() {
        let name = c.get("clusterName").and_then(|v| v.as_str()).unwrap_or("?");
        let id = c.get("clusterId").and_then(|v| v.as_str()).unwrap_or("?");
        eprintln!("  {}) {} ({})", i + 1, name, id);
    }
    loop {
        eprint!("Select [1-{}]: ", clusters.len());
        stderr.lock().flush()?;
        let mut input = String::new();
        let n_read = stdin.lock().read_line(&mut input)?;
        if n_read == 0 {
            bail!("EOF on stdin");
        }
        let trimmed = input.trim();
        if let Ok(n) = trimmed.parse::<usize>() {
            if n >= 1 && n <= clusters.len() {
                let id = clusters[n - 1]
                    .get("clusterId")
                    .and_then(|v| v.as_str())
                    .context("Cluster has no ID")?;
                return Ok(Value::String(id.to_string()));
            }
        }
        eprintln!("Invalid selection, please try again.");
    }
}

/// Fetch projects from the API and present a numbered selection list.
async fn prompt_project_select(
    stdin: &io::Stdin,
    stderr: &io::Stderr,
    client: &ApiClient,
) -> Result<Value> {
    let resp = client
        .call("GET", "/v2/projects", None, None)
        .await
        .context("Failed to fetch project list")?;

    let projects: Vec<&Value> = resp
        .get("projects")
        .or_else(|| resp.get("data").and_then(|d| d.get("projects")))
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().collect())
        .unwrap_or_default();

    if projects.is_empty() {
        bail!("No projects found.");
    }

    if projects.len() == 1 {
        let name = projects[0]
            .get("projectName")
            .and_then(|v| v.as_str())
            .unwrap_or("?");
        let id = projects[0]
            .get("projectId")
            .and_then(|v| v.as_str())
            .context("Project has no ID")?;
        eprintln!("Using project: {} ({})", name, id);
        return Ok(Value::String(id.to_string()));
    }

    eprintln!("Select project:");
    for (i, p) in projects.iter().enumerate() {
        let name = p.get("projectName").and_then(|v| v.as_str()).unwrap_or("?");
        let id = p.get("projectId").and_then(|v| v.as_str()).unwrap_or("?");
        eprintln!("  {}) {} ({})", i + 1, name, id);
    }
    loop {
        eprint!("Select [1-{}]: ", projects.len());
        stderr.lock().flush()?;
        let mut input = String::new();
        let n_read = stdin.lock().read_line(&mut input)?;
        if n_read == 0 {
            bail!("EOF on stdin");
        }
        let trimmed = input.trim();
        if let Ok(n) = trimmed.parse::<usize>() {
            if n >= 1 && n <= projects.len() {
                let id = projects[n - 1]
                    .get("projectId")
                    .and_then(|v| v.as_str())
                    .context("Project has no ID")?;
                return Ok(Value::String(id.to_string()));
            }
        }
        eprintln!("Invalid selection, please try again.");
    }
}

/// Print backup list as a curated table with formatted createTime.
fn print_backup_table(result: &Value, no_header: bool) {
    let backups = result
        .get("backups")
        .and_then(|v| v.as_array())
        .cloned()
        .unwrap_or_default();

    if backups.is_empty() {
        println!("No backups found.");
        return;
    }

    let columns = &[
        "backupName",
        "backupId",
        "backupType",
        "clusterName",
        "clusterId",
        "projectId",
        "creationMethod",
        "createTime",
        "status",
    ];
    let rows: Vec<Vec<String>> = backups
        .iter()
        .map(|b| {
            columns
                .iter()
                .map(|col| {
                    let val = b
                        .get(*col)
                        .map(|v| match v {
                            Value::String(s) => s.clone(),
                            Value::Null => String::new(),
                            other => other.to_string(),
                        })
                        .unwrap_or_default();
                    if *col == "createTime" {
                        // "2026-01-15T10:30:00Z" -> "2026-01-15 10:30"
                        format_datetime(&val)
                    } else {
                        val
                    }
                })
                .collect()
        })
        .collect();

    let mut table = formatter::create_table(columns, no_header);
    for row in rows {
        table.add_row(row);
    }
    println!("{}", table);
}

/// Format an ISO 8601 timestamp to "YYYY-MM-DD HH:mm".
fn format_datetime(ts: &str) -> String {
    // Expect "2026-01-15T10:30:00Z" or similar ISO format
    if ts.len() >= 16 && ts.as_bytes().get(10) == Some(&b'T') {
        format!("{} {}", &ts[..10], &ts[11..16])
    } else {
        ts.to_string()
    }
}

/// Fetch backups for a cluster and present a numbered selection list.
async fn prompt_backup_select(
    stdin: &io::Stdin,
    stderr: &io::Stderr,
    client: &ApiClient,
    cluster_id: &str,
) -> Result<Value> {
    let body = serde_json::json!({
        "clusterId": cluster_id,
        "pageSize": 100,
        "currentPage": 1,
    });
    let resp = client
        .call("GET", "/v2/backups", None, Some(&body))
        .await
        .context("Failed to fetch backup list")?;

    let backups: Vec<&Value> = resp
        .get("backups")
        .or_else(|| resp.get("data").and_then(|d| d.get("backups")))
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().collect())
        .unwrap_or_default();

    if backups.is_empty() {
        bail!("No backups found.");
    }

    if backups.len() == 1 {
        let name = backups[0]
            .get("backupName")
            .and_then(|v| v.as_str())
            .unwrap_or("?");
        let id = backups[0]
            .get("backupId")
            .and_then(|v| v.as_str())
            .context("Backup has no ID")?;
        eprintln!("Using backup: {} ({})", name, id);
        return Ok(Value::String(id.to_string()));
    }

    eprintln!("Select backup:");
    for (i, b) in backups.iter().enumerate() {
        let name = b.get("backupName").and_then(|v| v.as_str()).unwrap_or("?");
        let id = b.get("backupId").and_then(|v| v.as_str()).unwrap_or("?");
        let btype = b.get("backupType").and_then(|v| v.as_str()).unwrap_or("?");
        let status = b.get("status").and_then(|v| v.as_str()).unwrap_or("?");
        eprintln!("  {}) {} ({}) - {} - {}", i + 1, name, id, btype, status);
    }
    loop {
        eprint!("Select [1-{}]: ", backups.len());
        stderr.lock().flush()?;
        let mut input = String::new();
        let n_read = stdin.lock().read_line(&mut input)?;
        if n_read == 0 {
            bail!("EOF on stdin");
        }
        let trimmed = input.trim();
        if let Ok(n) = trimmed.parse::<usize>() {
            if n >= 1 && n <= backups.len() {
                let id = backups[n - 1]
                    .get("backupId")
                    .and_then(|v| v.as_str())
                    .context("Backup has no ID")?;
                return Ok(Value::String(id.to_string()));
            }
        }
        eprintln!("Invalid selection, please try again.");
    }
}

/// Fetch volumes for a project and present a numbered selection list.
async fn prompt_volume_select(
    stdin: &io::Stdin,
    stderr: &io::Stderr,
    client: &ApiClient,
    project_id: &str,
) -> Result<Value> {
    let body = serde_json::json!({
        "projectId": project_id,
        "pageSize": 100,
        "currentPage": 1,
    });
    let resp = client
        .call("GET", "/v2/volumes", None, Some(&body))
        .await
        .context("Failed to fetch volume list")?;

    let volumes: Vec<&Value> = resp
        .get("volumes")
        .or_else(|| resp.get("data").and_then(|d| d.get("volumes")))
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().collect())
        .unwrap_or_default();

    if volumes.is_empty() {
        bail!("No volumes found.");
    }

    if volumes.len() == 1 {
        let name = volumes[0]
            .get("volumeName")
            .and_then(|v| v.as_str())
            .context("Volume has no name")?;
        let region = volumes[0]
            .get("regionId")
            .and_then(|v| v.as_str())
            .unwrap_or("?");
        eprintln!("Using volume: {} ({})", name, region);
        return Ok(Value::String(name.to_string()));
    }

    eprintln!("Select volume:");
    for (i, v) in volumes.iter().enumerate() {
        let name = v.get("volumeName").and_then(|v| v.as_str()).unwrap_or("?");
        let region = v.get("regionId").and_then(|v| v.as_str()).unwrap_or("?");
        eprintln!("  {}) {} ({})", i + 1, name, region);
    }
    loop {
        eprint!("Select [1-{}]: ", volumes.len());
        stderr.lock().flush()?;
        let mut input = String::new();
        let n_read = stdin.lock().read_line(&mut input)?;
        if n_read == 0 {
            bail!("EOF on stdin");
        }
        let trimmed = input.trim();
        if let Ok(n) = trimmed.parse::<usize>() {
            if n >= 1 && n <= volumes.len() {
                let name = volumes[n - 1]
                    .get("volumeName")
                    .and_then(|v| v.as_str())
                    .context("Volume has no name")?;
                return Ok(Value::String(name.to_string()));
            }
        }
        eprintln!("Invalid selection, please try again.");
    }
}

/// Fetch import jobs for a cluster and present a numbered selection list.
async fn prompt_import_job_select(
    stdin: &io::Stdin,
    stderr: &io::Stderr,
    client: &ApiClient,
    cluster_id: &str,
) -> Result<Value> {
    let body = serde_json::json!({
        "clusterId": cluster_id,
        "pageSize": 100,
        "currentPage": 1,
    });
    let resp = client
        .call("POST", "/v2/vectordb/jobs/import/list", None, Some(&body))
        .await
        .context("Failed to fetch import job list")?;

    let jobs: Vec<&Value> = resp
        .get("records")
        .or_else(|| resp.get("data").and_then(|d| d.get("records")))
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().collect())
        .unwrap_or_default();

    if jobs.is_empty() {
        bail!("No import jobs found.");
    }

    if jobs.len() == 1 {
        let id = jobs[0]
            .get("jobId")
            .and_then(|v| v.as_str())
            .context("Job has no ID")?;
        let status = jobs[0].get("state").and_then(|v| v.as_str()).unwrap_or("?");
        eprintln!("Using import job: {} ({})", id, status);
        return Ok(Value::String(id.to_string()));
    }

    eprintln!("Select import job:");
    for (i, j) in jobs.iter().enumerate() {
        let id = j.get("jobId").and_then(|v| v.as_str()).unwrap_or("?");
        let status = j.get("state").and_then(|v| v.as_str()).unwrap_or("?");
        let progress = j.get("progress").and_then(|v| v.as_i64()).unwrap_or(0);
        eprintln!("  {}) {} - {} - {}%", i + 1, id, status, progress);
    }
    loop {
        eprint!("Select [1-{}]: ", jobs.len());
        stderr.lock().flush()?;
        let mut input = String::new();
        let n_read = stdin.lock().read_line(&mut input)?;
        if n_read == 0 {
            bail!("EOF on stdin");
        }
        let trimmed = input.trim();
        if let Ok(n) = trimmed.parse::<usize>() {
            if n >= 1 && n <= jobs.len() {
                let id = jobs[n - 1]
                    .get("jobId")
                    .and_then(|v| v.as_str())
                    .context("Job has no ID")?;
                return Ok(Value::String(id.to_string()));
            }
        }
        eprintln!("Invalid selection, please try again.");
    }
}

fn prompt_string(stdin: &io::Stdin, stderr: &io::Stderr, label: &str) -> Result<Value> {
    loop {
        eprint!("{}: ", label);
        stderr.lock().flush()?;
        let mut input = String::new();
        let n_read = stdin.lock().read_line(&mut input)?;
        if n_read == 0 {
            bail!("EOF on stdin");
        }
        let trimmed = input.trim();
        if !trimmed.is_empty() {
            return Ok(Value::String(trimmed.to_string()));
        }
        eprintln!("Value required, please try again.");
    }
}

fn prompt_integer(stdin: &io::Stdin, stderr: &io::Stderr, label: &str) -> Result<Value> {
    loop {
        eprint!("{}: ", label);
        stderr.lock().flush()?;
        let mut input = String::new();
        let n_read = stdin.lock().read_line(&mut input)?;
        if n_read == 0 {
            bail!("EOF on stdin");
        }
        let trimmed = input.trim();
        if trimmed.is_empty() {
            eprintln!("Value required, please try again.");
            continue;
        }
        match trimmed.parse::<i64>() {
            Ok(n) => return Ok(Value::Number(n.into())),
            Err(_) => eprintln!("Invalid integer, please try again."),
        }
    }
}

fn prompt_boolean(stdin: &io::Stdin, stderr: &io::Stderr, label: &str) -> Result<Value> {
    loop {
        eprint!("{} [y/N]: ", label);
        stderr.lock().flush()?;
        let mut input = String::new();
        let n_read = stdin.lock().read_line(&mut input)?;
        if n_read == 0 {
            bail!("EOF on stdin");
        }
        let trimmed = input.trim();
        if trimmed.is_empty() {
            return Ok(Value::Bool(false));
        }
        match trimmed.to_ascii_lowercase().as_str() {
            "y" | "yes" => return Ok(Value::Bool(true)),
            "n" | "no" => return Ok(Value::Bool(false)),
            _ => eprintln!("Please enter y or n."),
        }
    }
}

fn prompt_choices(
    stdin: &io::Stdin,
    stderr: &io::Stderr,
    label: &str,
    choices: &[String],
) -> Result<Value> {
    eprintln!("{}:", label);
    for (i, choice) in choices.iter().enumerate() {
        eprintln!("  [{}] {}", i + 1, choice);
    }
    loop {
        eprint!("Select [1-{}]: ", choices.len());
        stderr.lock().flush()?;
        let mut input = String::new();
        let n_read = stdin.lock().read_line(&mut input)?;
        if n_read == 0 {
            bail!("EOF on stdin");
        }
        let trimmed = input.trim();
        if let Ok(n) = trimmed.parse::<usize>() {
            if n >= 1 && n <= choices.len() {
                return Ok(Value::String(choices[n - 1].clone()));
            }
        }
        eprintln!("Invalid selection, please try again.");
    }
}

fn find_operation(models: &Models, resource: &str, operation: &str) -> Result<(Operation, bool)> {
    // Check control plane first
    if let Some(res) = models.control_plane.resources.get(resource) {
        if let Some(op) = res.operations.get(operation) {
            return Ok((op.clone(), false));
        }
    }

    // Then data plane
    if let Some(res) = models.data_plane.resources.get(resource) {
        if let Some(op) = res.operations.get(operation) {
            return Ok((op.clone(), true));
        }
    }

    // Check if resource exists but operation doesn't
    for model in [&models.control_plane, &models.data_plane] {
        if let Some(res) = model.resources.get(resource) {
            let ops: Vec<&str> = res.operations.keys().map(|s| s.as_str()).collect();
            let suggestion = suggest_closest(operation, &ops);
            bail!(
                "Unknown operation '{}' for resource '{}'. {}Available operations: {}",
                operation,
                resource,
                suggestion,
                ops.join(", ")
            );
        }
    }

    let mut all_resources = Vec::new();
    for model in [&models.control_plane, &models.data_plane] {
        for key in model.resources.keys() {
            all_resources.push(key.as_str());
        }
    }
    let suggestion = suggest_closest(resource, &all_resources);
    bail!(
        "Unknown resource '{}'. {}Available resources: {}",
        resource,
        suggestion,
        all_resources.join(", ")
    );
}

/// Suggest the closest match using edit distance.
fn suggest_closest(input: &str, candidates: &[&str]) -> String {
    let mut best: Option<(&str, usize)> = None;
    for &c in candidates {
        let d = edit_distance(input, c);
        if d <= 3 && (best.is_none() || d < best.unwrap().1) {
            best = Some((c, d));
        }
    }
    match best {
        Some((name, _)) => format!("Did you mean '{}'? ", name),
        None => String::new(),
    }
}

/// Simple Levenshtein edit distance.
fn edit_distance(a: &str, b: &str) -> usize {
    let a: Vec<char> = a.chars().collect();
    let b: Vec<char> = b.chars().collect();
    let mut dp = vec![vec![0usize; b.len() + 1]; a.len() + 1];
    for (i, row) in dp.iter_mut().enumerate().take(a.len() + 1) {
        row[0] = i;
    }
    for (j, val) in dp[0].iter_mut().enumerate().take(b.len() + 1) {
        *val = j;
    }
    for i in 1..=a.len() {
        for j in 1..=b.len() {
            let cost = if a[i - 1] == b[j - 1] { 0 } else { 1 };
            dp[i][j] = (dp[i - 1][j] + 1)
                .min(dp[i][j - 1] + 1)
                .min(dp[i - 1][j - 1] + cost);
        }
    }
    dp[a.len()][b.len()]
}

/// Parse `--flag value` style args into a param values map (compatible with zilliz-cli).
///
/// Supports:
///   --cluster-id abc123          (matched by cli field in model)
///   --name my_col                (matched by param name)
///   --auto-id                    (boolean flag, no value needed)
///   --body '{"key": "value"}'    (raw JSON body)
///   --body file://schema.json    (JSON from file)
pub fn parse_args(raw_args: &[String], operation: &Operation) -> Result<HashMap<String, Value>> {
    let mut values = HashMap::new();
    let mut i = 0;

    while i < raw_args.len() {
        let arg = &raw_args[i];

        if !arg.starts_with("--") {
            bail!("Unexpected argument '{}'. Use --flag value format.", arg);
        }

        let flag = arg.as_str();

        // Handle --body specially (raw JSON passthrough)
        if flag == "--body" {
            i += 1;
            let body_str = raw_args
                .get(i)
                .context("--body requires a JSON value or file://path")?;
            let body_json = parse_json_or_file(body_str)?;
            // Merge body fields into values
            if let Value::Object(map) = body_json {
                for (k, v) in map {
                    values.insert(k, v);
                }
            } else {
                bail!("--body must be a JSON object");
            }
            i += 1;
            continue;
        }

        // Find the matching param by cli flag name or --param-name
        let param = operation.params.iter().find(|p| {
            p.cli_flag() == flag
                || format!("--{}", p.name) == flag
                || format!("--{}", p.name.replace('_', "-")) == flag
        });

        match param {
            Some(p) => {
                if p.param_type == "boolean" {
                    // Boolean flags: if next arg looks like a value, consume it; otherwise true
                    let next = raw_args.get(i + 1);
                    match next.map(|s| s.as_str()) {
                        Some("true") | Some("false") => {
                            let b: bool = next.unwrap().parse().unwrap();
                            values.insert(p.name.clone(), Value::Bool(b));
                            i += 2;
                        }
                        _ => {
                            values.insert(p.name.clone(), Value::Bool(true));
                            i += 1;
                        }
                    }
                } else {
                    // Non-boolean: consume next arg as value
                    i += 1;
                    let val_str = raw_args
                        .get(i)
                        .with_context(|| format!("{} requires a value", flag))?;

                    let typed_value = match p.param_type.as_str() {
                        "integer" => {
                            let n: i64 = val_str
                                .parse()
                                .with_context(|| format!("{} must be an integer", flag))?;
                            Value::Number(n.into())
                        }
                        "array" | "object" => parse_json_or_file(val_str)
                            .with_context(|| format!("{} must be valid JSON", flag))?,
                        _ => Value::String(val_str.to_string()),
                    };
                    values.insert(p.name.clone(), typed_value);
                    i += 1;
                }
            }
            None => {
                bail!(
                    "Unknown flag '{}'. Available flags: {}",
                    flag,
                    operation
                        .params
                        .iter()
                        .map(|p| p.cli_flag())
                        .collect::<Vec<_>>()
                        .join(", ")
                );
            }
        }
    }

    Ok(values)
}

/// Parse a JSON string or read from a file:// path.
fn parse_json_or_file(input: &str) -> Result<Value> {
    if let Some(path) = input.strip_prefix("file://") {
        let content =
            std::fs::read_to_string(path).with_context(|| format!("Cannot read file: {}", path))?;
        serde_json::from_str(&content).with_context(|| format!("Invalid JSON in file: {}", path))
    } else {
        serde_json::from_str(input).with_context(|| format!("Invalid JSON for --body: {}", input))
    }
}