//! `fafb-q sub` command — subscribe to a topic and stream messages.

use crate::cli::{
    CliConfig, CliError, MessageDisplay, OutputFormat, SubArgs, build_client_config,
    parse_timestamp, parse_topic, print_message, resolve_namespace,
};
use crate::client::FafbClient;
use crate::codec::frame::Frame;
use crate::types::{GroupName, TopicName};

/// Run the `sub` command.
pub async fn run(config: CliConfig, args: SubArgs) -> 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.count {
        return Err(CliError::Error("--count must be greater than 0".into()));
    }

    let group = args.group.map(GroupName);

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

    // Send the appropriate subscribe frame.
    let subscribe_frame = if let Some(offset) = args.from {
        Frame::SubscribeAt {
            topic: topic.clone(),
            offset: crate::types::TopicOffset(offset),
            group: group.clone(),
        }
    } else if let Some(ref since_str) = args.since {
        let ts = parse_timestamp(since_str).map_err(CliError::Error)?;
        Frame::SubscribeFrom {
            topic: topic.clone(),
            timestamp_millis: ts,
            group: group.clone(),
        }
    } else {
        Frame::Subscribe {
            topic: topic.clone(),
            group: group.clone(),
        }
    };

    client
        .send_frame(subscribe_frame)
        .await
        .map_err(|e| CliError::Error(format!("failed to subscribe: {e}")))?;

    // Wait for SubscribeAck.
    let ack_frame = client
        .recv_frame()
        .await
        .map_err(|e| CliError::Error(format!("failed to receive SubscribeAck: {e}")))?;

    match &ack_frame {
        Frame::SubscribeAck {
            start_offset,
            matched_count,
            ..
        } => match format {
            OutputFormat::Human => {
                let mut msg = format!("subscribed to {} from offset {}", topic.0, start_offset.0);
                if *matched_count > 0 {
                    msg.push_str(&format!(" ({matched_count} matching topics)"));
                }
                eprintln!("{msg}");
            }
            OutputFormat::Json => {
                let mut obj = serde_json::json!({
                    "type": "subscribed",
                    "topic": topic.0,
                    "start_offset": start_offset.0,
                });
                if *matched_count > 0 {
                    obj["matched_count"] = serde_json::Value::Number((*matched_count).into());
                }
                println!("{obj}");
            }
            OutputFormat::Raw => {}
        },
        Frame::Error { code, message } => {
            return Err(CliError::Error(format!(
                "subscribe error [{code:?}]: {message}"
            )));
        }
        other => {
            return Err(CliError::Error(format!(
                "unexpected frame after subscribe: {other:?}"
            )));
        }
    }

    // Main receive loop with Ctrl+C handling.
    //
    // On first Ctrl+C: send Unsubscribe, drain pending Delivers until
    // UnsubscribeAck, then exit. On second Ctrl+C: force-close immediately.
    let mut remaining = args.count;
    let auto_ack = !args.no_ack;
    let mut shutting_down = false;

    loop {
        tokio::select! {
            frame_result = client.recv_frame() => {
                let frame = match frame_result {
                    Ok(f) => f,
                    Err(e) => {
                        if shutting_down {
                            break;
                        }
                        return Err(CliError::Error(format!("connection error: {e}")));
                    }
                };

                match frame {
                    Frame::Deliver {
                        message_id,
                        offset,
                        timestamp_millis,
                        ttl_ms,
                        topic: deliver_topic,
                        headers,
                        payload,
                    } => {
                        // During shutdown drain, skip printing — just ack.
                        if shutting_down {
                            if auto_ack {
                                let _ = client.send_frame(Frame::Ack { message_id }).await;
                            }
                            continue;
                        }

                        print_message(
                            format,
                            &MessageDisplay {
                                topic: &deliver_topic.0,
                                offset: offset.0,
                                timestamp_millis,
                                ttl_ms,
                                message_id: Some(message_id.0),
                                headers: &headers,
                                payload: &payload,
                            },
                        );

                        if auto_ack {
                            let _ = client.send_frame(Frame::Ack { message_id }).await;
                        }

                        if let Some(ref mut n) = remaining {
                            *n = n.saturating_sub(1);
                            if *n == 0 {
                                let _ = unsubscribe(&mut client, &topic, &group).await;
                                break;
                            }
                        }
                    }
                    Frame::UnsubscribeAck { .. } => {
                        if shutting_down {
                            break;
                        }
                    }
                    Frame::Error { code, message } => {
                        if code.is_fatal() {
                            return Err(CliError::Error(format!(
                                "server error [{code:?}]: {message}"
                            )));
                        }
                        eprintln!("warning: [{code:?}] {message}");
                    }
                    Frame::GoingAway { reason: _, deadline_ms, message } => {
                        eprintln!("server going away ({deadline_ms}ms): {message}");
                        break;
                    }
                    _ => {}
                }
            }

            _ = tokio::signal::ctrl_c() => {
                if shutting_down {
                    // Second Ctrl+C — force close.
                    eprintln!("\nforce closing");
                    break;
                }
                shutting_down = true;
                eprintln!("\nunsubscribing...");

                // Send Unsubscribe — the recv loop will drain Delivers
                // until UnsubscribeAck arrives (or timeout).
                if client.send_frame(Frame::Unsubscribe {
                    topic: topic.clone(),
                    group: group.clone(),
                }).await.is_err() {
                    break;
                }

                // Set a deadline so we don't wait forever for the ack.
                // The select loop will keep running; if the deadline fires
                // before UnsubscribeAck, we force-break.
                tokio::spawn({
                    let notify = tokio::sync::Notify::new();
                    // This is a safety net — the main loop breaks on
                    // UnsubscribeAck. We just need a timeout to force exit.
                    async move {
                        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
                        notify.notify_one();
                    }
                });
            }
        }

        // Safety timeout: if shutting_down and we've been looping too long,
        // break. We rely on the recv_frame arm matching UnsubscribeAck above.
    }

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

/// Send Unsubscribe and drain frames until UnsubscribeAck (or timeout).
///
/// Used when `--count` is reached. The Ctrl+C path sends Unsubscribe
/// inline and lets the main loop drain.
async fn unsubscribe(
    client: &mut FafbClient,
    topic: &TopicName,
    group: &Option<GroupName>,
) -> Result<(), CliError> {
    client
        .send_frame(Frame::Unsubscribe {
            topic: topic.clone(),
            group: group.clone(),
        })
        .await
        .map_err(|e| CliError::Error(format!("failed to send Unsubscribe: {e}")))?;

    // Drain frames until UnsubscribeAck or timeout.
    let deadline = tokio::time::sleep(std::time::Duration::from_secs(5));
    tokio::pin!(deadline);

    loop {
        tokio::select! {
            frame = client.recv_frame() => {
                match frame {
                    Ok(Frame::UnsubscribeAck { .. }) => return Ok(()),
                    Ok(Frame::Error { code, message }) => {
                        eprintln!("unsubscribe error [{code:?}]: {message}");
                        return Ok(());
                    }
                    Ok(_) => continue, // Drain pending Delivers
                    Err(e) => {
                        eprintln!("error during unsubscribe: {e}");
                        return Ok(());
                    }
                }
            }
            _ = &mut deadline => {
                eprintln!("unsubscribe timed out");
                return Ok(());
            }
        }
    }
}
