//! `fafb-q fetch` command — pull messages from a topic.
//!
//! Supports both TCP (default) and HTTP (`--http`) transport modes.

use crate::cli::{
    CliConfig, CliError, FetchArgs, MessageDisplay, OutputFormat, build_client_config,
    build_http_client, extract_error_message, http_base_url, parse_timestamp, parse_topic,
    print_message, resolve_namespace, response_json,
};
use crate::client::{FafbClient, FetchResult};
use crate::codec::frame::FetchedMessage;
use crate::types::TopicOffset;

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

    // 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)?;

    // Validate arguments.
    if args.from.is_some() && args.since.is_some() {
        return Err(CliError::Error(
            "--from and --since are mutually exclusive".into(),
        ));
    }
    if let Some(0) = args.tail {
        return Err(CliError::Error("--tail count must be > 0".into()));
    }
    if args.from.is_none() && args.since.is_none() && args.tail.is_none() && !args.all {
        return Err(CliError::Error(
            "specify --from <offset>, --since <timestamp>, --tail <N>, or --all".into(),
        ));
    }

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

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

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

    // Determine fetch mode.
    let result = if let Some(tail_count) = args.tail {
        // --tail N: fetch last N messages via client-side resolution.
        fetch_tail(&mut client, &topic, tail_count, format).await
    } else if let Some(ref since_str) = args.since {
        let ts = parse_timestamp(since_str).map_err(CliError::Error)?;
        fetch_from_timestamp(&mut client, &topic, ts, args.count, args.all, format).await
    } else {
        // --from or --all (--all defaults to offset 0).
        let start = args.from.unwrap_or(0);
        if start < 0 {
            // Negative offset: resolve client-side via fetch_tail.
            fetch_tail(&mut client, &topic, start.unsigned_abs() as u32, format).await
        } else {
            fetch_from_offset(
                &mut client,
                &topic,
                start as u64,
                args.count,
                args.all,
                format,
            )
            .await
        }
    };

    let _ = client.close().await;
    result
}

// ---------------------------------------------------------------------------
// HTTP fetch mode
// ---------------------------------------------------------------------------

async fn run_http(
    config: CliConfig,
    namespace: &str,
    topic: &crate::types::TopicName,
    args: FetchArgs,
    format: OutputFormat,
) -> Result<(), CliError> {
    let 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())
    })?;

    // Build initial query: offset, timestamp, or tail.
    let (initial_param, count) = if let Some(tail_count) = args.tail {
        // --tail N: negative offset resolved server-side.
        (format!("offset=-{tail_count}"), tail_count)
    } else if let Some(ref since_str) = args.since {
        let ts = parse_timestamp(since_str).map_err(CliError::Error)?;
        (format!("timestamp={ts}"), args.count)
    } else {
        let start = args.from.unwrap_or(0);
        (format!("offset={start}"), args.count)
    };

    let mut param = initial_param;
    loop {
        let url = format!(
            "{base}/v1/{namespace}/{topic}/fetch?{param}&max_count={count}",
            topic = topic.0,
        );

        let resp = client
            .get(&url)
            .bearer_auth(api_key)
            .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!(
                "server error ({status}): {}",
                extract_error_message(&body)
            )));
        }

        if let Some(w) = body["warning"].as_str() {
            eprintln!("warning: {w}");
        }

        print_http_messages(&body, &topic.0, format);

        if !args.all {
            break;
        }

        let next = body["next_offset"].as_u64().unwrap_or(0);
        let head = body["head_offset"].as_u64().unwrap_or(0);
        let msgs = body["messages"].as_array().map_or(0, |a| a.len());
        if next >= head || msgs == 0 {
            break;
        }

        // Continue pagination with offset-based fetch.
        param = format!("offset={next}");
    }

    Ok(())
}

/// Print messages from an HTTP JSON fetch response.
fn print_http_messages(body: &serde_json::Value, topic: &str, format: OutputFormat) {
    use base64::Engine;
    let b64 = base64::engine::general_purpose::STANDARD;

    let messages = match body["messages"].as_array() {
        Some(arr) => arr,
        None => return,
    };

    for msg in messages {
        let offset = msg["offset"].as_u64().unwrap_or(0);
        let ts = msg["timestamp_millis"].as_u64().unwrap_or(0);
        let ttl_ms = msg["ttl_ms"].as_u64().unwrap_or(0) as u32;

        // Decode base64 payload.
        let payload_bytes = msg["payload_base64"]
            .as_str()
            .and_then(|s| b64.decode(s).ok())
            .unwrap_or_default();

        // Parse headers from JSON array.
        let headers = parse_http_headers(msg);

        print_message(
            format,
            &MessageDisplay {
                topic,
                offset,
                timestamp_millis: ts,
                ttl_ms,
                message_id: None,
                headers: &headers,
                payload: &payload_bytes,
            },
        );
    }
}

/// Parse message headers from an HTTP fetch JSON response message object.
fn parse_http_headers(msg: &serde_json::Value) -> Vec<crate::types::Header> {
    use base64::Engine;
    let b64 = base64::engine::general_purpose::STANDARD;

    let arr = match msg["headers"].as_array() {
        Some(a) => a,
        None => return vec![],
    };

    arr.iter()
        .filter_map(|h| {
            let key = h["key"].as_str()?.as_bytes().to_vec();
            let value = if let Some(v) = h["value"].as_str() {
                v.as_bytes().to_vec()
            } else if let Some(b) = h["value_base64"].as_str() {
                b64.decode(b).ok()?
            } else {
                return None;
            };
            Some(crate::types::Header { key, value })
        })
        .collect()
}

/// Fetch the last `count` messages using client-side head resolution.
async fn fetch_tail(
    client: &mut FafbClient,
    topic: &crate::types::TopicName,
    count: u32,
    format: OutputFormat,
) -> Result<(), CliError> {
    let result = client
        .fetch_tail(topic.clone(), count)
        .await
        .map_err(|e| CliError::Error(format!("{e}")))?;

    print_warning(&result);
    print_messages(&result.messages, &topic.0, format);
    Ok(())
}

/// Fetch starting from an offset, with optional pagination.
async fn fetch_from_offset(
    client: &mut FafbClient,
    topic: &crate::types::TopicName,
    start_offset: u64,
    max_count: u32,
    paginate: bool,
    format: OutputFormat,
) -> Result<(), CliError> {
    let mut current_offset = start_offset;

    loop {
        let result = client
            .fetch(topic.clone(), TopicOffset(current_offset), max_count)
            .await
            .map_err(|e| CliError::Error(format!("{e}")))?;

        print_warning(&result);
        print_messages(&result.messages, &topic.0, format);

        if !paginate || result.next_offset >= result.head_offset || result.messages.is_empty() {
            break;
        }

        current_offset = result.next_offset.0;
    }
    Ok(())
}

/// Fetch starting from a timestamp. First request uses FetchFrom, then
/// paginates with Fetch using next_offset.
async fn fetch_from_timestamp(
    client: &mut FafbClient,
    topic: &crate::types::TopicName,
    timestamp_millis: u64,
    max_count: u32,
    paginate: bool,
    format: OutputFormat,
) -> Result<(), CliError> {
    // First request: FetchFrom (timestamp-based).
    let result = client
        .fetch_from(topic.clone(), timestamp_millis, max_count)
        .await
        .map_err(|e| CliError::Error(format!("{e}")))?;

    print_warning(&result);
    print_messages(&result.messages, &topic.0, format);

    if !paginate || result.next_offset >= result.head_offset || result.messages.is_empty() {
        return Ok(());
    }

    // Continue paginating with offset-based fetch.
    fetch_from_offset(client, topic, result.next_offset.0, max_count, true, format).await
}

fn print_warning(result: &FetchResult) {
    if let Some(ref w) = result.warning {
        eprintln!("warning: {w}");
    }
}

fn print_messages(messages: &[FetchedMessage], topic: &str, format: OutputFormat) {
    for msg in messages {
        print_message(
            format,
            &MessageDisplay {
                topic,
                offset: msg.offset.0,
                timestamp_millis: msg.timestamp_millis,
                ttl_ms: msg.ttl_ms,
                message_id: None,
                headers: &msg.headers,
                payload: &msg.payload,
            },
        );
    }
}
