//! Langfuse ingestion client. Phase 4 feature-gated on `server`.

use std::sync::Arc;
#[cfg(feature = "server")]
use std::time::Duration;

use async_trait::async_trait;
use chrono::Utc;
use thiserror::Error;
#[cfg(feature = "server")]
use tokio::sync::Mutex;
#[cfg(feature = "server")]
use tracing::warn;

use dirigent_protocol::streaming::{
    BusEvent, SessionStream, StreamKind, StreamOutcome, StreamScope, StreamSummary,
};
#[cfg(feature = "server")]
use dirigent_protocol::streaming::StreamError;

#[cfg(feature = "server")]
use crate::mapping::{bus_event_to_items, IngestItem};

/// Langfuse stream configuration (credentials + host).
#[derive(Debug, Clone)]
pub struct LangfuseConfig {
    pub host: String,
    pub public_key: String,
    pub secret_key: String,
}

#[derive(Debug, Error)]
#[cfg_attr(not(feature = "server"), allow(dead_code))]
pub enum LangfuseError {
    #[error("transport: {0}")]
    Transport(String),
    #[error("unexpected status: {0}")]
    Status(u16),
    #[error("serialisation: {0}")]
    Serialisation(String),
}

/// Thin wrapper around `reqwest::Client` that POSTs batches to
/// `{host}/api/public/ingestion` with HTTP basic auth.
#[cfg(feature = "server")]
pub(crate) struct LangfuseClient {
    http: reqwest::Client,
    host: String,
    auth: (String, String),
}

#[cfg(feature = "server")]
impl LangfuseClient {
    pub fn new(config: LangfuseConfig) -> Result<Self, LangfuseError> {
        let http = reqwest::Client::builder()
            .timeout(Duration::from_secs(10))
            .build()
            .map_err(|e| LangfuseError::Transport(e.to_string()))?;
        Ok(Self {
            http,
            host: config.host,
            auth: (config.public_key, config.secret_key),
        })
    }

    pub async fn ingest_batch(&self, batch: Vec<IngestItem>) -> Result<(), LangfuseError> {
        if batch.is_empty() {
            return Ok(());
        }
        let url = format!("{}/api/public/ingestion", self.host.trim_end_matches('/'));
        let payload = serde_json::json!({ "batch": batch });

        let mut attempt = 0u32;
        let mut delay_ms = 100u64;
        loop {
            let resp = self
                .http
                .post(&url)
                .basic_auth(&self.auth.0, Some(&self.auth.1))
                .json(&payload)
                .send()
                .await;

            match resp {
                Ok(r) if r.status().is_success() => return Ok(()),
                Ok(r) if r.status().is_server_error() && attempt < 3 => {
                    warn!(status = %r.status(), attempt, "langfuse ingestion 5xx; retrying");
                }
                Ok(r) => return Err(LangfuseError::Status(r.status().as_u16())),
                Err(e) if attempt < 3 => {
                    warn!(error = %e, attempt, "langfuse transport error; retrying");
                }
                Err(e) => return Err(LangfuseError::Transport(e.to_string())),
            }
            tokio::time::sleep(Duration::from_millis(delay_ms)).await;
            attempt += 1;
            delay_ms = (delay_ms * 2).min(1000);
        }
    }
}

/// A live Langfuse stream. Buffers items in-memory and flushes every N ms
/// or M items, whichever is first.
pub struct LangfuseStream {
    pub config: LangfuseConfig,
    pub scope: StreamScope,
    pub name: String,
    pub active_since: chrono::DateTime<chrono::Utc>,
    #[cfg(feature = "server")]
    client: Arc<LangfuseClient>,
    #[cfg(feature = "server")]
    buffer: Arc<Mutex<Vec<IngestItem>>>,
}

#[cfg(feature = "server")]
const FLUSH_ITEMS: usize = 32;

impl LangfuseStream {
    #[cfg(feature = "server")]
    pub fn new(
        name: String,
        config: LangfuseConfig,
        scope: StreamScope,
    ) -> Result<Arc<Self>, LangfuseError> {
        let client = Arc::new(LangfuseClient::new(config.clone())?);
        Ok(Arc::new(Self {
            config,
            scope,
            name,
            active_since: Utc::now(),
            client,
            buffer: Arc::new(Mutex::new(Vec::new())),
        }))
    }

    #[cfg(not(feature = "server"))]
    pub fn new(name: String, config: LangfuseConfig, scope: StreamScope) -> Arc<Self> {
        Arc::new(Self {
            config,
            scope,
            name,
            active_since: Utc::now(),
        })
    }

    #[cfg(feature = "server")]
    async fn flush(&self) -> Result<(), LangfuseError> {
        let mut buf = self.buffer.lock().await;
        if buf.is_empty() {
            return Ok(());
        }
        let batch: Vec<_> = buf.drain(..).collect();
        drop(buf);
        self.client.ingest_batch(batch).await
    }
}

#[async_trait]
impl SessionStream for LangfuseStream {
    fn summary(&self) -> StreamSummary {
        StreamSummary {
            name: self.name.clone(),
            kind: StreamKind::Langfuse,
            target: format!("langfuse: {}", self.config.host),
            active_since: self.active_since,
        }
    }
    fn scope(&self) -> StreamScope {
        self.scope.clone()
    }

    #[cfg(feature = "server")]
    async fn on_event(&self, event: &BusEvent) -> StreamOutcome {
        let items = bus_event_to_items(event);
        if items.is_empty() {
            return StreamOutcome::Skipped;
        }

        let mut buf = self.buffer.lock().await;
        buf.extend(items);
        if buf.len() >= FLUSH_ITEMS {
            let batch: Vec<_> = buf.drain(..).collect();
            drop(buf);
            match self.client.ingest_batch(batch).await {
                Ok(()) => StreamOutcome::Ok,
                Err(e) => StreamOutcome::Failed(StreamError::Transport(e.to_string())),
            }
        } else {
            StreamOutcome::Ok
        }
    }

    #[cfg(not(feature = "server"))]
    async fn on_event(&self, _event: &BusEvent) -> StreamOutcome {
        StreamOutcome::Ok
    }

    async fn shutdown(&self) {
        #[cfg(feature = "server")]
        {
            let _ = self.flush().await;
        }
    }
}
