//! CLI command handlers and shared utilities.
//!
//! This module contains the non-server CLI commands (`pub`, `sub`, `fetch`,
//! `info`, `drain`, `health`, `config`, `bench`) and shared helpers for
//! output formatting, topic parsing, and CLI config file loading.

use std::collections::HashMap;
use std::fmt;
use std::path::PathBuf;

use colored::Colorize;
use serde::Deserialize;

use crate::client::{ClientConfig, ClientTlsConfig};
use crate::types::{ApiKey, NamespaceId, TopicName};

pub mod bench;
pub mod config;
pub mod drain;
pub mod fetch;
pub mod health;
pub mod info;
pub mod publish;
pub mod subscribe;

// ---------------------------------------------------------------------------
// CliError — all CLI commands return this instead of calling process::exit
// ---------------------------------------------------------------------------

/// Error type returned by CLI command runners.
///
/// The `main()` dispatch handles printing the error and choosing the exit code.
#[derive(Debug)]
pub enum CliError {
    /// General error (exit code 1).
    Error(String),
    /// Timeout (exit code 2).
    Timeout(String),
}

impl fmt::Display for CliError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            CliError::Error(msg) => write!(f, "{msg}"),
            CliError::Timeout(msg) => write!(f, "{msg}"),
        }
    }
}

impl CliError {
    pub fn exit_code(&self) -> i32 {
        match self {
            CliError::Error(_) => 1,
            CliError::Timeout(_) => 2,
        }
    }
}

// ---------------------------------------------------------------------------
// Output formatting
// ---------------------------------------------------------------------------

/// Controls how CLI commands format their output.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OutputFormat {
    /// Colored, aligned, human-readable (default for TTY).
    Human,
    /// One JSON object per line — JSONL (default for pipe).
    Json,
    /// Raw payload bytes only (`--payload-only`).
    Raw,
}

impl OutputFormat {
    /// Auto-detect format based on whether stdout is a terminal.
    /// Explicit `--json` or `--payload-only` flags override this.
    pub fn auto_detect(json_flag: bool, payload_only: bool) -> Self {
        if payload_only {
            return Self::Raw;
        }
        if json_flag {
            return Self::Json;
        }
        if std::io::IsTerminal::is_terminal(&std::io::stdout()) {
            Self::Human
        } else {
            Self::Json
        }
    }
}

// ---------------------------------------------------------------------------
// Topic / namespace parsing
// ---------------------------------------------------------------------------

/// Parse a `"namespace/topic"` string into its components.
///
/// If the input contains a `/`, everything before the first `/` is the
/// namespace and everything after is the topic. If there is no `/`, the
/// whole string is treated as the topic and the namespace is `None`
/// (must be supplied via `--namespace` flag or config).
pub fn parse_topic(input: &str) -> (Option<NamespaceId>, TopicName) {
    if let Some((ns, topic)) = input.split_once('/') {
        (
            Some(NamespaceId(ns.to_owned())),
            TopicName(topic.to_owned()),
        )
    } else {
        (None, TopicName(input.to_owned()))
    }
}

/// Resolve the effective namespace from the parsed topic, the `--namespace`
/// flag, and the CLI config file. Returns an error message if no namespace
/// can be determined.
pub fn resolve_namespace(
    parsed: Option<NamespaceId>,
    flag: Option<&str>,
    config: &CliConfig,
) -> Result<NamespaceId, String> {
    if let Some(ns) = parsed {
        return Ok(ns);
    }
    if let Some(ns) = flag {
        return Ok(NamespaceId(ns.to_owned()));
    }
    if let Some(ref ns) = config.namespace {
        return Ok(NamespaceId(ns.clone()));
    }
    Err("no namespace specified — use ns/topic syntax, --namespace flag, or config file".into())
}

// ---------------------------------------------------------------------------
// Color mode
// ---------------------------------------------------------------------------

/// Controls when colored output is used.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, clap::ValueEnum)]
pub enum ColorMode {
    /// Color when stdout is a TTY (default).
    #[default]
    Auto,
    /// Always produce colored output.
    Always,
    /// Never produce colored output.
    Never,
}

/// Apply the chosen color mode globally via `colored::control`.
pub fn apply_color_mode(mode: ColorMode) {
    match mode {
        ColorMode::Auto => {} // colored crate auto-detects by default
        ColorMode::Always => colored::control::set_override(true),
        ColorMode::Never => colored::control::set_override(false),
    }
}

// ---------------------------------------------------------------------------
// Global CLI options (the flags shared across non-serve commands)
// ---------------------------------------------------------------------------

/// Global flags shared by all non-serve subcommands.
///
/// Derived as `clap::Args` so it can be `#[command(flatten)]`-ed into the
/// top-level `Cli` struct. Clap's `args_conflicts_with_subcommands` ensures
/// these don't collide with `ServeArgs` in the no-subcommand (backward compat)
/// path.
#[derive(Debug, Clone, Default, clap::Args)]
pub struct GlobalOpts {
    /// Server address for TCP commands (host:port)
    #[arg(short = 's', long, env = "FAFB_SERVER", global = true)]
    pub server: Option<String>,

    /// HTTP server address for info/health/drain commands (host:port).
    /// Defaults to --server value if not set.
    #[arg(long, env = "FAFB_HTTP_SERVER", global = true)]
    pub http_server: Option<String>,

    /// API key for authentication
    #[arg(short = 'k', long, env = "FAFB_API_KEY", global = true)]
    pub api_key: Option<String>,

    /// Default namespace
    #[arg(short = 'n', long, global = true)]
    pub namespace: Option<String>,

    /// Command timeout in seconds
    #[arg(long, default_value = "30", global = true)]
    pub timeout: u64,

    /// JSON output for all commands
    #[arg(long, global = true)]
    pub json: bool,

    /// Verbose output (show protocol frames)
    #[arg(short = 'v', long, global = true)]
    pub verbose: bool,

    /// Use HTTP instead of TCP
    #[arg(long, global = true)]
    pub http: bool,

    /// Enable TLS
    #[arg(long, global = true)]
    pub tls: bool,

    /// CA certificate for TLS verification
    #[arg(long, global = true)]
    pub tls_ca: Option<String>,

    /// Client certificate for mTLS
    #[arg(long, global = true)]
    pub tls_cert: Option<String>,

    /// Client key for mTLS
    #[arg(long, global = true)]
    pub tls_key: Option<String>,

    /// Named profile from config file
    #[arg(long, global = true)]
    pub profile: Option<String>,

    /// When to use colored output [default: auto]
    #[arg(long, value_enum, default_value = "auto", global = true)]
    pub color: ColorMode,
}

// ---------------------------------------------------------------------------
// CLI config file (~/.fafb-q/config.toml)
// ---------------------------------------------------------------------------

/// Top-level CLI configuration file structure.
///
/// Lives at `~/.fafb-q/config.toml`. Entirely separate from the server
/// config — this stores connection profiles for the CLI client.
#[derive(Debug, Clone, Default, Deserialize)]
pub struct CliConfigFile {
    /// Default connection settings.
    #[serde(default)]
    pub default: CliProfile,
    /// Named profiles for multi-environment use.
    #[serde(default)]
    pub profiles: HashMap<String, CliProfile>,
}

/// A single connection profile.
#[derive(Debug, Clone, Default, Deserialize)]
pub struct CliProfile {
    /// Server address for TCP commands (e.g., `"localhost:9876"`).
    pub server: Option<String>,
    /// HTTP server address for info/health/drain (e.g., `"localhost:9877"`).
    /// Falls back to `server` if not set.
    pub http_server: Option<String>,
    /// API key for authentication.
    pub api_key: Option<String>,
    /// Read the API key from this environment variable instead of storing
    /// it directly.
    pub api_key_env: Option<String>,
    /// Default namespace.
    pub namespace: Option<String>,
    /// Enable TLS.
    #[serde(default)]
    pub tls: bool,
    /// CA certificate path for TLS verification.
    pub tls_ca: Option<String>,
    /// Client certificate path for mTLS.
    pub tls_cert: Option<String>,
    /// Client key path for mTLS.
    pub tls_key: Option<String>,
}

impl CliProfile {
    /// Merge `other` onto `self`: fields in `other` override fields in `self`.
    fn overlay(&self, other: &CliProfile) -> CliProfile {
        CliProfile {
            server: other.server.clone().or_else(|| self.server.clone()),
            http_server: other
                .http_server
                .clone()
                .or_else(|| self.http_server.clone()),
            api_key: other.api_key.clone().or_else(|| self.api_key.clone()),
            api_key_env: other
                .api_key_env
                .clone()
                .or_else(|| self.api_key_env.clone()),
            namespace: other.namespace.clone().or_else(|| self.namespace.clone()),
            tls: other.tls || self.tls,
            tls_ca: other.tls_ca.clone().or_else(|| self.tls_ca.clone()),
            tls_cert: other.tls_cert.clone().or_else(|| self.tls_cert.clone()),
            tls_key: other.tls_key.clone().or_else(|| self.tls_key.clone()),
        }
    }
}

/// Resolved CLI connection configuration after merging config file, env vars,
/// and CLI flags.
#[derive(Debug, Clone)]
pub struct CliConfig {
    /// TCP server address (for pub/sub/fetch).
    pub server: String,
    /// HTTP server address (for info/health/drain). Falls back to `server`.
    pub http_server: String,
    pub api_key: Option<String>,
    pub namespace: Option<String>,
    pub timeout_secs: u64,
    pub json: bool,
    pub verbose: bool,
    pub use_http: bool,
    pub tls: bool,
    pub tls_ca: Option<String>,
    pub tls_cert: Option<String>,
    pub tls_key: Option<String>,
}

impl Default for CliConfig {
    fn default() -> Self {
        Self {
            server: "localhost:9876".to_owned(),
            http_server: "localhost:9877".to_owned(),
            api_key: None,
            namespace: None,
            timeout_secs: 30,
            json: false,
            verbose: false,
            use_http: false,
            tls: false,
            tls_ca: None,
            tls_cert: None,
            tls_key: None,
        }
    }
}

/// Load the CLI config file from the default path (`~/.fafb-q/config.toml`).
///
/// Returns `CliConfigFile::default()` if the file does not exist — missing
/// config is not an error.
pub fn load_cli_config() -> CliConfigFile {
    let path = cli_config_path();
    match path {
        Some(p) if p.exists() => {
            let contents = match std::fs::read_to_string(&p) {
                Ok(c) => c,
                Err(e) => {
                    eprintln!("warning: failed to read CLI config {}: {e}", p.display());
                    return CliConfigFile::default();
                }
            };
            match toml::from_str(&contents) {
                Ok(cfg) => cfg,
                Err(e) => {
                    eprintln!("warning: failed to parse CLI config {}: {e}", p.display());
                    CliConfigFile::default()
                }
            }
        }
        _ => CliConfigFile::default(),
    }
}

/// Returns the path to the CLI config file (`~/.fafb-q/config.toml`).
fn cli_config_path() -> Option<PathBuf> {
    dirs_path().map(|d| d.join("config.toml"))
}

/// Returns the CLI config directory (`~/.fafb-q/`).
fn dirs_path() -> Option<PathBuf> {
    home_dir().map(|h| h.join(".fafb-q"))
}

/// Best-effort home directory lookup.
fn home_dir() -> Option<PathBuf> {
    std::env::var_os("HOME").map(PathBuf::from)
}

/// Build a resolved `CliConfig` by layering: config file < CLI flags.
///
/// When `--profile <name>` is specified, that profile's values are overlaid
/// on top of `[default]` before CLI flags are applied.
pub fn resolve_cli_config(opts: &GlobalOpts) -> CliConfig {
    let file_config = load_cli_config();

    // Start with [default], overlay [profiles.<name>] if --profile given.
    let profile = match &opts.profile {
        Some(name) => match file_config.profiles.get(name) {
            Some(named) => file_config.default.overlay(named),
            None => {
                eprintln!("warning: profile '{name}' not found in config, using defaults");
                file_config.default.clone()
            }
        },
        None => file_config.default.clone(),
    };

    // Resolve API key: flag > profile.api_key_env > profile.api_key
    let api_key = opts
        .api_key
        .clone()
        .or_else(|| {
            profile
                .api_key_env
                .as_ref()
                .and_then(|var| std::env::var(var).ok())
        })
        .or_else(|| profile.api_key.clone());

    let tcp_server = opts
        .server
        .clone()
        .or_else(|| profile.server.clone())
        .unwrap_or_else(|| "localhost:9876".to_owned());

    // HTTP server: flag > profile > fallback to TCP server address.
    let http_server = opts
        .http_server
        .clone()
        .or_else(|| profile.http_server.clone())
        .unwrap_or_else(|| tcp_server.clone());

    CliConfig {
        server: tcp_server,
        http_server,
        api_key,
        namespace: opts.namespace.clone().or_else(|| profile.namespace.clone()),
        timeout_secs: opts.timeout,
        json: opts.json,
        verbose: opts.verbose,
        use_http: opts.http,
        tls: opts.tls || profile.tls,
        tls_ca: opts.tls_ca.clone().or_else(|| profile.tls_ca.clone()),
        tls_cert: opts.tls_cert.clone().or_else(|| profile.tls_cert.clone()),
        tls_key: opts.tls_key.clone().or_else(|| profile.tls_key.clone()),
    }
}

// ---------------------------------------------------------------------------
// Shared client connection helper
// ---------------------------------------------------------------------------

/// Build a `ClientConfig` from a resolved `CliConfig` and namespace.
///
/// Shared by all commands that connect to a server. Returns `CliError` if
/// the API key is missing.
pub fn build_client_config(
    config: &CliConfig,
    namespace: NamespaceId,
) -> Result<ClientConfig, CliError> {
    let api_key = config
        .api_key
        .as_ref()
        .ok_or_else(|| {
            CliError::Error(
                "no API key specified — use --api-key, FAFB_API_KEY, or config file".into(),
            )
        })?
        .clone();

    Ok(ClientConfig {
        server: config.server.clone(),
        namespace,
        api_key: ApiKey(api_key),
        timeout: std::time::Duration::from_secs(config.timeout_secs),
        max_frame_size: 1_048_576,
        tls: ClientTlsConfig {
            enabled: config.tls,
            ca_cert_file: config.tls_ca.clone(),
            client_cert_file: config.tls_cert.clone(),
            client_key_file: config.tls_key.clone(),
            server_name: None,
        },
        verbose: config.verbose,
        ping_interval: std::time::Duration::from_secs(10),
    })
}

// ---------------------------------------------------------------------------
// Shared HTTP client helper
// ---------------------------------------------------------------------------

/// Build an HTTP base URL from a `CliConfig`.
///
/// Uses `config.http_server` (which falls back to `config.server` if not set).
/// Uses `https://` when TLS is enabled, `http://` otherwise.
pub fn http_base_url(config: &CliConfig) -> String {
    let scheme = if config.tls { "https" } else { "http" };
    format!("{scheme}://{}", config.http_server)
}

/// Build a `reqwest::Client` configured from `CliConfig`.
///
/// Handles TLS CA cert loading and timeout.
pub fn build_http_client(config: &CliConfig) -> Result<reqwest::Client, CliError> {
    let mut builder =
        reqwest::Client::builder().timeout(std::time::Duration::from_secs(config.timeout_secs));

    if config.tls {
        if let Some(ref ca_path) = config.tls_ca {
            let ca_bytes = std::fs::read(ca_path)
                .map_err(|e| CliError::Error(format!("failed to read CA cert {ca_path}: {e}")))?;
            let ca_cert = reqwest::Certificate::from_pem(&ca_bytes)
                .map_err(|e| CliError::Error(format!("failed to parse CA cert {ca_path}: {e}")))?;
            builder = builder.add_root_certificate(ca_cert);
        }

        if let (Some(cert_path), Some(key_path)) = (&config.tls_cert, &config.tls_key) {
            let mut id_pem = std::fs::read(cert_path).map_err(|e| {
                CliError::Error(format!("failed to read client cert {cert_path}: {e}"))
            })?;
            let key_pem = std::fs::read(key_path).map_err(|e| {
                CliError::Error(format!("failed to read client key {key_path}: {e}"))
            })?;
            id_pem.extend_from_slice(&key_pem);
            let identity = reqwest::Identity::from_pem(&id_pem)
                .map_err(|e| CliError::Error(format!("failed to build client identity: {e}")))?;
            builder = builder.identity(identity);
        }
    }

    builder
        .build()
        .map_err(|e| CliError::Error(format!("failed to build HTTP client: {e}")))
}

// ---------------------------------------------------------------------------
// Shared HTTP response helpers
// ---------------------------------------------------------------------------

/// Try to parse an HTTP response body as JSON. Falls back to wrapping the raw
/// text body in a `{"message": "..."}` object if JSON parsing fails.
pub async fn response_json(resp: reqwest::Response) -> Result<serde_json::Value, CliError> {
    let text = resp
        .text()
        .await
        .map_err(|e| CliError::Error(format!("failed to read response body: {e}")))?;

    match serde_json::from_str(&text) {
        Ok(v) => Ok(v),
        Err(_) => {
            // Not JSON — wrap raw text so callers can still extract a message.
            Ok(serde_json::json!({"message": text.trim()}))
        }
    }
}

/// Extract a human-readable error message from a JSON response body.
///
/// Tries common patterns: `error.message`, `message`, `error` (string).
pub fn extract_error_message(body: &serde_json::Value) -> String {
    body.pointer("/error/message")
        .and_then(|v| v.as_str())
        .or_else(|| body.get("message").and_then(|v| v.as_str()))
        .or_else(|| body.get("error").and_then(|v| v.as_str()))
        .unwrap_or("unknown error")
        .to_string()
}

// ---------------------------------------------------------------------------
// Shared formatting helpers
// ---------------------------------------------------------------------------

/// Format a byte count as a human-readable string (B, KB, MB, GB).
pub fn format_bytes(bytes: u64) -> String {
    const KB: u64 = 1024;
    const MB: u64 = 1024 * 1024;
    const GB: u64 = 1024 * 1024 * 1024;

    if bytes >= GB {
        format!("{:.1} GB", bytes as f64 / GB as f64)
    } else if bytes >= MB {
        format!("{:.1} MB", bytes as f64 / MB as f64)
    } else if bytes >= KB {
        format!("{:.1} KB", bytes as f64 / KB as f64)
    } else {
        format!("{bytes} B")
    }
}

// ---------------------------------------------------------------------------
// Typed command argument structs
// ---------------------------------------------------------------------------

/// Arguments for the `pub` command.
pub struct PubArgs {
    /// Topic in `ns/topic` format.
    pub topic: String,
    /// Message payload (reads from stdin if None).
    pub payload: Option<String>,
    /// Read payload from file.
    pub file: Option<String>,
    /// One message per line from stdin.
    pub lines: bool,
    /// Message TTL in seconds (0 = no expiry).
    pub ttl_secs: Option<u64>,
    /// Headers as `"Key: Value"` strings.
    pub headers: Vec<String>,
    /// Don't wait for PublishAck (fire-and-forget).
    pub no_ack: bool,
}

/// Arguments for the `sub` command.
pub struct SubArgs {
    /// Topic in `ns/topic` format (supports wildcards).
    pub topic: String,
    /// Subscribe from a specific offset.
    pub from: Option<u64>,
    /// Subscribe from a timestamp (ISO8601 or unix millis).
    pub since: Option<String>,
    /// Consumer group name.
    pub group: Option<String>,
    /// Receive N messages then exit.
    pub count: Option<u64>,
    /// Don't send Ack (receive only, for monitoring).
    pub no_ack: bool,
    /// Output only payload bytes.
    pub payload_only: bool,
}

/// Arguments for the `fetch` command.
pub struct FetchArgs {
    /// Topic in `ns/topic` format.
    pub topic: String,
    /// Start offset (negative = relative from head: -1 = last message).
    pub from: Option<i64>,
    /// Start timestamp (ISO8601).
    pub since: Option<String>,
    /// Fetch last N messages (shorthand for `--from -N --count N`).
    pub tail: Option<u32>,
    /// Max messages per batch (default 100).
    pub count: u32,
    /// Fetch all messages (paginate automatically).
    pub all: bool,
    /// Output only payload bytes.
    pub payload_only: bool,
}

/// Arguments for the `info` command.
pub struct InfoArgs {
    /// Namespace or `ns/topic`.
    pub target: Option<String>,
}

/// Arguments for the `drain` command.
pub struct DrainArgs {
    /// Admin API key (required for drain endpoints).
    pub admin_key: Option<String>,
    /// Custom drain timeout in seconds.
    pub timeout: Option<u64>,
    /// Check drain status instead of triggering drain.
    pub status: bool,
    /// Drain reason string.
    pub reason: Option<String>,
    /// Wait for drain to complete (poll status until done or timeout).
    pub wait: bool,
}

/// Arguments for the `health` command.
pub struct HealthArgs {
    /// Use liveness probe instead of readiness probe.
    pub live: bool,
}

/// Config sub-action (used as clap `Subcommand` in main.rs).
#[derive(Debug, Clone, clap::Subcommand)]
pub enum ConfigAction {
    /// Validate a config file
    Validate {
        /// Path to the config file to validate
        #[arg(long, short)]
        file: String,
    },
    /// Dump resolved config (with API keys redacted)
    Dump {
        /// Path to config file (searches default paths if omitted)
        #[arg(long, short)]
        file: Option<String>,
        /// Admin API key — when provided, fetches config from a running server
        /// via GET /admin/config instead of reading a local file
        #[arg(long, env = "FAFB_ADMIN_API_KEY")]
        admin_key: Option<String>,
    },
    /// Show example config
    Example,
}

/// Bench sub-action (used as clap `Subcommand` in main.rs).
#[derive(Debug, Clone, clap::Subcommand)]
pub enum BenchAction {
    /// Publish benchmark
    Pub {
        /// Topic in ns/topic format
        topic: String,
        /// Number of messages to publish
        #[arg(long, default_value = "10000")]
        messages: u64,
        /// Payload size in bytes
        #[arg(long, default_value = "256")]
        payload_size: usize,
        /// Number of parallel TCP connections
        #[arg(long, default_value = "1")]
        concurrency: u32,
        /// Fire-and-forget (don't wait for PublishAck) — measures raw throughput
        #[arg(long)]
        no_ack: bool,
    },
    /// Subscribe benchmark (consume as fast as possible)
    Sub {
        /// Topic in ns/topic format
        topic: String,
        /// Consumer group name
        #[arg(short = 'g', long)]
        group: Option<String>,
        /// Subscribe from a specific offset (default: head)
        #[arg(long)]
        from: Option<u64>,
        /// Stop after consuming N messages
        #[arg(long)]
        count: Option<u64>,
    },
    /// End-to-end latency benchmark
    Latency {
        /// Topic in ns/topic format
        topic: String,
        /// Number of messages to measure
        #[arg(long, default_value = "1000")]
        messages: u64,
        /// Number of warmup messages to discard before measuring
        #[arg(long, default_value = "0")]
        warmup: u64,
    },
}

// ---------------------------------------------------------------------------
// Output formatting helpers
// ---------------------------------------------------------------------------

/// Parse a `"Key: Value"` header string into `(key_bytes, value_bytes)`.
///
/// Returns an error string if the header is malformed.
pub fn parse_header(s: &str) -> Result<(Vec<u8>, Vec<u8>), String> {
    let (key, value) = s
        .split_once(':')
        .ok_or_else(|| format!("invalid header (missing ':'): {s}"))?;
    let key = key.trim();
    let value = value.trim();
    if key.is_empty() {
        return Err(format!("empty header key in: {s}"));
    }
    Ok((key.as_bytes().to_vec(), value.as_bytes().to_vec()))
}

/// Parse an ISO8601 timestamp string to millis since epoch.
///
/// Accepts:
/// - Unix millis as a plain integer (e.g. `"1711185330000"`)
/// - `YYYY-MM-DDTHH:MM:SSZ`
/// - `YYYY-MM-DDTHH:MM:SS.fffZ` (fractional seconds, truncated to millis)
/// - `YYYY-MM-DDTHH:MM:SS+00:00` (UTC offset)
/// - `YYYY-MM-DD` (date only, midnight UTC)
pub fn parse_timestamp(s: &str) -> Result<u64, String> {
    // Try plain integer first (unix millis).
    if let Ok(ms) = s.parse::<u64>() {
        return Ok(ms);
    }

    let s = s.trim();
    let s = s
        .strip_suffix('Z')
        .or_else(|| s.strip_suffix("+00:00"))
        .unwrap_or(s);

    // Date-only: YYYY-MM-DD → midnight UTC.
    if !s.contains('T') {
        let date_parts = parse_date_components(s)?;
        let days = civil_to_days(date_parts.0, date_parts.1, date_parts.2);
        return Ok(days * 86400 * 1000);
    }

    // Full datetime: YYYY-MM-DDTHH:MM:SS[.fff]
    let parts: Vec<&str> = s.split('T').collect();
    if parts.len() != 2 {
        return Err(format!(
            "invalid timestamp (expected YYYY-MM-DDTHH:MM:SSZ): {s}"
        ));
    }

    let (year, month, day) = parse_date_components(parts[0])?;

    // Split time part, handling optional fractional seconds.
    let (time_str, frac_millis) = if let Some((main, frac)) = parts[1].split_once('.') {
        // Parse fractional seconds: take up to 3 digits as millis.
        let frac_trimmed = if frac.len() > 3 { &frac[..3] } else { frac };
        let millis: u64 = frac_trimmed
            .parse()
            .map_err(|_| format!("invalid fractional seconds: .{frac}"))?;
        // Pad to 3 digits if fewer (e.g., ".1" → 100ms, ".12" → 120ms).
        let millis = millis * 10u64.pow(3 - frac_trimmed.len() as u32);
        (main, millis)
    } else {
        (parts[1], 0)
    };

    let time_parts: Vec<u64> = time_str
        .split(':')
        .map(|p| {
            p.parse::<u64>()
                .map_err(|_| format!("invalid time component: {p}"))
        })
        .collect::<Result<_, _>>()?;

    if time_parts.len() != 3 {
        return Err(format!("invalid timestamp format: {s}"));
    }

    let (hour, min, sec) = (time_parts[0], time_parts[1], time_parts[2]);
    let days = civil_to_days(year, month, day);
    let total_secs = days * 86400 + hour * 3600 + min * 60 + sec;
    Ok(total_secs * 1000 + frac_millis)
}

/// Parse "YYYY-MM-DD" into (year, month, day).
fn parse_date_components(s: &str) -> Result<(u64, u64, u64), String> {
    let parts: Vec<u64> = s
        .split('-')
        .map(|p| {
            p.parse::<u64>()
                .map_err(|_| format!("invalid date component: {p}"))
        })
        .collect::<Result<_, _>>()?;

    if parts.len() != 3 {
        return Err(format!("invalid date format: {s}"));
    }
    Ok((parts[0], parts[1], parts[2]))
}

/// Convert a civil date to days since Unix epoch (Hinnant algorithm).
fn civil_to_days(year: u64, month: u64, day: u64) -> u64 {
    let y = if month <= 2 { year - 1 } else { year };
    let era = y / 400;
    let yoe = y - era * 400;
    let m = if month > 2 { month - 3 } else { month + 9 };
    let doy = (153 * m + 2) / 5 + day - 1;
    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
    era * 146097 + doe - 719468
}

// ---------------------------------------------------------------------------
// Shared message formatting
// ---------------------------------------------------------------------------

/// Parameters for formatting a delivered or fetched message.
pub struct MessageDisplay<'a> {
    pub topic: &'a str,
    pub offset: u64,
    pub timestamp_millis: u64,
    pub ttl_ms: u32,
    pub message_id: Option<u64>,
    pub headers: &'a [crate::types::Header],
    pub payload: &'a [u8],
}

/// Format and print a single message to stdout.
///
/// Shared by `fetch` and `subscribe` commands to avoid duplicated formatting
/// logic. Flushes stdout after each message for real-time piping.
pub fn print_message(format: OutputFormat, msg: &MessageDisplay<'_>) {
    use std::io::Write;

    match format {
        OutputFormat::Human => {
            let ts = crate::types::millis_to_rfc3339(msg.timestamp_millis);
            println!(
                "{} {} {}",
                format!("[{}]", msg.topic).cyan(),
                format!("offset={}", msg.offset).dimmed(),
                format!("ts={ts}").dimmed()
            );
            match std::str::from_utf8(msg.payload) {
                Ok(s) => println!("{s}"),
                Err(_) => println!(
                    "{}",
                    format!("<binary {} bytes>", msg.payload.len()).dimmed()
                ),
            }
            for h in msg.headers {
                let key = String::from_utf8_lossy(&h.key);
                let val = String::from_utf8_lossy(&h.value);
                println!("  {}: {}", key.dimmed(), val);
            }
            println!();
            let _ = std::io::stdout().flush();
        }
        OutputFormat::Json => {
            let payload_str = String::from_utf8(msg.payload.to_vec()).ok();
            let mut obj = serde_json::json!({
                "topic": msg.topic,
                "offset": msg.offset,
                "timestamp_millis": msg.timestamp_millis,
            });
            if let Some(id) = msg.message_id {
                obj["message_id"] = serde_json::Value::Number(id.into());
            }
            if let Some(ref s) = payload_str {
                obj["payload"] = serde_json::Value::String(s.clone());
            } else {
                obj["payload_base64"] = serde_json::Value::String(base64_encode(msg.payload));
            }
            if !msg.headers.is_empty() {
                let hdrs: Vec<serde_json::Value> = msg
                    .headers
                    .iter()
                    .map(|h| {
                        let key = String::from_utf8_lossy(&h.key).into_owned();
                        match String::from_utf8(h.value.clone()) {
                            Ok(v) => serde_json::json!({"key": key, "value": v}),
                            Err(_) => {
                                serde_json::json!({"key": key, "value_base64": base64_encode(&h.value)})
                            }
                        }
                    })
                    .collect();
                obj["headers"] = serde_json::Value::Array(hdrs);
            }
            if msg.ttl_ms > 0 {
                obj["ttl_ms"] = serde_json::Value::Number(msg.ttl_ms.into());
            }
            println!("{obj}");
            let _ = std::io::stdout().flush();
        }
        OutputFormat::Raw => {
            let stdout = std::io::stdout();
            let mut lock = stdout.lock();
            let _ = lock.write_all(msg.payload);
            let _ = lock.write_all(b"\n");
            let _ = lock.flush();
        }
    }
}

/// Simple base64 encoding (no external dependency).
pub fn base64_encode(data: &[u8]) -> String {
    use std::fmt::Write;
    const CHARS: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";

    let mut result = String::with_capacity(data.len().div_ceil(3) * 4);
    for chunk in data.chunks(3) {
        let b0 = chunk[0] as u32;
        let b1 = if chunk.len() > 1 { chunk[1] as u32 } else { 0 };
        let b2 = if chunk.len() > 2 { chunk[2] as u32 } else { 0 };
        let n = (b0 << 16) | (b1 << 8) | b2;

        let _ = write!(result, "{}", CHARS[((n >> 18) & 0x3F) as usize] as char);
        let _ = write!(result, "{}", CHARS[((n >> 12) & 0x3F) as usize] as char);
        if chunk.len() > 1 {
            let _ = write!(result, "{}", CHARS[((n >> 6) & 0x3F) as usize] as char);
        } else {
            result.push('=');
        }
        if chunk.len() > 2 {
            let _ = write!(result, "{}", CHARS[(n & 0x3F) as usize] as char);
        } else {
            result.push('=');
        }
    }
    result
}

// ---------------------------------------------------------------------------
// Command runners — implemented commands delegate to submodules,
// unimplemented commands return an error.
// ---------------------------------------------------------------------------

pub async fn run_publish(config: CliConfig, args: PubArgs) -> Result<(), CliError> {
    publish::run(config, args).await
}

pub async fn run_subscribe(config: CliConfig, args: SubArgs) -> Result<(), CliError> {
    if config.use_http {
        return Err(CliError::Error(
            "subscribe requires a persistent TCP connection and is not supported over HTTP — remove --http flag".into(),
        ));
    }
    subscribe::run(config, args).await
}

pub async fn run_fetch(config: CliConfig, args: FetchArgs) -> Result<(), CliError> {
    fetch::run(config, args).await
}

pub async fn run_info(config: CliConfig, args: InfoArgs) -> Result<(), CliError> {
    info::run(config, args).await
}

pub async fn run_drain(config: CliConfig, args: DrainArgs) -> Result<(), CliError> {
    drain::run(config, args).await
}

pub async fn run_health(config: CliConfig, args: HealthArgs) -> Result<(), CliError> {
    health::run(config, args).await
}

pub async fn run_config_cmd(cli_config: CliConfig, action: ConfigAction) -> Result<(), CliError> {
    config::run(cli_config, action).await
}

pub async fn run_bench(config: CliConfig, action: BenchAction) -> Result<(), CliError> {
    bench::run(config, action).await
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parse_topic_with_namespace() {
        let (ns, topic) = parse_topic("myns/orders");
        assert_eq!(ns, Some(NamespaceId("myns".into())));
        assert_eq!(topic, TopicName("orders".into()));
    }

    #[test]
    fn parse_topic_without_namespace() {
        let (ns, topic) = parse_topic("orders");
        assert_eq!(ns, None);
        assert_eq!(topic, TopicName("orders".into()));
    }

    #[test]
    fn parse_topic_with_nested_path() {
        let (ns, topic) = parse_topic("myns/orders.us.east");
        assert_eq!(ns, Some(NamespaceId("myns".into())));
        assert_eq!(topic, TopicName("orders.us.east".into()));
    }

    #[test]
    fn output_format_auto_detect_json_flag() {
        assert_eq!(OutputFormat::auto_detect(true, false), OutputFormat::Json);
    }

    #[test]
    fn output_format_auto_detect_payload_only() {
        assert_eq!(OutputFormat::auto_detect(false, true), OutputFormat::Raw);
        // payload_only takes precedence over json
        assert_eq!(OutputFormat::auto_detect(true, true), OutputFormat::Raw);
    }

    #[test]
    fn resolve_namespace_from_parsed() {
        let cfg = CliConfig::default();
        let result = resolve_namespace(Some(NamespaceId("parsed".into())), Some("flag"), &cfg);
        assert_eq!(result.unwrap(), NamespaceId("parsed".into()));
    }

    #[test]
    fn resolve_namespace_from_flag() {
        let cfg = CliConfig::default();
        let result = resolve_namespace(None, Some("flag"), &cfg);
        assert_eq!(result.unwrap(), NamespaceId("flag".into()));
    }

    #[test]
    fn resolve_namespace_from_config() {
        let cfg = CliConfig {
            namespace: Some("fromconfig".into()),
            ..Default::default()
        };
        let result = resolve_namespace(None, None, &cfg);
        assert_eq!(result.unwrap(), NamespaceId("fromconfig".into()));
    }

    #[test]
    fn resolve_namespace_missing() {
        let cfg = CliConfig::default();
        let result = resolve_namespace(None, None, &cfg);
        assert!(result.is_err());
    }

    #[test]
    fn cli_config_file_missing_is_ok() {
        // Should not panic — returns defaults
        let cfg = load_cli_config();
        assert!(cfg.default.server.is_none());
    }

    #[test]
    fn cli_profile_deserialize() {
        let toml_str = r#"
            [default]
            server = "myhost:1234"
            api_key = "sk_test"
            namespace = "myns"
            tls = true
        "#;
        let cfg: CliConfigFile = toml::from_str(toml_str).unwrap();
        assert_eq!(cfg.default.server.as_deref(), Some("myhost:1234"));
        assert_eq!(cfg.default.api_key.as_deref(), Some("sk_test"));
        assert_eq!(cfg.default.namespace.as_deref(), Some("myns"));
        assert!(cfg.default.tls);
    }

    #[test]
    fn cli_named_profile_deserialize() {
        let toml_str = r#"
            [default]
            server = "localhost:9876"
            namespace = "dev"

            [profiles.production]
            server = "prod.internal:9876"
            api_key_env = "FAFB_PROD_KEY"
            namespace = "prod"
            tls = true
            tls_ca = "/etc/ssl/ca.pem"

            [profiles.staging]
            server = "staging.internal:9876"
            api_key = "sk_staging_xyz"
            namespace = "staging"
        "#;
        let cfg: CliConfigFile = toml::from_str(toml_str).unwrap();
        assert_eq!(cfg.profiles.len(), 2);

        let prod = cfg.profiles.get("production").unwrap();
        assert_eq!(prod.server.as_deref(), Some("prod.internal:9876"));
        assert_eq!(prod.api_key_env.as_deref(), Some("FAFB_PROD_KEY"));
        assert!(prod.tls);

        let staging = cfg.profiles.get("staging").unwrap();
        assert_eq!(staging.namespace.as_deref(), Some("staging"));
    }

    #[test]
    fn profile_overlay_merges_fields() {
        let base = CliProfile {
            server: Some("base:1234".into()),
            api_key: Some("base_key".into()),
            namespace: Some("base_ns".into()),
            tls: false,
            ..Default::default()
        };
        let overlay = CliProfile {
            server: Some("overlay:5678".into()),
            tls: true,
            ..Default::default()
        };
        let merged = base.overlay(&overlay);
        assert_eq!(merged.server.as_deref(), Some("overlay:5678"));
        assert_eq!(merged.api_key.as_deref(), Some("base_key")); // kept from base
        assert_eq!(merged.namespace.as_deref(), Some("base_ns")); // kept from base
        assert!(merged.tls); // overridden
    }

    #[test]
    fn resolve_config_uses_defaults_when_no_opts() {
        let opts = GlobalOpts {
            timeout: 30,
            ..Default::default()
        };
        let cfg = resolve_cli_config(&opts);
        assert_eq!(cfg.server, "localhost:9876");
        assert!(cfg.api_key.is_none());
        assert_eq!(cfg.timeout_secs, 30);
    }

    #[test]
    fn resolve_config_cli_flags_override() {
        let opts = GlobalOpts {
            server: Some("custom:1111".into()),
            api_key: Some("my_key".into()),
            namespace: Some("myns".into()),
            timeout: 60,
            json: true,
            verbose: true,
            http: true,
            tls: true,
            ..Default::default()
        };
        let cfg = resolve_cli_config(&opts);
        assert_eq!(cfg.server, "custom:1111");
        assert_eq!(cfg.api_key.as_deref(), Some("my_key"));
        assert_eq!(cfg.namespace.as_deref(), Some("myns"));
        assert_eq!(cfg.timeout_secs, 60);
        assert!(cfg.json);
        assert!(cfg.verbose);
        assert!(cfg.use_http);
        assert!(cfg.tls);
    }

    // Header parsing tests

    #[test]
    fn parse_header_valid() {
        let (k, v) = parse_header("Content-Type: application/json").unwrap();
        assert_eq!(k, b"Content-Type");
        assert_eq!(v, b"application/json");
    }

    #[test]
    fn parse_header_extra_colons() {
        let (k, v) = parse_header("X-Data: key:val:123").unwrap();
        assert_eq!(k, b"X-Data");
        assert_eq!(v, b"key:val:123");
    }

    #[test]
    fn parse_header_no_colon() {
        assert!(parse_header("no-colon-here").is_err());
    }

    #[test]
    fn parse_header_empty_key() {
        assert!(parse_header(": value").is_err());
    }

    // Timestamp parsing tests

    #[test]
    fn parse_timestamp_unix_millis() {
        assert_eq!(parse_timestamp("1711185330000").unwrap(), 1711185330000);
    }

    #[test]
    fn parse_timestamp_iso8601() {
        let ts = parse_timestamp("2026-03-23T10:15:30Z").unwrap();
        let year_2026_start = 1767225600000u64;
        assert!(ts > year_2026_start);
    }

    #[test]
    fn parse_timestamp_iso8601_with_offset() {
        let ts = parse_timestamp("2026-03-23T10:15:30+00:00").unwrap();
        let ts_z = parse_timestamp("2026-03-23T10:15:30Z").unwrap();
        assert_eq!(ts, ts_z);
    }

    #[test]
    fn parse_timestamp_fractional_seconds() {
        let ts_whole = parse_timestamp("2026-03-23T10:15:30Z").unwrap();
        let ts_frac = parse_timestamp("2026-03-23T10:15:30.500Z").unwrap();
        assert_eq!(ts_frac, ts_whole + 500);

        // Single digit fraction: .1 = 100ms
        let ts_frac1 = parse_timestamp("2026-03-23T10:15:30.1Z").unwrap();
        assert_eq!(ts_frac1, ts_whole + 100);

        // Two digit fraction: .12 = 120ms
        let ts_frac2 = parse_timestamp("2026-03-23T10:15:30.12Z").unwrap();
        assert_eq!(ts_frac2, ts_whole + 120);

        // More than 3 digits: truncated to millis
        let ts_frac4 = parse_timestamp("2026-03-23T10:15:30.1234Z").unwrap();
        assert_eq!(ts_frac4, ts_whole + 123);
    }

    #[test]
    fn parse_timestamp_date_only() {
        let ts = parse_timestamp("2026-03-23").unwrap();
        let ts_midnight = parse_timestamp("2026-03-23T00:00:00Z").unwrap();
        assert_eq!(ts, ts_midnight);
    }

    #[test]
    fn parse_timestamp_invalid() {
        assert!(parse_timestamp("not-a-timestamp").is_err());
    }

    // build_client_config tests

    #[test]
    fn build_client_config_success() {
        let cfg = CliConfig {
            api_key: Some("test-key".into()),
            server: "myhost:1234".into(),
            timeout_secs: 15,
            tls: true,
            tls_ca: Some("/ca.pem".into()),
            ..Default::default()
        };
        let cc = build_client_config(&cfg, NamespaceId("ns".into())).expect("should build config");
        assert_eq!(cc.server, "myhost:1234");
        assert_eq!(cc.api_key, ApiKey("test-key".into()));
        assert_eq!(cc.namespace, NamespaceId("ns".into()));
        assert_eq!(cc.timeout, std::time::Duration::from_secs(15));
        assert!(cc.tls.enabled);
        assert_eq!(cc.tls.ca_cert_file.as_deref(), Some("/ca.pem"));
    }

    #[test]
    fn build_client_config_missing_api_key() {
        let cfg = CliConfig::default();
        let result = build_client_config(&cfg, NamespaceId("ns".into()));
        assert!(result.is_err());
    }

    // http_base_url tests

    #[test]
    fn http_base_url_uses_http_server() {
        let cfg = CliConfig {
            server: "tcp-host:9876".into(),
            http_server: "http-host:9877".into(),
            ..Default::default()
        };
        assert_eq!(http_base_url(&cfg), "http://http-host:9877");
    }

    #[test]
    fn http_base_url_uses_https_when_tls() {
        let cfg = CliConfig {
            http_server: "secure:443".into(),
            tls: true,
            ..Default::default()
        };
        assert_eq!(http_base_url(&cfg), "https://secure:443");
    }

    // resolve_cli_config http_server tests

    #[test]
    fn resolve_config_http_server_falls_back_to_server() {
        let opts = GlobalOpts {
            server: Some("myhost:9876".into()),
            timeout: 30,
            ..Default::default()
        };
        let cfg = resolve_cli_config(&opts);
        assert_eq!(cfg.server, "myhost:9876");
        assert_eq!(cfg.http_server, "myhost:9876");
    }

    #[test]
    fn resolve_config_http_server_explicit() {
        let opts = GlobalOpts {
            server: Some("tcp:9876".into()),
            http_server: Some("http:9877".into()),
            timeout: 30,
            ..Default::default()
        };
        let cfg = resolve_cli_config(&opts);
        assert_eq!(cfg.server, "tcp:9876");
        assert_eq!(cfg.http_server, "http:9877");
    }

    // extract_error_message tests

    #[test]
    fn extract_error_message_nested() {
        let body = serde_json::json!({"error": {"message": "not found"}});
        assert_eq!(extract_error_message(&body), "not found");
    }

    #[test]
    fn extract_error_message_flat() {
        let body = serde_json::json!({"message": "bad request"});
        assert_eq!(extract_error_message(&body), "bad request");
    }

    #[test]
    fn extract_error_message_string_error() {
        let body = serde_json::json!({"error": "something went wrong"});
        assert_eq!(extract_error_message(&body), "something went wrong");
    }

    #[test]
    fn extract_error_message_unknown() {
        let body = serde_json::json!({"foo": "bar"});
        assert_eq!(extract_error_message(&body), "unknown error");
    }

    // format_bytes tests

    #[test]
    fn format_bytes_values() {
        assert_eq!(format_bytes(0), "0 B");
        assert_eq!(format_bytes(512), "512 B");
        assert_eq!(format_bytes(1024), "1.0 KB");
        assert_eq!(format_bytes(1_048_576), "1.0 MB");
        assert_eq!(format_bytes(1_073_741_824), "1.0 GB");
        assert_eq!(format_bytes(2_684_354_560), "2.5 GB");
    }

    // profile http_server overlay test

    #[test]
    fn profile_overlay_includes_http_server() {
        let base = CliProfile {
            server: Some("base:1234".into()),
            http_server: Some("base-http:5678".into()),
            ..Default::default()
        };
        let overlay = CliProfile {
            http_server: Some("overlay-http:9999".into()),
            ..Default::default()
        };
        let merged = base.overlay(&overlay);
        assert_eq!(merged.server.as_deref(), Some("base:1234")); // kept
        assert_eq!(merged.http_server.as_deref(), Some("overlay-http:9999")); // overridden
    }

    // Profile selection tests (Phase 23.6)

    #[test]
    fn resolve_config_profile_overrides_defaults() {
        // Create a temp config file with a named profile.
        let dir = tempfile::tempdir().unwrap();
        let config_path = dir.path().join("config.toml");
        std::fs::write(
            &config_path,
            r#"
[default]
server = "default:9876"
api_key = "default_key"
namespace = "default_ns"

[profiles.prod]
server = "prod:9876"
api_key = "prod_key"
namespace = "prod_ns"
tls = true
"#,
        )
        .unwrap();

        // Parse and overlay.
        let contents = std::fs::read_to_string(&config_path).unwrap();
        let file_config: CliConfigFile = toml::from_str(&contents).unwrap();

        let prod = file_config.profiles.get("prod").unwrap();
        let merged = file_config.default.overlay(prod);

        assert_eq!(merged.server.as_deref(), Some("prod:9876"));
        assert_eq!(merged.api_key.as_deref(), Some("prod_key"));
        assert_eq!(merged.namespace.as_deref(), Some("prod_ns"));
        assert!(merged.tls);
    }

    #[test]
    fn resolve_config_profile_inherits_from_defaults() {
        let toml_str = r#"
[default]
server = "default:9876"
api_key = "default_key"

[profiles.staging]
namespace = "staging_ns"
"#;
        let file_config: CliConfigFile = toml::from_str(toml_str).unwrap();
        let staging = file_config.profiles.get("staging").unwrap();
        let merged = file_config.default.overlay(staging);

        // server and api_key inherited from default.
        assert_eq!(merged.server.as_deref(), Some("default:9876"));
        assert_eq!(merged.api_key.as_deref(), Some("default_key"));
        // namespace overridden by profile.
        assert_eq!(merged.namespace.as_deref(), Some("staging_ns"));
    }

    #[test]
    fn resolve_config_api_key_env_resolution() {
        let profile = CliProfile {
            api_key_env: Some("TEST_FAFB_KEY_12345".into()),
            ..Default::default()
        };

        // When the env var is NOT set, api_key_env resolves to None.
        let resolved = profile
            .api_key_env
            .as_ref()
            .and_then(|var| std::env::var(var).ok());
        assert!(resolved.is_none());

        // When the env var IS set, it resolves.
        // SAFETY: This test is single-threaded and the env var is cleaned up.
        unsafe {
            std::env::set_var("TEST_FAFB_KEY_12345", "env_secret");
        }
        let resolved = profile
            .api_key_env
            .as_ref()
            .and_then(|var| std::env::var(var).ok());
        assert_eq!(resolved.as_deref(), Some("env_secret"));
        unsafe {
            std::env::remove_var("TEST_FAFB_KEY_12345");
        }
    }

    #[test]
    fn output_format_auto_detect_tty_fallback() {
        // When not in a TTY (test harness), should default to Json.
        let format = OutputFormat::auto_detect(false, false);
        // Test output is typically piped, so this should be Json.
        assert_eq!(format, OutputFormat::Json);
    }

    #[test]
    fn build_client_config_includes_verbose() {
        let cfg = CliConfig {
            api_key: Some("key".into()),
            verbose: true,
            ..Default::default()
        };
        let cc = build_client_config(&cfg, NamespaceId("ns".into())).unwrap();
        assert!(cc.verbose);

        let cfg2 = CliConfig {
            api_key: Some("key".into()),
            ..Default::default()
        };
        let cc2 = build_client_config(&cfg2, NamespaceId("ns".into())).unwrap();
        assert!(!cc2.verbose);
    }
}
