foxhunt/crates/common/src/questdb.rs

//! QuestDB client with ring buffer for graceful degradation
//!
//! Non-critical path: if QuestDB is unavailable, metrics buffer locally
//! and flush when connection is restored. Trading never depends on QuestDB.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;
use tracing::{debug, info, warn};

/// Maximum entries in the ring buffer before oldest are dropped
const DEFAULT_BUFFER_CAPACITY: usize = 10_000;

/// Health check interval
const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(30);

/// A metric entry waiting to be flushed to QuestDB
#[derive(Debug, Clone)]
pub struct MetricEntry {
    pub table: String,
    pub columns: Vec<(String, MetricValue)>,
    pub timestamp_ns: i64,
    pub created_at: Instant,
}

/// Supported metric value types for ILP
#[derive(Debug, Clone)]
pub enum MetricValue {
    Symbol(String),
    Float(f64),
    Int(i64),
    Bool(bool),
}

/// QuestDB connection configuration
#[derive(Debug, Clone)]
pub struct QuestDBConfig {
    /// ILP host for writes (default: localhost)
    pub ilp_host: String,
    /// ILP port for writes (default: 9009)
    pub ilp_port: u16,
    /// PostgreSQL host for queries (default: localhost)
    pub pg_host: String,
    /// PostgreSQL port for queries (default: 8812)
    pub pg_port: u16,
    /// Ring buffer capacity (default: 10,000)
    pub buffer_capacity: usize,
}

impl Default for QuestDBConfig {
    fn default() -> Self {
        Self {
            ilp_host: "localhost".into(),
            ilp_port: 9009,
            pg_host: "localhost".into(),
            pg_port: 8812,
            buffer_capacity: DEFAULT_BUFFER_CAPACITY,
        }
    }
}

/// QuestDB client with graceful degradation via ring buffer
pub struct QuestDBClient {
    config: QuestDBConfig,
    buffer: Arc<Mutex<VecDeque<MetricEntry>>>,
    connected: Arc<AtomicBool>,
    last_health_check: Arc<Mutex<Instant>>,
}

impl QuestDBClient {
    /// Create a new QuestDB client (does not connect immediately)
    pub fn new(config: QuestDBConfig) -> Self {
        Self {
            buffer: Arc::new(Mutex::new(VecDeque::with_capacity(config.buffer_capacity))),
            connected: Arc::new(AtomicBool::new(false)),
            last_health_check: Arc::new(Mutex::new(Instant::now())),
            config,
        }
    }

    /// Check if QuestDB is currently connected
    pub fn is_connected(&self) -> bool {
        self.connected.load(Ordering::Relaxed)
    }

    /// Get current buffer size (metrics waiting to be flushed)
    pub async fn buffer_size(&self) -> usize {
        self.buffer.lock().await.len()
    }

    /// Get age of oldest buffered entry
    pub async fn buffer_age(&self) -> Duration {
        let buf = self.buffer.lock().await;
        buf.front()
            .map(|e| e.created_at.elapsed())
            .unwrap_or(Duration::ZERO)
    }

    /// Write a metric entry (buffers if QuestDB is unavailable)
    pub async fn write(&self, entry: MetricEntry) {
        let mut buf = self.buffer.lock().await;

        // If buffer is full, drop oldest entry
        if buf.len() >= self.config.buffer_capacity {
            buf.pop_front();
            debug!("QuestDB buffer full, dropped oldest entry");
        }

        buf.push_back(entry);

        // Try to flush if connected
        if self.connected.load(Ordering::Relaxed) {
            drop(buf); // Release lock before flush
            if let Err(e) = self.try_flush().await {
                warn!("QuestDB flush failed: {}", e);
                self.connected.store(false, Ordering::Relaxed);
            }
        }
    }

    /// Attempt to flush buffered entries to QuestDB via ILP
    async fn try_flush(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let mut buf = self.buffer.lock().await;
        if buf.is_empty() {
            return Ok(());
        }

        let mut sender = questdb::ingress::SenderBuilder::new(
            questdb::ingress::Protocol::Tcp,
            &self.config.ilp_host,
            self.config.ilp_port,
        )
        .build()?;

        let mut ilp_buf = questdb::ingress::Buffer::new();
        let entries_to_flush: Vec<MetricEntry> = buf.drain(..).collect();
        let count = entries_to_flush.len();

        for entry in &entries_to_flush {
            ilp_buf.table(entry.table.as_str())?;
            for (name, value) in &entry.columns {
                match value {
                    MetricValue::Symbol(s) => {
                        ilp_buf.symbol(name.as_str(), s.as_str())?;
                    }
                    MetricValue::Float(f) => {
                        ilp_buf.column_f64(name.as_str(), *f)?;
                    }
                    MetricValue::Int(i) => {
                        ilp_buf.column_i64(name.as_str(), *i)?;
                    }
                    MetricValue::Bool(b) => {
                        ilp_buf.column_bool(name.as_str(), *b)?;
                    }
                }
            }
            ilp_buf.at(questdb::ingress::TimestampNanos::new(entry.timestamp_ns))?;
        }

        sender.flush(&mut ilp_buf)?;
        info!("Flushed {} entries to QuestDB", count);

        Ok(())
    }

    /// Periodic health check — call from a background task
    pub async fn health_check(&self) -> bool {
        let mut last = self.last_health_check.lock().await;
        if last.elapsed() < HEALTH_CHECK_INTERVAL {
            return self.connected.load(Ordering::Relaxed);
        }
        *last = Instant::now();
        drop(last);

        // Try connecting via ILP
        let connected = match questdb::ingress::SenderBuilder::new(
            questdb::ingress::Protocol::Tcp,
            &self.config.ilp_host,
            self.config.ilp_port,
        )
        .build()
        {
            Ok(_sender) => {
                if !self.connected.load(Ordering::Relaxed) {
                    info!("QuestDB connection restored");
                }
                true
            }
            Err(e) => {
                if self.connected.load(Ordering::Relaxed) {
                    warn!("QuestDB connection lost: {}", e);
                }
                false
            }
        };

        self.connected.store(connected, Ordering::Relaxed);

        // If reconnected, try flushing buffer
        if connected {
            if let Err(e) = self.try_flush().await {
                warn!("QuestDB reconnect flush failed: {}", e);
            }
        }

        connected
    }

    /// Get ILP connection string for the configured host
    pub fn ilp_address(&self) -> String {
        format!("{}:{}", self.config.ilp_host, self.config.ilp_port)
    }

    /// Get PostgreSQL connection string for queries
    pub fn pg_connection_string(&self) -> String {
        format!(
            "postgresql://admin:quest@{}:{}/qdb",
            self.config.pg_host, self.config.pg_port
        )
    }
}

impl std::fmt::Debug for QuestDBClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("QuestDBClient")
            .field("connected", &self.connected.load(Ordering::Relaxed))
            .field("config", &self.config)
            .finish()
    }
}

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

    #[tokio::test]
    async fn test_client_creation() {
        let client = QuestDBClient::new(QuestDBConfig::default());
        assert!(!client.is_connected());
        assert_eq!(client.buffer_size().await, 0);
    }

    #[tokio::test]
    async fn test_buffer_stores_entries() {
        let client = QuestDBClient::new(QuestDBConfig::default());
        let entry = MetricEntry {
            table: "test".into(),
            columns: vec![("value".into(), MetricValue::Float(1.0))],
            timestamp_ns: 1_234_567_890,
            created_at: Instant::now(),
        };
        client.write(entry).await;
        assert_eq!(client.buffer_size().await, 1);
    }

    #[tokio::test]
    async fn test_buffer_capacity_limit() {
        let config = QuestDBConfig {
            buffer_capacity: 3,
            ..Default::default()
        };
        let client = QuestDBClient::new(config);
        for i in 0..5 {
            let entry = MetricEntry {
                table: "test".into(),
                columns: vec![("i".into(), MetricValue::Int(i))],
                timestamp_ns: i,
                created_at: Instant::now(),
            };
            client.write(entry).await;
        }
        // Buffer capacity is 3, so oldest 2 should be dropped
        assert_eq!(client.buffer_size().await, 3);
    }

    #[tokio::test]
    async fn test_buffer_age_empty() {
        let client = QuestDBClient::new(QuestDBConfig::default());
        assert_eq!(client.buffer_age().await, Duration::ZERO);
    }

    #[tokio::test]
    async fn test_default_config() {
        let config = QuestDBConfig::default();
        assert_eq!(config.ilp_host, "localhost");
        assert_eq!(config.ilp_port, 9009);
        assert_eq!(config.pg_host, "localhost");
        assert_eq!(config.pg_port, 8812);
        assert_eq!(config.buffer_capacity, 10_000);
    }

    #[test]
    fn test_pg_connection_string() {
        let client = QuestDBClient::new(QuestDBConfig::default());
        assert_eq!(
            client.pg_connection_string(),
            "postgresql://admin:quest@localhost:8812/qdb"
        );
    }
}