//! QuestDB ILP sink for training metrics.
//!
//! Writes one row per epoch to a `training_epochs` table in QuestDB via ILP.
//! Sits alongside the Prometheus `training_metrics` module — Prometheus gives
//! real-time gauge snapshots, QuestDB gives persistent time-series history.
//!
//! # Usage
//!
//! ```rust,ignore
//! use common::metrics::questdb_sink;
//!
//! questdb_sink::init(None); // reads QUESTDB_ILP_HOST / QUESTDB_ILP_PORT env
//! // … per epoch …
//! questdb_sink::record_training_epoch(&questdb_sink::EpochRecord {
//!     model: "dqn",
//!     fold: "0",
//!     epoch: 1,
//!     loss: 0.042,
//!     ..Default::default()
//! });
//! // On shutdown:
//! questdb_sink::flush();
//! ```

use crate::questdb::{MetricEntry, MetricValue, QuestDBClient, QuestDBConfig};
use std::sync::OnceLock;
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use tokio::runtime::Handle;

static CLIENT: OnceLock<QuestDBClient> = OnceLock::new();

/// Initialize the global QuestDB training sink.
///
/// Reads `QUESTDB_ILP_HOST` (default `questdb.foxhunt.svc.cluster.local`)
/// and `QUESTDB_ILP_PORT` (default `9009`) from the environment.
/// Safe to call multiple times — second+ calls are no-ops.
pub fn init(config: Option<QuestDBConfig>) {
    CLIENT.get_or_init(|| {
        let cfg = config.unwrap_or_else(|| QuestDBConfig {
            ilp_host: std::env::var("QUESTDB_ILP_HOST")
                .unwrap_or_else(|_| "questdb.foxhunt.svc.cluster.local".into()),
            ilp_port: std::env::var("QUESTDB_ILP_PORT")
                .ok()
                .and_then(|s| s.parse().ok())
                .unwrap_or(9009),
            ..QuestDBConfig::default()
        });
        tracing::info!(
            "QuestDB training sink initialized: {}:{}",
            cfg.ilp_host,
            cfg.ilp_port
        );
        QuestDBClient::new(cfg)
    });
}

/// All metrics for a single training epoch.
#[derive(Debug, Clone)]
pub struct EpochRecord<'a> {
    // Identifiers (ILP symbols — indexed)
    pub model: &'a str,
    pub fold: &'a str,
    pub run_id: &'a str,
    pub symbol: &'a str,

    // Core training
    pub epoch: u32,
    pub loss: f64,
    pub val_loss: f64,
    pub batches_per_second: f64,
    pub epoch_duration_secs: f64,

    // RL diagnostics
    pub q_mean: f64,
    pub q_max: f64,
    pub gradient_norm: f64,
    pub learning_rate: f64,
    pub replay_buffer_size: f64,
    pub action_entropy: f64,
    pub action_diversity: f64,

    // Financial
    pub sharpe: f64,
    pub sortino: f64,
    pub win_rate: f64,
    pub max_drawdown: f64,
    pub profit_factor: f64,
    pub total_return: f64,
    pub avg_return: f64,
    pub total_trades: f64,
}

impl<'a> Default for EpochRecord<'a> {
    fn default() -> Self {
        Self {
            model: "unknown",
            fold: "0",
            run_id: "",
            symbol: "",
            epoch: 0,
            loss: 0.0,
            val_loss: 0.0,
            batches_per_second: 0.0,
            epoch_duration_secs: 0.0,
            q_mean: 0.0,
            q_max: 0.0,
            gradient_norm: 0.0,
            learning_rate: 0.0,
            replay_buffer_size: 0.0,
            action_entropy: 0.0,
            action_diversity: 0.0,
            sharpe: 0.0,
            sortino: 0.0,
            win_rate: 0.0,
            max_drawdown: 0.0,
            profit_factor: 0.0,
            total_return: 0.0,
            avg_return: 0.0,
            total_trades: 0.0,
        }
    }
}

/// Write a single epoch record to QuestDB.
///
/// Non-blocking: buffers locally if QuestDB is unavailable.
/// No-op if `init()` was not called.
pub fn record_training_epoch(rec: &EpochRecord<'_>) {
    let Some(client) = CLIENT.get() else {
        return;
    };

    let ts_ns = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos() as i64;

    let mut columns = Vec::with_capacity(24);

    // Symbols (indexed in QuestDB)
    columns.push(("model".into(), MetricValue::Symbol(rec.model.into())));
    columns.push(("fold".into(), MetricValue::Symbol(rec.fold.into())));
    if !rec.run_id.is_empty() {
        columns.push(("run_id".into(), MetricValue::Symbol(rec.run_id.into())));
    }
    if !rec.symbol.is_empty() {
        columns.push(("symbol".into(), MetricValue::Symbol(rec.symbol.into())));
    }

    // Core training (always present)
    columns.push(("epoch".into(), MetricValue::Int(rec.epoch as i64)));
    columns.push(("loss".into(), MetricValue::Float(rec.loss)));
    columns.push(("val_loss".into(), MetricValue::Float(rec.val_loss)));
    columns.push((
        "batches_per_second".into(),
        MetricValue::Float(rec.batches_per_second),
    ));
    columns.push((
        "epoch_duration_secs".into(),
        MetricValue::Float(rec.epoch_duration_secs),
    ));

    // RL diagnostics
    columns.push(("q_mean".into(), MetricValue::Float(rec.q_mean)));
    columns.push(("q_max".into(), MetricValue::Float(rec.q_max)));
    columns.push(("gradient_norm".into(), MetricValue::Float(rec.gradient_norm)));
    columns.push(("learning_rate".into(), MetricValue::Float(rec.learning_rate)));
    columns.push((
        "replay_buffer_size".into(),
        MetricValue::Float(rec.replay_buffer_size),
    ));
    columns.push((
        "action_entropy".into(),
        MetricValue::Float(rec.action_entropy),
    ));
    columns.push((
        "action_diversity".into(),
        MetricValue::Float(rec.action_diversity),
    ));

    // Financial
    columns.push(("sharpe".into(), MetricValue::Float(rec.sharpe)));
    columns.push(("sortino".into(), MetricValue::Float(rec.sortino)));
    columns.push(("win_rate".into(), MetricValue::Float(rec.win_rate)));
    columns.push(("max_drawdown".into(), MetricValue::Float(rec.max_drawdown)));
    columns.push((
        "profit_factor".into(),
        MetricValue::Float(rec.profit_factor),
    ));
    columns.push(("total_return".into(), MetricValue::Float(rec.total_return)));
    columns.push(("avg_return".into(), MetricValue::Float(rec.avg_return)));
    columns.push(("total_trades".into(), MetricValue::Float(rec.total_trades)));

    let entry = MetricEntry {
        table: "training_epochs".into(),
        columns,
        timestamp_ns: ts_ns,
        created_at: Instant::now(),
    };

    // QuestDBClient::write is async — bridge into current tokio runtime
    if let Ok(handle) = Handle::try_current() {
        handle.spawn(async move {
            client.write(entry).await;
        });
    } else {
        // No async runtime — try a blocking approach via a short-lived runtime
        // This shouldn't happen in practice (training binaries use tokio)
        tracing::debug!("No tokio runtime for QuestDB write, entry buffered");
    }
}

/// Flush any buffered entries to QuestDB. Call before process exit.
///
/// No-op if `init()` was not called.
pub fn flush() {
    let Some(client) = CLIENT.get() else {
        return;
    };

    if let Ok(handle) = Handle::try_current() {
        // We need to block on the async health_check which triggers a flush
        handle.block_on(async {
            // Force a health check which will attempt to flush
            client.health_check().await;
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_init_is_idempotent() {
        // Use a bogus host so it doesn't actually connect
        init(Some(QuestDBConfig {
            ilp_host: "localhost".into(),
            ilp_port: 19009,
            ..QuestDBConfig::default()
        }));
        // Second call is a no-op
        init(None);
        assert!(CLIENT.get().is_some());
    }

    #[test]
    fn test_record_without_init_is_noop() {
        // In a fresh test binary where CLIENT might not be set,
        // this should not panic
        let rec = EpochRecord {
            model: "dqn",
            epoch: 1,
            loss: 0.05,
            ..Default::default()
        };
        record_training_epoch(&rec);
    }

    #[test]
    fn test_default_epoch_record() {
        let rec = EpochRecord::default();
        assert_eq!(rec.model, "unknown");
        assert_eq!(rec.epoch, 0);
        assert_eq!(rec.loss, 0.0);
    }
}
