//! `fafb-q pub` command — publish messages to a topic.
//!
//! Supports both TCP (default) and HTTP (`--http`) transport modes.

use std::io::IsTerminal;

use bytes::Bytes;

use colored::Colorize;

use crate::cli::{
    CliConfig, CliError, OutputFormat, PubArgs, build_client_config, build_http_client,
    extract_error_message, http_base_url, parse_header, parse_topic, resolve_namespace,
    response_json,
};
use crate::client::FafbClient;
use crate::types::{Header, TopicName};

/// Run the `pub` command.
pub async fn run(config: CliConfig, args: PubArgs) -> Result<(), CliError> {
    let format = OutputFormat::auto_detect(config.json, false);

    // Parse topic and resolve namespace.
    let (parsed_ns, topic) = parse_topic(&args.topic);
    let namespace = resolve_namespace(parsed_ns, None, &config).map_err(CliError::Error)?;

    // Parse headers.
    let headers: Vec<Header> = args
        .headers
        .iter()
        .map(|h| parse_header(h).map(|(k, v)| Header { key: k, value: v }))
        .collect::<Result<Vec<_>, _>>()
        .map_err(CliError::Error)?;

    // TTL in seconds (0 = no expiry).
    let ttl_secs = args.ttl_secs.unwrap_or(0);

    if config.use_http {
        return run_http(
            config,
            &namespace.0,
            &topic,
            &headers,
            ttl_secs,
            &args,
            format,
        )
        .await;
    }

    // TCP mode.
    let ttl_ms = ttl_secs.saturating_mul(1000) as u32;

    // Connect.
    let client_config = build_client_config(&config, namespace)?;

    let mut client = FafbClient::connect(&client_config)
        .await
        .map_err(|e| CliError::Error(format!("{e}")))?;

    // Determine payload source and publish.
    let result = if args.lines {
        publish_lines(&mut client, &topic, &headers, ttl_ms, args.no_ack, format).await
    } else {
        let payload = resolve_payload(&args)?;
        publish_single(
            &mut client,
            &topic,
            payload,
            &headers,
            ttl_ms,
            args.no_ack,
            format,
        )
        .await
    };

    // Graceful close (best-effort).
    let _ = client.close().await;
    result
}

// ---------------------------------------------------------------------------
// HTTP publish mode
// ---------------------------------------------------------------------------

async fn run_http(
    config: CliConfig,
    namespace: &str,
    topic: &TopicName,
    headers: &[Header],
    ttl_secs: u64,
    args: &PubArgs,
    format: OutputFormat,
) -> Result<(), CliError> {
    let http_client = build_http_client(&config)?;
    let base = http_base_url(&config);
    let api_key = config.api_key.as_deref().ok_or_else(|| {
        CliError::Error("no API key specified — use --api-key, FAFB_API_KEY, or config file".into())
    })?;

    let ttl_query = if ttl_secs > 0 {
        format!("?ttl={ttl_secs}")
    } else {
        String::new()
    };
    let url = format!(
        "{base}/v1/{namespace}/{topic}/publish{ttl_query}",
        topic = topic.0
    );

    if args.lines {
        return publish_lines_http(&http_client, &url, api_key, headers, args.no_ack, format).await;
    }

    let payload = resolve_payload(args)?;
    publish_single_http(
        &http_client,
        &url,
        api_key,
        headers,
        payload,
        args.no_ack,
        format,
    )
    .await
}

async fn publish_single_http(
    client: &reqwest::Client,
    url: &str,
    api_key: &str,
    headers: &[Header],
    payload: Bytes,
    no_ack: bool,
    format: OutputFormat,
) -> Result<(), CliError> {
    let mut req = client.post(url).bearer_auth(api_key);

    // Add X-Fafb-* headers for message headers.
    for h in headers {
        let key = format!("X-Fafb-{}", String::from_utf8_lossy(&h.key));
        let val = String::from_utf8_lossy(&h.value).into_owned();
        req = req.header(&key, &val);
    }

    let resp = req
        .body(payload.to_vec())
        .send()
        .await
        .map_err(|e| CliError::Error(format!("HTTP request failed: {e}")))?;

    let status = resp.status();
    let body = response_json(resp).await?;

    if !status.is_success() {
        return Err(CliError::Error(format!(
            "publish failed ({status}): {}",
            extract_error_message(&body)
        )));
    }

    if no_ack {
        // HTTP publish always returns a response, but we mimic no-ack output.
        match format {
            OutputFormat::Human => {
                let topic = body["topic"].as_str().unwrap_or("?");
                println!(
                    "{} {}",
                    "published".green(),
                    format!("topic={topic}").dimmed()
                );
            }
            OutputFormat::Json => println!("{body}"),
            OutputFormat::Raw => {}
        }
    } else {
        let msg_id = body["message_id"].as_u64().unwrap_or(0);
        let topic = body["topic"].as_str().unwrap_or("?");
        print_publish_result(format, topic, msg_id);
    }

    Ok(())
}

async fn publish_lines_http(
    client: &reqwest::Client,
    url: &str,
    api_key: &str,
    headers: &[Header],
    no_ack: bool,
    format: OutputFormat,
) -> Result<(), CliError> {
    use std::io::BufRead;

    let stdin = std::io::stdin();
    let reader = stdin.lock();
    let mut count: u64 = 0;

    for line in reader.lines() {
        let line = line.map_err(|e| CliError::Error(format!("failed to read stdin: {e}")))?;
        if line.is_empty() {
            continue;
        }

        publish_single_http(
            client,
            url,
            api_key,
            headers,
            Bytes::from(line.into_bytes()),
            no_ack,
            format,
        )
        .await?;
        count += 1;
    }

    if no_ack && count > 0 {
        // Summary for lines mode.
        match format {
            OutputFormat::Human => {
                println!("{}", format!("published {count} messages").green());
            }
            OutputFormat::Json => {
                println!("{}", serde_json::json!({"count": count}));
            }
            OutputFormat::Raw => {}
        }
    }

    Ok(())
}

/// Resolve the payload from args: positional arg > --file > stdin.
fn resolve_payload(args: &PubArgs) -> Result<Bytes, CliError> {
    if let Some(ref text) = args.payload {
        return Ok(Bytes::from(text.clone().into_bytes()));
    }

    if let Some(ref path) = args.file {
        let data = std::fs::read(path)
            .map_err(|e| CliError::Error(format!("failed to read file {path}: {e}")))?;
        return Ok(Bytes::from(data));
    }

    // Read from stdin.
    if std::io::stdin().is_terminal() {
        return Err(CliError::Error(
            "no message payload — provide as argument, --file, or pipe via stdin".into(),
        ));
    }

    std::io::read_to_string(std::io::stdin())
        .map(|s| Bytes::from(s.into_bytes()))
        .map_err(|e| CliError::Error(format!("failed to read stdin: {e}")))
}

/// Publish a single message.
async fn publish_single(
    client: &mut FafbClient,
    topic: &TopicName,
    payload: Bytes,
    headers: &[Header],
    ttl_ms: u32,
    no_ack: bool,
    format: OutputFormat,
) -> Result<(), CliError> {
    if no_ack {
        client
            .publish_no_ack(topic.clone(), payload, headers.to_vec(), ttl_ms)
            .await
            .map_err(|e| CliError::Error(format!("{e}")))?;
        match format {
            OutputFormat::Human => {
                println!(
                    "{} {}",
                    "published (no-ack)".green(),
                    format!("topic={}", topic.0).dimmed()
                );
            }
            OutputFormat::Json => {
                println!("{}", serde_json::json!({"topic": topic.0, "ack": false}));
            }
            OutputFormat::Raw => {}
        }
    } else {
        let result = client
            .publish(topic.clone(), payload, headers.to_vec(), ttl_ms)
            .await
            .map_err(|e| CliError::Error(format!("{e}")))?;
        print_publish_result(format, &topic.0, result.message_id.0);
    }
    Ok(())
}

/// Publish one message per line from stdin.
async fn publish_lines(
    client: &mut FafbClient,
    topic: &TopicName,
    headers: &[Header],
    ttl_ms: u32,
    no_ack: bool,
    format: OutputFormat,
) -> Result<(), CliError> {
    use std::io::BufRead;

    let stdin = std::io::stdin();
    let reader = stdin.lock();
    let mut count: u64 = 0;

    for line in reader.lines() {
        let line = line.map_err(|e| CliError::Error(format!("failed to read stdin: {e}")))?;
        if line.is_empty() {
            continue;
        }

        let payload = Bytes::from(line.into_bytes());
        if no_ack {
            client
                .publish_no_ack(topic.clone(), payload, headers.to_vec(), ttl_ms)
                .await
                .map_err(|e| CliError::Error(format!("{e}")))?;
        } else {
            let result = client
                .publish(topic.clone(), payload, headers.to_vec(), ttl_ms)
                .await
                .map_err(|e| CliError::Error(format!("{e}")))?;
            print_publish_result(format, &topic.0, result.message_id.0);
        }
        count += 1;
    }

    if no_ack {
        match format {
            OutputFormat::Human => {
                println!(
                    "{} {}",
                    format!("published {count} messages (no-ack)").green(),
                    format!("topic={}", topic.0).dimmed()
                );
            }
            OutputFormat::Json => {
                println!(
                    "{}",
                    serde_json::json!({"topic": topic.0, "count": count, "ack": false})
                );
            }
            OutputFormat::Raw => {}
        }
    }
    Ok(())
}

fn print_publish_result(format: OutputFormat, topic: &str, message_id: u64) {
    match format {
        OutputFormat::Human => {
            println!(
                "{} {} {}",
                "published".green(),
                format!("message_id={message_id}").dimmed(),
                format!("topic={topic}").dimmed()
            );
        }
        OutputFormat::Json => {
            println!(
                "{}",
                serde_json::json!({"message_id": message_id, "topic": topic})
            );
        }
        OutputFormat::Raw => {}
    }
}
