use std::cell::Cell;
use std::collections::BTreeMap;
use std::fmt;
use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};

use phoxal_api::y2026_1 as api;
use phoxal_bus::{Bus, LogicalTime, OwnerCap, Publisher};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::field::{Field, Visit};
use tracing::{Event, Level, Metadata};
use tracing_subscriber::Layer;
use tracing_subscriber::layer::Context;
use tracing_subscriber::registry::LookupSpan;

const DEFAULT_BUFFER_CAPACITY: usize = 1024;

thread_local! {
    static IN_BUS_LOG_PUBLISH: Cell<bool> = const { Cell::new(false) };
}

#[derive(Debug)]
struct ActivePublisher {
    token: u64,
    sender: mpsc::Sender<LogRecord>,
}

#[derive(Debug)]
pub(crate) struct BusLogState {
    active: Mutex<Option<ActivePublisher>>,
    dropped: AtomicU64,
    max_level: AtomicU8,
    next_token: AtomicU64,
}

impl BusLogState {
    fn new() -> Self {
        Self {
            active: Mutex::new(None),
            dropped: AtomicU64::new(0),
            max_level: AtomicU8::new(level_to_gate(Level::INFO)),
            next_token: AtomicU64::new(1),
        }
    }

    fn set_max_level(&self, level: Level) {
        self.max_level
            .store(level_to_gate(level), Ordering::Relaxed);
    }

    fn try_enqueue(&self, record: LogRecord) {
        let active = self.active.lock().expect("bus log mutex poisoned");
        let Some(active) = active.as_ref() else {
            return;
        };
        match active.sender.try_send(record) {
            Ok(()) => {}
            Err(mpsc::error::TrySendError::Full(_)) => {
                self.dropped.fetch_add(1, Ordering::Relaxed);
            }
            Err(mpsc::error::TrySendError::Closed(_)) => {}
        }
    }

    fn should_publish(&self, metadata: &Metadata<'_>) -> bool {
        self.allows(*metadata.level(), metadata.target())
    }

    fn allows(&self, level: Level, target: &str) -> bool {
        if target_is_filtered(target) {
            return false;
        }
        if IN_BUS_LOG_PUBLISH.with(Cell::get) {
            return false;
        }
        level_to_gate(level) <= self.max_level.load(Ordering::Relaxed)
    }

    fn install_sender(&self, sender: mpsc::Sender<LogRecord>) -> u64 {
        let token = self.next_token.fetch_add(1, Ordering::Relaxed);
        let mut active = self.active.lock().expect("bus log mutex poisoned");
        *active = Some(ActivePublisher { token, sender });
        token
    }

    fn clear_sender(&self, token: u64) {
        let mut active = self.active.lock().expect("bus log mutex poisoned");
        if active.as_ref().is_some_and(|active| active.token == token) {
            *active = None;
        }
    }

    fn take_dropped(&self) -> u32 {
        self.dropped
            .swap(0, Ordering::Relaxed)
            .min(u64::from(u32::MAX)) as u32
    }
}

pub(crate) struct BusLogLayer {
    state: Arc<BusLogState>,
}

impl BusLogLayer {
    pub(crate) fn new(state: Arc<BusLogState>) -> Self {
        Self { state }
    }
}

impl<S> Layer<S> for BusLogLayer
where
    S: tracing::Subscriber + for<'span> LookupSpan<'span>,
{
    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
        if !self.state.should_publish(event.metadata()) {
            return;
        }
        self.state.try_enqueue(LogRecord::from_event(event));
    }
}

#[derive(Clone, Debug)]
struct LogRecord {
    time: api::logs::Timestamp,
    level: api::logs::Level,
    target: String,
    message: String,
    fields: BTreeMap<String, api::logs::LogValue>,
}

impl LogRecord {
    fn from_event(event: &Event<'_>) -> Self {
        let mut visitor = FieldVisitor::default();
        event.record(&mut visitor);
        Self {
            time: timestamp_now(),
            level: level_from_tracing(*event.metadata().level()),
            target: event.metadata().target().to_string(),
            message: visitor.message.unwrap_or_default(),
            fields: visitor.fields,
        }
    }

    fn into_event(self, seq: u64, dropped: u32) -> api::logs::Event {
        api::logs::Event {
            seq,
            time: self.time,
            level: self.level,
            target: self.target,
            message: self.message,
            fields: self.fields,
            dropped,
        }
    }

    fn logical_time(&self) -> LogicalTime {
        let seconds = u64::try_from(self.time.unix_seconds).unwrap_or(0);
        let nanos = seconds
            .saturating_mul(1_000_000_000)
            .saturating_add(u64::from(self.time.nanos));
        LogicalTime::new(0, nanos)
    }
}

#[derive(Default)]
struct FieldVisitor {
    message: Option<String>,
    fields: BTreeMap<String, api::logs::LogValue>,
}

impl FieldVisitor {
    fn record_value(&mut self, field: &Field, value: api::logs::LogValue) {
        let name = field.name();
        if name == "message" {
            self.message = Some(log_value_to_string(&value));
        } else {
            self.fields.insert(name.to_string(), value);
        }
    }
}

impl Visit for FieldVisitor {
    fn record_bool(&mut self, field: &Field, value: bool) {
        self.record_value(field, api::logs::LogValue::Bool(value));
    }

    fn record_i64(&mut self, field: &Field, value: i64) {
        self.record_value(field, api::logs::LogValue::I64(value));
    }

    fn record_u64(&mut self, field: &Field, value: u64) {
        self.record_value(field, api::logs::LogValue::U64(value));
    }

    fn record_f64(&mut self, field: &Field, value: f64) {
        self.record_value(field, api::logs::LogValue::F64(value));
    }

    fn record_str(&mut self, field: &Field, value: &str) {
        self.record_value(field, api::logs::LogValue::String(value.to_string()));
    }

    fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) {
        self.record_value(field, api::logs::LogValue::String(format!("{value:?}")));
    }
}

pub(crate) fn new_state_from_env() -> Arc<BusLogState> {
    let state = Arc::new(BusLogState::new());
    if let Some(level) = std::env::var("PHOXAL_BUS_LOG_LEVEL")
        .ok()
        .and_then(|value| parse_level(&value))
    {
        state.set_max_level(level);
    }
    state
}

pub(crate) fn attach(bus: Bus, participant_id: &str) -> BusLogGuard {
    attach_with_capacity(bus, participant_id, DEFAULT_BUFFER_CAPACITY)
}

fn attach_with_capacity(bus: Bus, participant_id: &str, capacity: usize) -> BusLogGuard {
    let state = crate::participant::runner::bus_log_state();
    let (sender, receiver) = mpsc::channel(capacity.max(1));
    let token = state.install_sender(sender);
    let participant_id = participant_id.to_string();
    let task = tokio::spawn(drain_loop(
        Arc::clone(&state),
        bus,
        participant_id,
        receiver,
    ));
    BusLogGuard {
        token,
        state,
        task: Some(task),
    }
}

pub(crate) struct BusLogGuard {
    token: u64,
    state: Arc<BusLogState>,
    task: Option<JoinHandle<()>>,
}

impl BusLogGuard {
    pub(crate) async fn shutdown(mut self) {
        self.state.clear_sender(self.token);
        if let Some(task) = self.task.take() {
            let _ = task.await;
        }
    }
}

impl Drop for BusLogGuard {
    fn drop(&mut self) {
        self.state.clear_sender(self.token);
        if let Some(task) = self.task.take() {
            task.abort();
        }
    }
}

async fn drain_loop(
    state: Arc<BusLogState>,
    bus: Bus,
    participant_id: String,
    mut receiver: mpsc::Receiver<LogRecord>,
) {
    let topic = api::topic::internal::new(OwnerCap::__mint())
        .logs(&participant_id)
        .topic();
    let publisher = match Publisher::<api::logs::Event>::new(bus, &topic) {
        Ok(publisher) => publisher,
        Err(_) => return,
    };
    let mut seq = 0_u64;
    while let Some(record) = receiver.recv().await {
        let at = record.logical_time();
        let event = record.into_event(seq, state.take_dropped());
        seq = seq.wrapping_add(1);
        let result = IN_BUS_LOG_PUBLISH.with(|guard| {
            guard.set(true);
            let result = publisher.try_publish(at, event);
            guard.set(false);
            result
        });
        if result.is_err() {
            state.dropped.fetch_add(1, Ordering::Relaxed);
        }
    }
}

fn target_is_filtered(target: &str) -> bool {
    target.starts_with("zenoh")
        || target.starts_with("phoxal_bus")
        || target.starts_with("phoxal.bus")
}

fn level_to_gate(level: Level) -> u8 {
    match level {
        Level::ERROR => 1,
        Level::WARN => 2,
        Level::INFO => 3,
        Level::DEBUG => 4,
        Level::TRACE => 5,
    }
}

fn level_from_tracing(level: Level) -> api::logs::Level {
    match level {
        Level::ERROR => api::logs::Level::Error,
        Level::WARN => api::logs::Level::Warn,
        Level::INFO => api::logs::Level::Info,
        Level::DEBUG => api::logs::Level::Debug,
        Level::TRACE => api::logs::Level::Trace,
    }
}

fn parse_level(value: &str) -> Option<Level> {
    match value.trim().to_ascii_lowercase().as_str() {
        "error" => Some(Level::ERROR),
        "warn" | "warning" => Some(Level::WARN),
        "info" => Some(Level::INFO),
        "debug" => Some(Level::DEBUG),
        "trace" => Some(Level::TRACE),
        _ => None,
    }
}

fn timestamp_now() -> api::logs::Timestamp {
    match SystemTime::now().duration_since(UNIX_EPOCH) {
        Ok(duration) => api::logs::Timestamp {
            unix_seconds: i64::try_from(duration.as_secs()).unwrap_or(i64::MAX),
            nanos: duration.subsec_nanos(),
        },
        Err(error) => {
            let duration = error.duration();
            api::logs::Timestamp {
                unix_seconds: -i64::try_from(duration.as_secs()).unwrap_or(i64::MAX),
                nanos: duration.subsec_nanos(),
            }
        }
    }
}

fn log_value_to_string(value: &api::logs::LogValue) -> String {
    match value {
        api::logs::LogValue::Bool(value) => value.to_string(),
        api::logs::LogValue::I64(value) => value.to_string(),
        api::logs::LogValue::U64(value) => value.to_string(),
        api::logs::LogValue::F64(value) => value.to_string(),
        api::logs::LogValue::String(value) => value.clone(),
    }
}

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

    fn record() -> LogRecord {
        LogRecord {
            time: api::logs::Timestamp {
                unix_seconds: 1,
                nanos: 2,
            },
            level: api::logs::Level::Info,
            target: "test".to_string(),
            message: "hello".to_string(),
            fields: BTreeMap::new(),
        }
    }

    #[test]
    fn level_gate_defaults_to_info_plus() {
        let state = BusLogState::new();
        assert!(state.allows(Level::INFO, "app"));
        assert!(state.allows(Level::WARN, "app"));
        assert!(!state.allows(Level::DEBUG, "app"));

        state.set_max_level(Level::DEBUG);
        assert!(state.allows(Level::DEBUG, "app"));
    }

    #[test]
    fn reentrancy_targets_are_filtered() {
        let state = BusLogState::new();
        assert!(!state.allows(Level::INFO, "zenoh"));
        assert!(!state.allows(Level::INFO, "zenoh_transport"));
        assert!(!state.allows(Level::INFO, "phoxal_bus::session"));
        assert!(!state.allows(Level::INFO, "phoxal.bus"));
    }

    #[tokio::test]
    async fn bounded_sender_counts_drops_under_flood() {
        let state = BusLogState::new();
        let (sender, mut receiver) = mpsc::channel(1);
        let token = state.install_sender(sender);

        state.try_enqueue(record());
        state.try_enqueue(record());
        state.try_enqueue(record());

        assert!(receiver.try_recv().is_ok());
        assert_eq!(state.take_dropped(), 2);
        state.clear_sender(token);
    }

    #[test]
    fn no_bus_attached_is_a_noop() {
        let state = BusLogState::new();
        state.try_enqueue(record());
        assert_eq!(state.take_dropped(), 0);
    }
}
