tecton 0.1.1

//! Tecton CLI — headless entrypoint for storage and compute services.

mod benchmark;

use clap::{Parser, Subcommand, ValueEnum};
use std::path::PathBuf;
use std::time::Instant;
use tecton_compute::{execute_query_json, ComputeEngine};
use tecton_core::{
    init_tracing, RuntimeMode, ServiceState, SystemStatus, TectonConfig, VERSION,
};
use tecton_io::{collect_storage_stats, ensure_data_dir, start_server, StorageEngine, DATA_DIR};
use tracing::{error, info};

#[derive(Debug, Parser)]
#[command(
    name = "tecton",
    author = "Roberto de Souza <rabbittrix@hotmail.com>",
    version = VERSION,
    about = "Tecton — unified storage + compute data infrastructure",
    long_about = "A modular, blazing-fast alternative to legacy Hadoop/MinIO stacks.\n\
                  Start S3-compatible storage, a SQL engine, or both from a single binary."
)]
struct Cli {
    /// Path to a TOML/YAML config file (optional).
    #[arg(short, long, global = true, env = "TECTON_CONFIG")]
    config: Option<PathBuf>,

    #[command(subcommand)]
    command: Commands,
}

#[derive(Debug, Subcommand)]
enum Commands {
    /// Start Tecton services (storage, compute, or full).
    Start {
        /// Runtime mode.
        #[arg(long, short = 'm', value_enum, default_value_t = ModeArg::Full)]
        mode: ModeArg,

        /// Override bind host.
        #[arg(long, env = "TECTON_HOST")]
        host: Option<String>,

        /// Override bind port.
        #[arg(long, short, env = "TECTON_PORT")]
        port: Option<u16>,

        /// Override storage root directory (fs backend).
        #[arg(long)]
        data_dir: Option<PathBuf>,
    },

    /// Print aggregated system status as JSON.
    Status {
        #[arg(long, short = 'm', value_enum, default_value_t = ModeArg::Full)]
        mode: ModeArg,
    },

    /// Execute SQL against a CSV file (registered as table `data`).
    ///
    /// Example:
    ///   tecton query --file data.csv --sql "SELECT * FROM data LIMIT 5"
    Query {
        /// Path to the CSV file (absolute, relative, or a name under `./data`).
        #[arg(long, short = 'f')]
        file: PathBuf,

        /// SQL statement. Use table name `data` for the CSV contents.
        #[arg(long, short = 's')]
        sql: String,
    },

    /// List objects in the configured storage backend.
    Ls {
        #[arg(default_value = "")]
        prefix: String,
    },

    /// Benchmark CSV → Parquet conversion and query performance.
    ///
    /// Example:
    ///   tecton benchmark --file apps/tecton-ui/src-tauri/data/BANK_DATA.csv
    Benchmark {
        /// Path to a large CSV dataset (≥100 MiB recommended for conversion).
        #[arg(long, short = 'f')]
        file: PathBuf,

        /// Keep the generated `.parquet` cache after the run (default: true).
        #[arg(long, default_value_t = true)]
        keep_parquet: bool,
    },

    /// Print version information.
    Version,
}

#[derive(Debug, Clone, Copy, ValueEnum)]
enum ModeArg {
    Storage,
    Compute,
    Full,
}

impl From<ModeArg> for RuntimeMode {
    fn from(value: ModeArg) -> Self {
        match value {
            ModeArg::Storage => RuntimeMode::Storage,
            ModeArg::Compute => RuntimeMode::Compute,
            ModeArg::Full => RuntimeMode::Full,
        }
    }
}

#[tokio::main]
async fn main() {
    if let Err(err) = run().await {
        eprintln!("error: {err}");
        std::process::exit(1);
    }
}

async fn run() -> anyhow::Result<()> {
    let cli = Cli::parse();

    match cli.command {
        Commands::Version => {
            println!("{} {}", tecton_core::PRODUCT_NAME, VERSION);
            Ok(())
        }
        Commands::Start {
            mode,
            host,
            port,
            data_dir,
        } => {
            let mut cfg = TectonConfig::load(cli.config.as_deref())?;
            cfg.mode = mode.into();
            if let Some(h) = host {
                validate_host(&h)?;
                cfg.server.host = h;
            }
            if let Some(p) = port {
                if p == 0 {
                    anyhow::bail!("port must be non-zero");
                }
                cfg.server.port = p;
            }
            if let Some(dir) = data_dir {
                cfg.storage.root = dir.clone();
                cfg.compute.data_dir = dir.join("datasets");
            }
            cfg.validate()?;
            init_tracing(&cfg.log_level)?;
            // Ensure local upload directory exists alongside the storage service.
            ensure_data_dir().await?;
            start_services(cfg).await
        }
        Commands::Status { mode } => {
            let mut cfg = TectonConfig::load(cli.config.as_deref())?;
            cfg.mode = mode.into();
            init_tracing("warn")?;
            let status = build_status(&cfg).await?;
            println!("{}", serde_json::to_string_pretty(&status)?);
            Ok(())
        }
        Commands::Query { file, sql } => {
            init_tracing("warn")?;
            validate_sql_arg(&sql)?;
            let csv_path = resolve_csv_path(&file)?;
            let json = execute_query_json(&csv_path, &sql).await?;
            println!("{json}");
            Ok(())
        }
        Commands::Ls { prefix } => {
            let cfg = TectonConfig::load(cli.config.as_deref())?;
            cfg.validate()?;
            init_tracing(&cfg.log_level)?;
            let engine = StorageEngine::from_config(&cfg.storage).await?;
            let keys = engine.list(&prefix).await?;
            for key in keys {
                println!("{key}");
            }
            Ok(())
        }
        Commands::Benchmark { file, keep_parquet } => {
            init_tracing("info")?;
            benchmark::run_benchmark(file, keep_parquet).await
        }
    }
}

/// Resolve `--file` to a concrete CSV path.
///
/// Accepts an existing path as-is; otherwise looks under `./data/<name>`.
fn resolve_csv_path(file: &PathBuf) -> anyhow::Result<String> {
    if file.as_os_str().is_empty() {
        anyhow::bail!("--file must not be empty");
    }

    if file.exists() {
        return Ok(file.to_string_lossy().into_owned());
    }

    let under_data = PathBuf::from(DATA_DIR).join(file);
    if under_data.exists() {
        return Ok(under_data.to_string_lossy().into_owned());
    }

    // Bare filename like `data.csv` → `./data/data.csv`
    if file.components().count() == 1 {
        let candidate = PathBuf::from(DATA_DIR).join(file);
        if candidate.exists() {
            return Ok(candidate.to_string_lossy().into_owned());
        }
        anyhow::bail!(
            "CSV not found: '{}' (also checked '{}/{}')",
            file.display(),
            DATA_DIR,
            file.display()
        );
    }

    anyhow::bail!("CSV not found: {}", file.display());
}

async fn start_services(cfg: TectonConfig) -> anyhow::Result<()> {
    info!(version = VERSION, mode = %cfg.mode, "starting Tecton");

    let started = Instant::now();
    let mut storage_handle = None;
    let mut compute_engine = None;

    if cfg.mode.includes_storage() {
        storage_handle = Some(start_server(&cfg).await?);
        info!(addr = %storage_handle.as_ref().unwrap().addr, "storage API online");
    }

    if cfg.mode.includes_compute() {
        let engine = ComputeEngine::from_config(&cfg.compute).await?;
        let tables = engine.list_tables().await.unwrap_or_default();
        info!(tables = tables.len(), "compute engine online");
        compute_engine = Some(engine);
    }

    let _compute_engine = compute_engine;

    println!(
        "Tecton {} running in '{}' mode (ctrl+c to stop)",
        VERSION, cfg.mode
    );
    if let Some(handle) = &storage_handle {
        println!("  Storage API : http://{}", handle.addr);
        println!("  Health      : http://{}/health", handle.addr);
        println!("  Upload CSV  : POST http://{}/upload", handle.addr);
        println!("  Local data  : {DATA_DIR}");
    }
    if cfg.mode.includes_compute() {
        println!("  Compute     : ready");
        println!(
            "  Query example: tecton query --file {DATA_DIR}/data.csv --sql \"SELECT * FROM data LIMIT 5\""
        );
        println!("  Datasets    : {}", cfg.compute.data_dir.display());
    }
    println!("  Elapsed     : {}ms", started.elapsed().as_millis());

    tokio::signal::ctrl_c().await?;
    info!("shutdown signal received");

    if let Some(handle) = storage_handle {
        if let Err(e) = handle.shutdown().await {
            error!(error = %e, "storage shutdown error");
        }
    }

    println!("Tecton stopped.");
    Ok(())
}

async fn build_status(cfg: &TectonConfig) -> anyhow::Result<SystemStatus> {
    let mut status = SystemStatus::stopped(cfg.mode.to_string());

    if cfg.mode.includes_storage() {
        match StorageEngine::from_config(&cfg.storage).await {
            Ok(engine) => {
                status.storage = ServiceState::Running;
                status.storage_stats = collect_storage_stats(&engine).await.unwrap_or_default();
                status.bind_addr = Some(format!("{}:{}", cfg.server.host, cfg.server.port));
            }
            Err(_) => status.storage = ServiceState::Error,
        }
    }

    if cfg.mode.includes_compute() {
        match ComputeEngine::from_config(&cfg.compute).await {
            Ok(_) => status.compute = ServiceState::Running,
            Err(_) => status.compute = ServiceState::Error,
        }
    }

    status.state = match (status.storage, status.compute, cfg.mode) {
        (ServiceState::Running, _, RuntimeMode::Storage)
        | (_, ServiceState::Running, RuntimeMode::Compute)
        | (ServiceState::Running, ServiceState::Running, RuntimeMode::Full) => ServiceState::Running,
        (ServiceState::Error, _, _) | (_, ServiceState::Error, _) => ServiceState::Degraded,
        _ => ServiceState::Stopped,
    };

    Ok(status)
}

fn validate_host(host: &str) -> anyhow::Result<()> {
    if host.is_empty() || host.len() > 253 {
        anyhow::bail!("invalid host");
    }
    if host.chars().any(|c| c.is_control() || c == ' ') {
        anyhow::bail!("host contains invalid characters");
    }
    Ok(())
}

fn validate_sql_arg(sql: &str) -> anyhow::Result<()> {
    if sql.trim().is_empty() {
        anyhow::bail!("SQL query must not be empty");
    }
    if sql.len() > 32_768 {
        anyhow::bail!("SQL query too large");
    }
    Ok(())
}