//! Log records exported through the `tracing` pipeline (#147): they carry
//! the trace_id / span_id of the span they were emitted in — for native
//! `tracing` macros AND for `log::` records bridged via `LogTracer` — and
//! the span layer keeps only WARN/ERROR events as span events.
//!
//! Mirrors the layer stack in `otel::init_tracing`, with in-memory exporters
//! in place of OTLP.

use image_api::otel::span_layer_filter;
use opentelemetry::trace::{TraceContextExt, TracerProvider};
use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
use opentelemetry_sdk::logs::{InMemoryLogExporter, SdkLoggerProvider};
use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider};
use tracing_opentelemetry::OpenTelemetrySpanExt;
use tracing_subscriber::Layer;
use tracing_subscriber::layer::SubscriberExt;

struct Harness {
    logs: InMemoryLogExporter,
    spans: InMemorySpanExporter,
    tracer_provider: SdkTracerProvider,
}

impl Harness {
    fn run<R>(filter: &str, f: impl FnOnce() -> R) -> (Self, R) {
        // The global LogTracer forwards `log` records to the *current*
        // dispatcher, so each scoped subscriber below receives them.
        let _ = tracing_log::LogTracer::init();

        let logs = InMemoryLogExporter::default();
        let logger_provider = SdkLoggerProvider::builder()
            .with_simple_exporter(logs.clone())
            .build();
        let spans = InMemorySpanExporter::default();
        let tracer_provider = SdkTracerProvider::builder()
            .with_simple_exporter(spans.clone())
            .build();
        let tracer = tracer_provider.tracer("test");

        let subscriber = tracing_subscriber::registry()
            .with(tracing_subscriber::EnvFilter::new(filter))
            .with(
                tracing_opentelemetry::OpenTelemetryLayer::new(tracer)
                    .with_filter(span_layer_filter()),
            )
            .with(OpenTelemetryTracingBridge::new(&logger_provider));
        let out = tracing::subscriber::with_default(subscriber, f);
        (
            Harness {
                logs,
                spans,
                tracer_provider,
            },
            out,
        )
    }

    fn log_bodies(&self) -> Vec<String> {
        self.logs
            .get_emitted_logs()
            .expect("logs")
            .iter()
            .map(|l| format!("{:?}", l.record.body()))
            .collect()
    }
}

#[test]
fn log_records_carry_span_trace_context() {
    let (h, (trace_id, span_id, child_span_id)) = Harness::run("info", || {
        tracing::info!("outside");
        let span = tracing::info_span!("parent");
        let _g = span.enter();
        tracing::info!("tracing inside");
        log::info!("log inside");
        let child = tracing::info_span!("child");
        let child_id = {
            let _c = child.enter();
            tracing::warn!("tracing in child");
            child.context().span().span_context().span_id()
        };
        let ctx = span.context();
        let span_ref = ctx.span();
        let sc = span_ref.span_context();
        (sc.trace_id(), sc.span_id(), child_id)
    });

    let logs = h.logs.get_emitted_logs().expect("logs");
    let find = |body: &str| {
        logs.iter()
            .find(|l| format!("{:?}", l.record.body()).contains(body))
            .unwrap_or_else(|| panic!("no record with body {body}"))
    };

    assert!(find("outside").record.trace_context().is_none());

    for body in ["tracing inside", "log inside"] {
        let tc = find(body).record.trace_context().expect(body);
        assert_eq!(tc.trace_id, trace_id, "{body}");
        assert_eq!(tc.span_id, span_id, "{body}");
    }

    let tc = find("tracing in child")
        .record
        .trace_context()
        .expect("child");
    assert_eq!(tc.trace_id, trace_id);
    assert_eq!(tc.span_id, child_span_id);

    // Bridged records keep their module in code.namespace (the bridge's
    // own target is the fixed "log").
    let namespace = find("log inside")
        .record
        .attributes_iter()
        .find(|(k, _)| k.as_str() == "code.namespace")
        .map(|(_, v)| format!("{v:?}"));
    assert!(
        namespace.is_some_and(|ns| ns.contains("otel_log_correlation")),
        "bridged record lost its module"
    );
}

#[test]
fn span_events_keep_only_warn_and_error() {
    let (h, ()) = Harness::run("info", || {
        let span = tracing::info_span!("parent");
        let _g = span.enter();
        tracing::info!("info event");
        tracing::warn!("warn event");
        tracing::error!("error event");
    });
    h.tracer_provider.force_flush().expect("flush");

    let spans = h.spans.get_finished_spans().expect("spans");
    let parent = spans.iter().find(|s| s.name == "parent").expect("parent");
    let names: Vec<String> = parent.events.iter().map(|e| e.name.to_string()).collect();
    assert_eq!(names, vec!["warn event", "error event"]);

    // The info event is still exported as a (correlated) log record.
    assert!(h.log_bodies().iter().any(|b| b.contains("info event")));
}

/// Per-module `RUST_LOG` directives must still reach bridged `log::`
/// records, as they did under env_logger.
#[test]
fn per_module_directives_apply_to_bridged_log_records() {
    let (h, ()) = Harness::run("info,otel_log_correlation=debug", || {
        log::debug!("bridged debug");
        tracing::debug!("native debug");
    });
    let bodies = h.log_bodies();
    assert!(
        bodies.iter().any(|b| b.contains("native debug")),
        "{bodies:?}"
    );
    assert!(
        bodies.iter().any(|b| b.contains("bridged debug")),
        "{bodies:?}"
    );

    // ...and without the directive, the same records are filtered out.
    let (h, ()) = Harness::run("info", || {
        log::debug!("bridged debug");
        tracing::debug!("native debug");
    });
    assert!(h.log_bodies().is_empty(), "{:?}", h.log_bodies());
}