//! Rate counters and colored CLI lines (mirrors Python `[GLOBAL-REG]` / `[DEXTER]`).

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Mutex, Once};
use std::thread;
use std::time::{Duration, Instant};

use colored::Colorize;
use once_cell::sync::{Lazy, OnceCell};
use tracing::Subscriber;
use tracing_subscriber::fmt::format::{Format, FormatEvent, FormatFields, Full, Writer};
use tracing_subscriber::fmt::time::SystemTime;
use tracing_subscriber::fmt::FmtContext;
use tracing_subscriber::registry::LookupSpan;
use prometheus::IntCounter;

/// When set (HTTP server startup), pipeline totals are also exported on `/metrics`.
static PROM_PIPELINE: OnceCell<(IntCounter, IntCounter)> = OnceCell::new();

/// Wire `rate_add_scraped` / `rate_add_inserted` into Prometheus counters (no-op if unset).
pub fn wire_pipeline_prometheus_counters(scraped: IntCounter, inserted: IntCounter) {
    let _ = PROM_PIPELINE.set((scraped, inserted));
}

pub static TELEMETRY: AtomicBool = AtomicBool::new(false);

static RATE_SCRAPED: AtomicU64 = AtomicU64::new(0);
static RATE_INSERTED: AtomicU64 = AtomicU64::new(0);

static CLI_RATE_START: Lazy<Instant> = Lazy::new(Instant::now);

/// EMA + last sample for smoothed CLI rates (updated on every log line).
struct CliRateEma {
    last_i: u64,
    last_s: u64,
    last_t: Instant,
    ema_ins: f64,
    ema_scr: f64,
}

static CLI_RATE_EMA: Lazy<Mutex<CliRateEma>> = Lazy::new(|| {
    Mutex::new(CliRateEma {
        last_i: 0,
        last_s: 0,
        last_t: Instant::now(),
        ema_ins: 0.0,
        ema_scr: 0.0,
    })
});

/// Weight for new samples vs previous EMA (lower = smoother).
const RATE_EMA_ALPHA: f64 = 0.08;
/// Blend EMA with lifetime average so the readout does not whiplash on micro-gaps.
const RATE_LONG_BLEND: f64 = 0.35;

fn cap_rate(v: f64) -> f64 {
    v.clamp(0.0, 99_999.9)
}

fn refresh_cli_rate_ema(g: &mut CliRateEma, cur_i: u64, cur_s: u64, now: Instant) {
    let dt = now.duration_since(g.last_t).as_secs_f64().max(0.002);
    let instant_i = (cur_i.saturating_sub(g.last_i)) as f64 / dt;
    let instant_s = (cur_s.saturating_sub(g.last_s)) as f64 / dt;
    g.last_i = cur_i;
    g.last_s = cur_s;
    g.last_t = now;
    g.ema_ins = RATE_EMA_ALPHA * instant_i + (1.0 - RATE_EMA_ALPHA) * g.ema_ins;
    g.ema_scr = RATE_EMA_ALPHA * instant_s + (1.0 - RATE_EMA_ALPHA) * g.ema_scr;
}

fn format_cli_rate_plain(cur_i: u64, cur_s: u64, g: &CliRateEma) -> String {
    let uptime = CLI_RATE_START.elapsed().as_secs_f64().max(1.0);
    let long_ins = (cur_i as f64 / uptime).min(99_999.9);
    let long_scr = (cur_s as f64 / uptime).min(99_999.9);
    let show_ins = cap_rate(
        (1.0 - RATE_LONG_BLEND) * g.ema_ins + RATE_LONG_BLEND * long_ins,
    );
    let show_scr = cap_rate(
        (1.0 - RATE_LONG_BLEND) * g.ema_scr + RATE_LONG_BLEND * long_scr,
    );
    format!(
        "│ flow {:>7.1} ins/s │ {:>7.1} scr/s │ ",
        show_ins, show_scr
    )
}

/// Background refresh interval so EMA decays during quiet periods without a log line.
const CLI_RATE_TICK_MS: u64 = 200;

static CLI_RATE_BG_TICK: Once = Once::new();

fn cli_rate_tick_from_atomics() {
    let cur_i = RATE_INSERTED.load(Ordering::Relaxed);
    let cur_s = RATE_SCRAPED.load(Ordering::Relaxed);
    let now = Instant::now();
    if let Ok(mut g) = CLI_RATE_EMA.lock() {
        refresh_cli_rate_ema(&mut g, cur_i, cur_s, now);
    }
}

fn spawn_cli_rate_background_ticks() {
    CLI_RATE_BG_TICK.call_once(|| {
        let _ = thread::Builder::new()
            .name("dexter-flow-rate".into())
            .spawn(|| {
                cli_rate_tick_from_atomics();
                let interval = Duration::from_millis(CLI_RATE_TICK_MS);
                loop {
                    thread::sleep(interval);
                    cli_rate_tick_from_atomics();
                }
            });
    });
}

/// Fixed-width pink prefix: smoothed ins/s + scr/s (updated here and on a 200ms background tick).
pub fn cli_rate_prefix() -> String {
    let cur_i = RATE_INSERTED.load(Ordering::Relaxed);
    let cur_s = RATE_SCRAPED.load(Ordering::Relaxed);
    let now = Instant::now();
    let mut g = CLI_RATE_EMA.lock().unwrap_or_else(|e| e.into_inner());
    refresh_cli_rate_ema(&mut g, cur_i, cur_s, now);
    let plain = format_cli_rate_plain(cur_i, cur_s, &g);
    if colored::control::SHOULD_COLORIZE.should_colorize() {
        plain.truecolor(255, 105, 180).to_string()
    } else {
        plain
    }
}

/// `dexter-scrape` / CLI: timestamps + `[ins/s scr/s]` prefix, no Rust module path.
pub fn init_scrape_cli_subscriber() {
    spawn_cli_rate_background_ticks();
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .with_target(false)
        .event_format(CliPrefixedFormat::default())
        .init();
}

#[derive(Clone)]
struct CliPrefixedFormat {
    inner: Format<Full, SystemTime>,
}

impl Default for CliPrefixedFormat {
    fn default() -> Self {
        Self {
            inner: Format::default()
                .with_target(false)
                .with_source_location(false),
        }
    }
}

impl<S, N> FormatEvent<S, N> for CliPrefixedFormat
where
    S: Subscriber + for<'a> LookupSpan<'a>,
    N: for<'writer> FormatFields<'writer> + 'static,
{
    fn format_event(
        &self,
        ctx: &FmtContext<'_, S, N>,
        mut writer: Writer<'_>,
        event: &tracing::Event<'_>,
    ) -> std::fmt::Result {
        writer.write_str(&cli_rate_prefix())?;
        self.inner.format_event(ctx, writer, event)
    }
}

pub fn rate_reset() {
    RATE_SCRAPED.store(0, Ordering::Relaxed);
    RATE_INSERTED.store(0, Ordering::Relaxed);
    if let Ok(mut g) = CLI_RATE_EMA.lock() {
        g.last_i = 0;
        g.last_s = 0;
        g.last_t = Instant::now();
        g.ema_ins = 0.0;
        g.ema_scr = 0.0;
    }
}

pub fn rate_add_scraped(n: u64) {
    RATE_SCRAPED.fetch_add(n, Ordering::Relaxed);
    if let Some((s, _)) = PROM_PIPELINE.get() {
        s.inc_by(n);
    }
}

pub fn rate_add_inserted(n: u64) {
    RATE_INSERTED.fetch_add(n, Ordering::Relaxed);
    if let Some((_, i)) = PROM_PIPELINE.get() {
        i.inc_by(n);
    }
}

pub async fn rate_logger_loop_global() {
    rate_logger_loop_inner("[GLOBAL-REG]").await;
}

pub async fn rate_logger_loop_nl() {
    rate_logger_loop_inner("[DEXTER]").await;
}

async fn rate_logger_loop_inner(tag: &'static str) {
    let mut last_t = Instant::now();
    let mut last_s = RATE_SCRAPED.load(Ordering::Relaxed);
    let mut last_i = RATE_INSERTED.load(Ordering::Relaxed);
    loop {
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
        let now = Instant::now();
        let elapsed = now.duration_since(last_t).as_secs_f64();
        if elapsed <= 0.0 {
            continue;
        }
        let s = RATE_SCRAPED.load(Ordering::Relaxed);
        let i = RATE_INSERTED.load(Ordering::Relaxed);
        let ds = (s - last_s) as f64;
        let di = (i - last_i) as f64;
        let scrapes_per_min = (ds / elapsed) * 60.0;
        let inserts_per_min = (di / elapsed) * 60.0;
        ansi_info(
            tag,
            &format!(
                "Rate (last {:.0}s): scrapes {:.1}/min, inserts {:.1}/min",
                elapsed, scrapes_per_min, inserts_per_min
            ),
        );
        last_t = now;
        last_s = s;
        last_i = i;
    }
}

pub fn ansi_info(tag: &str, msg: &str) {
    println!(
        "{} {} {}",
        tag.magenta(),
        "[INFO]".cyan(),
        msg
    );
}

pub fn ansi_success(tag: &str, msg: &str) {
    println!(
        "{} {} {}",
        tag.magenta(),
        "[SUCCESS]".green(),
        msg
    );
}

pub fn ansi_warning(tag: &str, msg: &str) {
    println!(
        "{} {} {}",
        tag.magenta(),
        "[WARNING]".yellow(),
        msg
    );
}

pub fn ansi_error(tag: &str, msg: &str) {
    eprintln!(
        "{} {} {}",
        tag.magenta(),
        "[ERROR]".red(),
        msg
    );
}
