zc2 0.0.25

//! Terminal UI for the broker.
//!
//! Both the local dashboard (`run_tui`) and the remote attach dashboard
//! (`run_remote_tui`) render through a single code path:
//! [`crate::tui::views::monitor::render`], fed by a
//! [`crate::tui::data::DataSource`]. This module now only owns the terminal
//! lifecycle, the event loop, and (for remote) the background fetch/diff thread;
//! all drawing lives in `tui::views::monitor` and all the color choices in
//! `tui::theme`.

use std::collections::HashSet;
use std::io;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use crossterm::{
    event::{self, Event, KeyCode, KeyEventKind, KeyModifiers},
    execute,
    terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
};
use ratatui::{backend::CrosstermBackend, Terminal};

use super::stats::{StatsCollector, StatsResponse, TransactionStatus};
use super::SharedBrokerState;

use crate::tui::data::{DataSource, LocalSource, RemoteSource};
use crate::tui::theme::Theme;
use crate::tui::views::monitor::{self, MonitorUi, Panel};

/// Cycle the highlighted panel (shared by both dashboards).
fn next_panel(p: Panel) -> Panel {
    match p {
        Panel::Transactions => Panel::Workers,
        Panel::Workers => Panel::Metrics,
        Panel::Metrics => Panel::Transactions,
    }
}

/// Initialize and run the local broker dashboard.
pub fn run_tui(
    state: SharedBrokerState,
    stats: Arc<StatsCollector>,
    running: Arc<AtomicBool>,
) -> io::Result<()> {
    enable_raw_mode()?;
    let mut stdout = io::stdout();
    execute!(stdout, EnterAlternateScreen)?;
    let backend = CrosstermBackend::new(stdout);
    let mut terminal = Terminal::new(backend)?;

    let source = LocalSource::new(state, Arc::clone(&stats));
    let theme = Theme::dark();
    let mut ui = MonitorUi::default();

    let tick_rate = Duration::from_millis(100);
    let mut last_tick = Instant::now();

    while running.load(Ordering::Relaxed) {
        let snapshot = source.snapshot();
        terminal.draw(|f| monitor::render(f, &snapshot, &theme, &ui))?;

        let timeout = tick_rate.saturating_sub(last_tick.elapsed());
        if event::poll(timeout)? {
            if let Event::Key(key) = event::read()? {
                if key.kind == KeyEventKind::Press {
                    match key.code {
                        KeyCode::Char('q') | KeyCode::Esc => {
                            running.store(false, Ordering::Relaxed);
                            break;
                        }
                        KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => {
                            running.store(false, Ordering::Relaxed);
                            break;
                        }
                        KeyCode::Char('?') | KeyCode::F(1) => ui.show_help = !ui.show_help,
                        KeyCode::Tab => ui.selected_panel = next_panel(ui.selected_panel),
                        _ => {}
                    }
                }
            }
        }

        if last_tick.elapsed() >= tick_rate {
            stats.tick_rps();
            last_tick = Instant::now();
        }
    }

    disable_raw_mode()?;
    execute!(terminal.backend_mut(), LeaveAlternateScreen)?;
    terminal.show_cursor()?;
    Ok(())
}

/// Cleanup terminal on panic.
pub fn cleanup_terminal() {
    let _ = disable_raw_mode();
    let _ = execute!(io::stdout(), LeaveAlternateScreen);
}

// ============================================================================
// Remote TUI (attach mode)
// ============================================================================

const MAX_STREAM_LINES: usize = 100;

/// Blocking fetch: returns Ok(stats) or Err(message). Runs in background thread.
fn fetch_stats_once(
    broker_url: &str,
    api_key: Option<&str>,
    user_id: Option<&str>,
) -> Result<StatsResponse, String> {
    let base = broker_url.trim_end_matches('/');
    let url = match user_id {
        Some(uid) => format!("{}/stats?user={}", base, uid),
        None => format!("{}/stats", base),
    };
    // Long timeout so we get historical data even when broker is busy with /execute
    let mut req = ureq::get(&url)
        .config()
        .timeout_global(Some(std::time::Duration::from_secs(60)))
        .build();
    if let Some(key) = api_key {
        req = req.header("Authorization", &format!("Bearer {}", key));
    }
    match req.call() {
        Ok(response) => {
            let body = response
                .into_body()
                .read_to_string()
                .map_err(|e| format!("Read error: {}", e))?;
            serde_json::from_str(&body).map_err(|e| format!("Parse error: {}", e))
        }
        Err(e) => {
            let msg = e.to_string();
            let hint = if msg.contains("timed out") || msg.contains("timeout") {
                " — retrying in background (broker may be busy with long tasks; press r to retry now)"
            } else {
                " — press r to retry"
            };
            Err(format!("{}{}", msg, hint))
        }
    }
}

/// Derive the user_id embedded in a `zk_<user>_<rest>` API key.
fn user_id_from_key(api_key: Option<&str>) -> Option<String> {
    api_key.and_then(|key| {
        key.strip_prefix("zk_").and_then(|rest| {
            rest.rfind('_').and_then(|pos| {
                let uid = &rest[..pos];
                if uid.is_empty() {
                    None
                } else {
                    Some(uid.to_string())
                }
            })
        })
    })
}

/// Run the remote TUI (attach mode). Stats are fetched in a background thread
/// so the UI never blocks or hangs on slow/unreachable brokers.
pub fn run_remote_tui(broker_url: String, api_key: Option<String>) -> io::Result<()> {
    enable_raw_mode()?;
    let mut stdout = io::stdout();
    execute!(stdout, EnterAlternateScreen)?;
    let backend = CrosstermBackend::new(stdout);
    let mut terminal = Terminal::new(backend)?;

    let source = RemoteSource::new(broker_url.clone());
    let theme = Theme::dark();
    let mut ui = MonitorUi {
        selected_panel: Panel::Transactions,
        show_help: false,
    };

    // Handles for the background fetch thread.
    let shared = Arc::clone(&source.shared);
    let refresh_requested = Arc::clone(&source.refresh_requested);
    let fetch_broker_url = broker_url.clone();
    let fetch_api_key = api_key.clone();
    let fetch_user_id = user_id_from_key(api_key.as_deref());

    // Background thread: fetch /stats, diff for stream events, update shared state.
    crate::async_exec::spawn_detached(move || {
        let fetch_interval = Duration::from_millis(400);
        let poll_interval = Duration::from_millis(80);
        let mut last_fetch = Instant::now();
        let mut prev_stats: Option<StatsResponse> = None;
        loop {
            let now = Instant::now();
            let should_fetch = refresh_requested.swap(false, Ordering::Relaxed)
                || now.duration_since(last_fetch) >= fetch_interval;
            if should_fetch {
                match fetch_stats_once(
                    &fetch_broker_url,
                    fetch_api_key.as_deref(),
                    fetch_user_id.as_deref(),
                ) {
                    Ok(stats) => {
                        let mut new_events: Vec<String> = Vec::new();
                        if let Some(ref prev) = prev_stats {
                            let prev_offer_ids: HashSet<_> = prev
                                .task_offers
                                .iter()
                                .map(|o| o.task_id.as_str())
                                .collect();
                            for o in &stats.task_offers {
                                if !prev_offer_ids.contains(o.task_id.as_str()) {
                                    new_events.push(format!(
                                        "{}  TASK_OFFER  task={}  user={}  max_hr={:.2}",
                                        o.timestamp.format("%H:%M:%S"),
                                        o.task_id.chars().take(8).collect::<String>(),
                                        o.requester_user_id.chars().take(8).collect::<String>(),
                                        o.max_price_per_hour,
                                    ));
                                }
                            }
                            let prev_tx_nums: HashSet<u64> =
                                prev.transactions.iter().map(|t| t.tx_num).collect();
                            for tx in &stats.transactions {
                                if !prev_tx_nums.contains(&tx.tx_num) {
                                    let status = match tx.status {
                                        TransactionStatus::Ok => "ok",
                                        TransactionStatus::Fail => "fail",
                                        TransactionStatus::Pending => "pending",
                                    };
                                    new_events.push(format!(
                                        "{}  EXECUTE {}  {}  worker={}  cost={:.4}  {}ms",
                                        tx.timestamp.format("%H:%M:%S"),
                                        status,
                                        tx.request_id
                                            .as_deref()
                                            .unwrap_or("-")
                                            .chars()
                                            .take(8)
                                            .collect::<String>(),
                                        tx.worker.as_deref().unwrap_or("-"),
                                        tx.cost,
                                        tx.duration_ms as u64,
                                    ));
                                }
                            }
                        }
                        if let Ok(mut g) = shared.lock() {
                            for line in new_events {
                                g.event_log.push_back(line);
                                while g.event_log.len() > MAX_STREAM_LINES {
                                    g.event_log.pop_front();
                                }
                            }
                            g.stats = Some(stats.clone());
                            g.last_error = None;
                        }
                        prev_stats = Some(stats);
                    }
                    Err(e) => {
                        if let Ok(mut g) = shared.lock() {
                            g.last_error = Some(e);
                        }
                    }
                }
                last_fetch = Instant::now();
            }
            std::thread::sleep(poll_interval);
        }
    });

    let tick_rate = Duration::from_millis(100);
    let mut last_tick = Instant::now();

    loop {
        let snapshot = source.snapshot();
        terminal.draw(|f| monitor::render(f, &snapshot, &theme, &ui))?;

        let timeout = tick_rate.saturating_sub(last_tick.elapsed());
        if event::poll(timeout)? {
            if let Event::Key(key) = event::read()? {
                if key.kind == KeyEventKind::Press {
                    match key.code {
                        KeyCode::Char('q') | KeyCode::Esc => break,
                        KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => {
                            break
                        }
                        KeyCode::Char('?') | KeyCode::F(1) => ui.show_help = !ui.show_help,
                        KeyCode::Tab => ui.selected_panel = next_panel(ui.selected_panel),
                        KeyCode::Char('r') => source.request_refresh(),
                        _ => {}
                    }
                }
            }
        }

        if last_tick.elapsed() >= tick_rate {
            last_tick = Instant::now();
        }
    }

    disable_raw_mode()?;
    execute!(terminal.backend_mut(), LeaveAlternateScreen)?;
    terminal.show_cursor()?;
    Ok(())
}