mecha10-cli 0.1.47

//! Topics command handlers
//!
//! Orchestrates topic discovery and monitoring using CliContext with TUI.

use crate::ui::{MessageBuffer, TopicMessage, TopicsListTui};
use anyhow::Result;
use crossterm::{
    event::{self, Event},
    execute,
    terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
};
use ratatui::{backend::CrosstermBackend, Terminal};
use serde_json::Value;
use std::io::stdout;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;

/// Arguments for topics list command
#[derive(Debug, Clone)]
pub struct TopicsListArgs {
    pub pattern: String,
    #[allow(dead_code)]
    pub verbose: bool,
    #[allow(dead_code)]
    pub grouped: bool,
}

/// Handle topics list command with TUI
///
/// Lists all topics matching the given pattern with interactive TUI.
///
/// # Arguments
///
/// * `ctx` - CLI execution context
/// * `args` - Topics list command arguments
pub async fn handle_topics_list(ctx: &mut crate::context::CliContext, args: &TopicsListArgs) -> Result<()> {
    // Get Redis connection from context
    let redis_url = ctx.redis_url().to_string();
    let redis = ctx.redis()?;

    // Connect to Redis and scan for topics
    let mut conn = redis.get_connection().await.map_err(|e| {
        eprintln!("❌ Failed to connect to Redis at {}", redis_url);
        eprintln!("   Error: {}", e);
        eprintln!();
        eprintln!("Hint: Start Redis with:");
        eprintln!("  docker compose up -d redis");
        eprintln!("  # or");
        eprintln!("  redis-server");
        e
    })?;

    // Scan Redis for topics
    let topics: Vec<String> = redis::cmd("KEYS")
        .arg(&args.pattern)
        .query_async(&mut conn)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to scan Redis: {}", e))?;

    if topics.is_empty() {
        println!();
        println!("No topics found matching pattern: {}", args.pattern);
        println!();
        println!("Hint: Start some nodes to publish data:");
        println!("  mecha10 dev");
        println!();
        return Ok(());
    }

    // Setup terminal for TUI
    enable_raw_mode()?;
    let mut stdout_handle = stdout();
    execute!(stdout_handle, EnterAlternateScreen)?;
    let backend = CrosstermBackend::new(stdout_handle);
    let mut terminal = Terminal::new(backend)?;

    // Create topics list TUI
    let mut tui = TopicsListTui::new(topics);

    // Setup signal handler
    let running = Arc::new(AtomicBool::new(true));
    let r = running.clone();

    ctrlc::set_handler(move || {
        r.store(false, Ordering::SeqCst);
    })?;

    // Channel for monitoring task communication
    let (monitor_tx, mut monitor_rx) = mpsc::channel::<MonitorCommand>(10);
    let mut monitor_handle: Option<tokio::task::JoinHandle<()>> = None;

    // Channel for topic discovery
    let (topic_tx, mut topic_rx) = mpsc::channel::<Vec<String>>(10);

    // Start background topic discovery task
    let pattern_clone = args.pattern.clone();
    let redis_url_clone = redis_url.clone();
    let discovery_handle =
        tokio::spawn(async move { discover_topics_background(&redis_url_clone, &pattern_clone, topic_tx).await });

    // Main event loop
    let result = loop {
        // Check interrupt flag
        if !running.load(Ordering::SeqCst) || tui.should_quit() {
            break Ok(());
        }

        // Draw TUI
        if let Err(e) = terminal.draw(|f| tui.draw(f)) {
            break Err(e.into());
        }

        // Poll for events
        if event::poll(Duration::from_millis(100))? {
            if let Event::Key(key_event) = event::read()? {
                let prev_monitoring = tui.monitoring_topic().map(|s| s.to_string());

                // Handle key
                tui.handle_key(key_event);

                let current_monitoring = tui.monitoring_topic().map(|s| s.to_string());

                // Check if monitoring state changed
                if prev_monitoring != current_monitoring {
                    // Stop previous monitor if any
                    if monitor_handle.is_some() {
                        let _ = monitor_tx.send(MonitorCommand::Stop).await;
                        if let Some(handle) = monitor_handle.take() {
                            handle.abort();
                        }
                    }

                    // Start new monitor if topic selected
                    if let Some(topic) = current_monitoring {
                        let buffer = tui.message_buffer();
                        let redis_url_clone = redis_url.clone();
                        let (stop_tx, stop_rx) = mpsc::channel::<()>(1);

                        monitor_handle = Some(tokio::spawn(async move {
                            let _ = monitor_topic_background(&redis_url_clone, &topic, buffer, stop_rx).await;
                        }));

                        // Drop stop_tx to allow the task to exit when aborted
                        drop(stop_tx);
                    }
                }
            }
        }

        // Check for monitor completion messages
        if let Ok(cmd) = monitor_rx.try_recv() {
            match cmd {
                MonitorCommand::Stop => {
                    // Monitor stopped
                }
            }
        }

        // Check for newly discovered topics
        if let Ok(new_topics) = topic_rx.try_recv() {
            tui.update_topics(new_topics);
        }
    };

    // Cleanup monitor task
    if monitor_handle.is_some() {
        let _ = monitor_tx.send(MonitorCommand::Stop).await;
        if let Some(handle) = monitor_handle.take() {
            handle.abort();
        }
    }

    // Cleanup discovery task
    discovery_handle.abort();

    // Cleanup terminal
    disable_raw_mode()?;
    execute!(terminal.backend_mut(), LeaveAlternateScreen)?;
    terminal.show_cursor()?;

    result
}

/// Command for monitor task
enum MonitorCommand {
    Stop,
}

/// Periodically discover topics in the background
async fn discover_topics_background(redis_url: &str, pattern: &str, tx: mpsc::Sender<Vec<String>>) -> Result<()> {
    // Create Redis connection
    let client = redis::Client::open(redis_url)?;

    loop {
        // Connect and scan for topics
        if let Ok(mut conn) = client.get_multiplexed_async_connection().await {
            if let Ok(topics) = redis::cmd("KEYS")
                .arg(pattern)
                .query_async::<Vec<String>>(&mut conn)
                .await
            {
                // Send discovered topics to main thread
                if tx.send(topics).await.is_err() {
                    // Channel closed, exit
                    break;
                }
            }
        }

        // Wait before next scan (2 seconds)
        tokio::time::sleep(Duration::from_secs(2)).await;
    }

    Ok(())
}

/// Monitor a topic in the background and push messages to buffer
async fn monitor_topic_background(
    redis_url: &str,
    topic: &str,
    buffer: MessageBuffer,
    mut stop_rx: mpsc::Receiver<()>,
) -> Result<()> {
    // Create Redis connection
    let client = redis::Client::open(redis_url)?;
    let mut conn = client.get_multiplexed_async_connection().await?;

    // First, fetch recent history (last 20 messages) using XREVRANGE
    let history_result: redis::RedisResult<redis::Value> = redis::cmd("XREVRANGE")
        .arg(topic)
        .arg("+") // End (most recent)
        .arg("-") // Start (oldest)
        .arg("COUNT")
        .arg(20)
        .query_async(&mut conn)
        .await;

    let mut last_id = "$".to_string(); // Default: start from NOW

    if let Ok(redis::Value::Array(entries)) = history_result {
        // Parse historical messages (they come in reverse order, newest first)
        let mut history_msgs: Vec<(String, TopicMessage)> = Vec::new();
        for entry_data in entries {
            if let Ok(Some(msg)) = parse_stream_entry(&entry_data) {
                history_msgs.push((msg.entry_id.clone(), msg));
            }
        }

        // Reverse to get chronological order (oldest first)
        history_msgs.reverse();

        // Push to buffer and track last ID
        for (entry_id, msg) in history_msgs {
            last_id = entry_id;
            buffer.push(msg);
        }

        tracing::debug!("Loaded recent history for topic '{}', last_id='{}'", topic, last_id);
    }

    tracing::debug!("Starting monitor for topic '{}' with last_id='{}'", topic, last_id);

    loop {
        tokio::select! {
            // Check for stop signal
            _ = stop_rx.recv() => {
                tracing::debug!("Stop signal received for topic '{}'", topic);
                break;
            }

            // Read next messages from Redis stream
            result = read_stream_messages(&mut conn, topic, &last_id) => {
                match result {
                    Ok(Some((messages, new_last_id))) => {
                        tracing::debug!(
                            "Read {} messages from '{}', last_id: '{}' -> '{}'",
                            messages.len(),
                            topic,
                            last_id,
                            new_last_id
                        );
                        for msg in messages {
                            buffer.push(msg);
                        }
                        last_id = new_last_id;
                    }
                    Ok(None) => {
                        // Timeout - no messages, continue
                        tracing::trace!("No messages from '{}' (timeout)", topic);
                    }
                    Err(e) => {
                        tracing::error!("Error reading from '{}': {}", topic, e);
                        // Don't stop on error, keep trying
                        tokio::time::sleep(Duration::from_millis(500)).await;
                    }
                }
            }
        }
    }

    Ok(())
}

/// Read next batch of messages from Redis stream
async fn read_stream_messages(
    conn: &mut redis::aio::MultiplexedConnection,
    topic: &str,
    last_id: &str,
) -> Result<Option<(Vec<TopicMessage>, String)>> {
    let result: redis::RedisResult<redis::Value> = redis::cmd("XREAD")
        .arg("BLOCK")
        .arg(1000) // Block for 1000ms
        .arg("STREAMS")
        .arg(topic)
        .arg(last_id)
        .query_async(conn)
        .await;

    match result {
        Ok(redis::Value::Array(streams)) => {
            let mut messages = Vec::new();
            let mut new_last_id = last_id.to_string();

            // Process each stream in the response
            for stream_data in streams {
                if let redis::Value::Array(stream_parts) = stream_data {
                    if stream_parts.len() >= 2 {
                        // stream_parts[1] is the array of entries
                        if let redis::Value::Array(entries) = &stream_parts[1] {
                            for entry_data in entries {
                                match parse_stream_entry(entry_data) {
                                    Ok(Some(msg)) => {
                                        new_last_id = msg.entry_id.clone();
                                        messages.push(msg);
                                    }
                                    Ok(None) => {
                                        tracing::warn!("Failed to parse stream entry (no data field?)");
                                    }
                                    Err(e) => {
                                        tracing::error!("Error parsing stream entry: {}", e);
                                    }
                                }
                            }
                        }
                    }
                }
            }

            if messages.is_empty() {
                Ok(None)
            } else {
                Ok(Some((messages, new_last_id)))
            }
        }
        Ok(redis::Value::Nil) => Ok(None),
        Ok(other) => {
            tracing::warn!("Unexpected Redis value type: {:?}", other);
            Ok(None)
        }
        Err(e) => {
            tracing::error!("Redis XREAD error: {}", e);
            Err(anyhow::anyhow!("Redis XREAD failed: {}", e))
        }
    }
}

/// Parse a single Redis stream entry into a TopicMessage
fn parse_stream_entry(entry_data: &redis::Value) -> Result<Option<TopicMessage>> {
    if let redis::Value::Array(entry_parts) = entry_data {
        if entry_parts.len() >= 2 {
            // entry_parts[0] is the entry ID
            let entry_id = match &entry_parts[0] {
                redis::Value::BulkString(bytes) => String::from_utf8_lossy(bytes).to_string(),
                _ => return Ok(None),
            };

            if let redis::Value::Array(fields) = &entry_parts[1] {
                // Find the "data" field
                let data_value = extract_data_field(fields)?;

                if let Some(data) = data_value {
                    // Parse the message payload
                    let payload = parse_message_payload(&data)?;

                    // Format timestamp for display
                    let now = chrono::Local::now();
                    let timestamp = now.format("%H:%M:%S").to_string();

                    return Ok(Some(TopicMessage {
                        entry_id,
                        timestamp,
                        payload,
                    }));
                }
            }
        }
    }
    Ok(None)
}

/// Extract the "data" field from Redis stream fields
fn extract_data_field(fields: &[redis::Value]) -> Result<Option<String>> {
    for i in (0..fields.len()).step_by(2) {
        if i + 1 < fields.len() {
            if let redis::Value::BulkString(field_name) = &fields[i] {
                if field_name.as_slice() == b"data" {
                    if let redis::Value::BulkString(value) = &fields[i + 1] {
                        return Ok(Some(String::from_utf8_lossy(value).to_string()));
                    }
                }
            }
        }
    }
    Ok(None)
}

/// Parse message payload from JSON string
fn parse_message_payload(data: &str) -> Result<Value> {
    // Try to parse as JSON envelope first
    if let Ok(envelope) = serde_json::from_str::<Value>(data) {
        // If it has a "payload" field, extract that
        if let Some(payload) = envelope.get("payload") {
            return Ok(payload.clone());
        }
        // Otherwise return the whole envelope
        return Ok(envelope);
    }

    // If not valid JSON, return as string value
    Ok(Value::String(data.to_string()))
}