use crate::observability::config::{LogFilterConfig, ObservabilityConfig, OtlpProtocol};
use crate::observability::config_file::{
    observability_config_from_rust_log, validate_observability_config,
    validate_observability_config_path, ConfigWatchGuard,
};
use crate::observability::filter::SharedOrderedFilter;
use crate::observability::js_filter_export::{
    clear_global_filter_export, install_global_filter_export,
};
use crate::observability::local::{build_console_layer, build_local_layer};
use crate::observability::metrics::build_metrics_provider;
use crate::observability::reload::{FilterReloadHandle, ObservabilityReloadHandle};
use crate::observability::remote::build_remote_log_layer;
use crate::observability::trace::build_trace_layer;
use anyhow::{Context as AnyhowContext, Result};
use opentelemetry::{global, propagation::TextMapCompositePropagator};
use opentelemetry_sdk::{
    logs::SdkLoggerProvider,
    metrics::SdkMeterProvider,
    propagation::{BaggagePropagator, TraceContextPropagator},
    trace::SdkTracerProvider,
};
use std::path::Path;
use tracing_subscriber::{layer::SubscriberExt, Layer, Registry};

/// Guards and providers that must outlive logging.
pub struct ObservabilityGuards {
    config_watch_guard: Option<ConfigWatchGuard>,
    grpc_runtime: Option<tokio::runtime::Runtime>,
    /// Keeps the non-blocking local file writer alive.
    pub file_guard: Option<tracing_appender::non_blocking::WorkerGuard>,
    /// OpenTelemetry logger provider, retained for shutdown.
    pub logger_provider: Option<SdkLoggerProvider>,
    /// OpenTelemetry tracer provider, retained for shutdown.
    pub tracer_provider: Option<SdkTracerProvider>,
    /// OpenTelemetry meter provider, retained for shutdown.
    pub meter_provider: Option<SdkMeterProvider>,
}

impl ObservabilityGuards {
    /// Flushes and shuts down OpenTelemetry providers.
    pub fn shutdown(self) {
        clear_global_filter_export();
        drop(self.config_watch_guard);
        if let Some(provider) = self.logger_provider {
            let _ = provider.shutdown();
        }
        if let Some(provider) = self.tracer_provider {
            let _ = provider.shutdown();
        }
        if let Some(provider) = self.meter_provider {
            let _ = provider.shutdown();
        }
        drop(self.file_guard);
        drop(self.grpc_runtime);
    }
}

/// Initializes local logging, remote OpenTelemetry logs, traces, and metrics.
///
/// The returned `ObservabilityReloadHandle` controls the local log, remote log,
/// and trace filters initialized by this function. This also installs a global
/// `log` compatibility bridge so `log::trace!` through `log::error!` records
/// enter the same observability layers and target-based filters.
///
/// Initialization fails when either a global tracing subscriber or a global
/// `log` logger such as log4rs or env_logger was already installed.
pub fn init_observability(
    config: ObservabilityConfig,
) -> Result<(ObservabilityReloadHandle, ObservabilityGuards)> {
    validate_observability_config(&config)?;
    let grpc_runtime = uses_grpc(&config)
        .then(tokio::runtime::Runtime::new)
        .transpose()
        .context("failed to create observability gRPC Tokio runtime")?;
    let grpc_runtime_guard = grpc_runtime.as_ref().map(|runtime| runtime.enter());
    let reload_enabled = config.log.dynamic.enabled;
    let remote_uses_local_filter =
        config.log.remote.enabled.unwrap_or(false) && config.log.remote.filter.is_none();
    let local_output_enabled = config.log.local.enabled.unwrap_or(true)
        || config
            .log
            .console
            .as_ref()
            .and_then(|console| console.enabled)
            .unwrap_or(false);
    let local_filter = SharedOrderedFilter::new(config.log.filter.clone())
        .context("failed to build observability filter")?;
    let local_reload = FilterReloadHandle::new(reload_enabled, "local", local_filter.clone());

    let format = config.log.format.unwrap_or_default();

    let mut layers: Vec<Box<dyn Layer<Registry> + Send + Sync>> = Vec::new();
    let mut file_guard = None;
    let mut logger_provider = None;
    let mut tracer_provider = None;
    let mut meter_provider = None;
    let mut remote_reload = None;
    let mut trace_reload = None;

    if config
        .log
        .console
        .as_ref()
        .and_then(|console| console.enabled)
        .unwrap_or(false)
    {
        layers.push(build_console_layer(format, local_filter.clone()));
    }

    if config.log.local.enabled.unwrap_or(true) {
        let (local_layer, guard) =
            build_local_layer(&config.log.local, format, local_filter.clone());
        file_guard = Some(guard);
        layers.push(local_layer);
    }

    if config.log.remote.enabled.unwrap_or(false) {
        let remote_filter = match &config.log.remote.filter {
            Some(remote_filter_config) => SharedOrderedFilter::new(remote_filter_config.clone())
                .context("failed to build remote observability filter")?,
            None => local_filter.clone(),
        };
        remote_reload = Some(FilterReloadHandle::new(
            reload_enabled,
            "remote",
            remote_filter.clone(),
        ));
        let (remote_layer, provider) =
            build_remote_log_layer(&config.log.remote, &config.trace, remote_filter)?;
        logger_provider = Some(provider);
        layers.push(remote_layer);
    }

    if config.trace.enabled {
        let trace_filter_config = config
            .trace
            .filter
            .clone()
            .unwrap_or_else(default_trace_filter);
        let trace_filter = SharedOrderedFilter::new(trace_filter_config)
            .context("failed to build trace observability filter")?;
        trace_reload = Some(FilterReloadHandle::new(
            reload_enabled,
            "trace",
            trace_filter.clone(),
        ));
        let (trace_layer, provider) = build_trace_layer(&config.trace, trace_filter)?;
        install_w3c_text_map_propagator();
        global::set_tracer_provider(provider.clone());
        tracer_provider = Some(provider);
        layers.push(trace_layer);
    }

    if let Some(metrics) = &config.metrics {
        if metrics.enabled {
            let provider = build_metrics_provider(metrics)?;
            global::set_meter_provider(provider.clone());
            meter_provider = Some(provider);
        }
    }

    tracing::dispatcher::set_global_default(tracing_subscriber::registry().with(layers).into())
        .context("failed to initialize tracing subscriber")?;
    tracing_log::LogTracer::builder()
        .with_max_level(log::LevelFilter::Trace)
        .init()
        .context(
            "failed to initialize log compatibility bridge; a global log logger may already be installed",
        )?;
    drop(grpc_runtime_guard);

    let reload = ObservabilityReloadHandle::new(
        local_reload,
        remote_reload,
        trace_reload,
        remote_uses_local_filter,
        local_output_enabled,
    );
    install_global_filter_export(reload.clone());

    Ok((
        reload,
        ObservabilityGuards {
            config_watch_guard: None,
            grpc_runtime,
            file_guard,
            logger_provider,
            tracer_provider,
            meter_provider,
        },
    ))
}

fn uses_grpc(config: &ObservabilityConfig) -> bool {
    (config.log.remote.enabled.unwrap_or(false) && config.log.remote.protocol == OtlpProtocol::Grpc)
        || (config.trace.enabled && config.trace.protocol == OtlpProtocol::Grpc)
        || config
            .metrics
            .as_ref()
            .map(|metrics| metrics.enabled && metrics.protocol == OtlpProtocol::Grpc)
            .unwrap_or(false)
}

fn install_w3c_text_map_propagator() {
    let propagator = TextMapCompositePropagator::new(vec![
        Box::new(BaggagePropagator::new()),
        Box::new(TraceContextPropagator::new()),
    ]);
    global::set_text_map_propagator(propagator);
}

/// Initializes observability from an optional configuration file path.
///
/// When `path` exists, the file is validated, used for initialization, and
/// watched for changes. A stable file change is applied to local, remote, and
/// trace filters after ten seconds. When `path` is `None` or does not exist, a
/// minimal console-only configuration is created from `RUST_LOG`.
pub fn init_observability_from_optional_path(
    path: Option<&Path>,
) -> Result<(ObservabilityReloadHandle, ObservabilityGuards)> {
    let Some(path) = path.filter(|path| path.exists()) else {
        let rust_log = std::env::var("RUST_LOG").ok();
        return init_observability(observability_config_from_rust_log(rust_log.as_deref())?);
    };

    let config = validate_observability_config_path(path)?;
    let (reload, mut guards) = init_observability(config)?;
    guards.config_watch_guard = Some(ConfigWatchGuard::start(path.to_path_buf(), reload.clone()));
    Ok((reload, guards))
}

fn default_trace_filter() -> LogFilterConfig {
    LogFilterConfig {
        default_level: "trace".to_string(),
        overrides: Vec::new(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::observability::config_file::observability_config_from_rust_log;

    #[test]
    fn grpc_runtime_is_created_only_for_enabled_grpc_outputs() {
        let mut config = observability_config_from_rust_log(None).unwrap();
        assert!(!uses_grpc(&config));

        config.log.remote.protocol = OtlpProtocol::Grpc;
        assert!(!uses_grpc(&config));

        config.log.remote.enabled = Some(true);
        assert!(uses_grpc(&config));
    }
}
