//! Mechanical OTel / tracing no-plaintext-in-spans enforcement (task C7.2).
//!
//! ADR-024 §2 states the "no plaintext secret in spans" invariant.  This test
//! module provides the mechanical enforcement that was previously documented as
//! convention-only: it installs a recording [`tracing::Subscriber`] layer,
//! exercises the `emit_event` instrumented path, and asserts that no span
//! attribute contains a sensitive value.
//!
//! # How the recorder works
//!
//! [`SpanRecorder`] implements [`tracing_subscriber::Layer`].  On
//! `on_new_span` it captures every field key **and** value (via a custom
//! [`tracing::field::Visit`] implementation).  Spans are stored as
//! [`SpanRecord`] entries in a `Mutex<Vec<SpanRecord>>`.
//!
//! Each test creates a `tracing::Dispatch` scoped to that specific call,
//! so tests do not interfere with one another (no `set_global_default`).
//!
//! # Positive control
//!
//! [`leaky_fn`] is a test-only function that intentionally does NOT skip its
//! `secret` parameter.  The positive-control test confirms the recorder
//! captures field values — without this, the main assertion would be vacuously
//! true if the recorder were broken.

use std::sync::{Arc, Mutex};

use tracing::{span, Subscriber};
use tracing_subscriber::{layer::Context, layer::SubscriberExt, Layer};

// ── Recording infrastructure ──────────────────────────────────────────────────

/// A single captured span: its name and every field as `(key, value)` pairs.
#[derive(Debug, Clone)]
struct SpanRecord {
    name: String,
    fields: Vec<(String, String)>,
}

/// Field visitor that accumulates key/value pairs as strings.
struct FieldVisitor<'a> {
    fields: &'a mut Vec<(String, String)>,
}

impl tracing::field::Visit for FieldVisitor<'_> {
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
        self.fields
            .push((field.name().to_string(), value.to_string()));
    }

    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
        self.fields
            .push((field.name().to_string(), format!("{value:?}")));
    }

    fn record_bool(&mut self, field: &tracing::field::Field, value: bool) {
        self.fields
            .push((field.name().to_string(), value.to_string()));
    }

    fn record_i64(&mut self, field: &tracing::field::Field, value: i64) {
        self.fields
            .push((field.name().to_string(), value.to_string()));
    }

    fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
        self.fields
            .push((field.name().to_string(), value.to_string()));
    }

    fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
        self.fields
            .push((field.name().to_string(), value.to_string()));
    }
}

/// A [`Layer`] that records every new span's name and fields.
struct SpanRecorder {
    spans: Arc<Mutex<Vec<SpanRecord>>>,
}

impl<S: Subscriber> Layer<S> for SpanRecorder {
    fn on_new_span(&self, attrs: &span::Attributes<'_>, _id: &span::Id, _ctx: Context<'_, S>) {
        let mut fields = Vec::new();
        attrs.record(&mut FieldVisitor {
            fields: &mut fields,
        });
        let record = SpanRecord {
            name: attrs.metadata().name().to_string(),
            fields,
        };
        self.spans.lock().unwrap().push(record);
    }
}

/// Install a recording subscriber for the current test scope.
///
/// Returns `(arc_to_recorded_spans, dispatch)`.  Keep `dispatch` alive for the
/// duration of the code under test; drop it when you want to stop recording.
fn make_recording_dispatch() -> (Arc<Mutex<Vec<SpanRecord>>>, tracing::Dispatch) {
    let spans: Arc<Mutex<Vec<SpanRecord>>> = Arc::new(Mutex::new(Vec::new()));
    let recorder = SpanRecorder {
        spans: Arc::clone(&spans),
    };
    let subscriber = tracing_subscriber::registry().with(recorder);
    let dispatch = tracing::Dispatch::new(subscriber);
    (spans, dispatch)
}

// ── Positive-control helper ───────────────────────────────────────────────────
//
// This function deliberately does NOT skip `secret` so the recorder will
// capture the plaintext value.  The positive-control test below confirms the
// recording layer is working — without it the no-plaintext assertions could
// pass vacuously if the recorder never captured anything.

/// Test-only function that intentionally leaks its `secret` parameter into a
/// span field.  Used by the positive-control test to prove the recorder works.
#[tracing::instrument(fields(profile = %profile, secret = %secret))]
fn leaky_fn(profile: &str, secret: &str) {
    // Intentional: no-op body; span instrumentation is the point.
}

// ── Tests ─────────────────────────────────────────────────────────────────────

/// Positive control: the recorder MUST capture field values from an
/// instrumented function that does not skip its parameter.
///
/// If this test fails, the recording infrastructure is broken and the
/// no-plaintext assertion below would be meaningless.
#[test]
fn recorder_captures_field_values_positive_control() {
    let (spans, dispatch) = make_recording_dispatch();

    tracing::dispatcher::with_default(&dispatch, || {
        leaky_fn("test-profile", "super_secret_value_abc");
    });

    drop(dispatch);

    let recorded = spans.lock().unwrap();
    assert!(
        !recorded.is_empty(),
        "recorder must capture at least one span from leaky_fn"
    );

    let all_values: Vec<&str> = recorded
        .iter()
        .flat_map(|s| s.fields.iter().map(|(_, v)| v.as_str()))
        .collect();

    assert!(
        all_values
            .iter()
            .any(|v| v.contains("super_secret_value_abc")),
        "positive control: recorder must capture the un-skipped secret field value; \
         got fields: {all_values:?}"
    );
}

/// Primary invariant: no span emitted by [`tsafe_core::events::emit_event`]
/// may contain a plaintext secret key name or a plaintext secret value.
///
/// [`emit_event`] is instrumented with `#[tracing::instrument(skip(key), …)]`
/// which means:
/// - `profile` and `operation` appear as span fields (acceptable — neither
///   contains a secret value).
/// - `key` (the plaintext secret-key name) is skipped and must not appear.
/// - Secret values are never passed to `emit_event` at all.
///
/// This test is the mechanical enforcement of ADR-024 §2.
#[test]
fn no_secret_value_appears_in_span_attributes() {
    let sensitive_key = "PROD_DB_PASSWORD";
    let (spans, dispatch) = make_recording_dispatch();

    tracing::dispatcher::with_default(&dispatch, || {
        // Exercise multiple operations — set, get, delete — all carrying the
        // sensitive key name.
        temp_env::with_vars(
            [
                ("TSAFE_EVENTS_OUTBOX", None::<&str>),
                ("TSAFE_EVENTS_WEBHOOK_URL", None),
                ("TSAFE_EVENTS_NATS_URL", None),
                ("TSAFE_EVENTS_NATS_SUBJECT", None),
            ],
            || {
                tsafe_core::events::emit_event("prod", "set", Some(sensitive_key));
                tsafe_core::events::emit_event("prod", "get", Some(sensitive_key));
                tsafe_core::events::emit_event("prod", "delete", Some(sensitive_key));
            },
        );
    });

    drop(dispatch);

    let recorded = spans.lock().unwrap();

    // The instrumented function must have produced at least one span.
    assert!(
        !recorded.is_empty(),
        "emit_event must produce at least one span; recorder captured nothing"
    );

    // Assert: no span field VALUE contains the plaintext key name.
    for span in recorded.iter() {
        for (field_name, field_value) in &span.fields {
            assert!(
                !field_value.contains(sensitive_key),
                "Span '{}' field '{}' contains plaintext secret key name '{}' — \
                 #[tracing::instrument(skip(key))] must prevent this",
                span.name,
                field_name,
                sensitive_key
            );
        }
    }

    // Confirm the expected safe fields ARE present (regression guard).
    let has_operation_field = recorded.iter().any(|s| {
        s.fields
            .iter()
            .any(|(k, _)| k == "operation" || k == "profile")
    });
    assert!(
        has_operation_field,
        "emit_event span must expose 'profile' or 'operation' fields for observability; \
         got spans: {recorded:?}"
    );
}

/// Confirm that the `key_ref` hash replacement happens before any adapter
/// serialises the payload — the span layer only sees skipped fields, but the
/// outbox adapter receives the fully-projected event with opaque refs.
///
/// This test combines the OTel assertion with the outbox-adapter assertion,
/// proving both invariants hold in the same execution.
#[test]
fn no_secret_in_spans_and_no_secret_in_outbox_event() {
    use tempfile::tempdir;

    let sensitive_key = "STRIPE_LIVE_SECRET";
    let dir = tempdir().unwrap();
    let outbox = dir.path().join("otel_test.jsonl");

    let (spans, dispatch) = make_recording_dispatch();

    tracing::dispatcher::with_default(&dispatch, || {
        temp_env::with_var("TSAFE_EVENTS_OUTBOX", outbox.to_str(), || {
            tsafe_core::events::emit_event("prod", "set", Some(sensitive_key));
        });
    });

    drop(dispatch);

    // OTel layer: no sensitive key in any span attribute.
    let recorded = spans.lock().unwrap();
    for span in recorded.iter() {
        for (field_name, field_value) in &span.fields {
            assert!(
                !field_value.contains(sensitive_key),
                "Span '{}' field '{}' leaks plaintext key name",
                span.name,
                field_name
            );
        }
    }

    // Outbox adapter: serialised event must not contain the plaintext key name.
    let content = std::fs::read_to_string(&outbox).unwrap();
    assert!(
        !content.contains(sensitive_key),
        "outbox event contains plaintext key name '{sensitive_key}'"
    );

    // The event must still be valid CloudEvents JSON.
    let v: serde_json::Value = serde_json::from_str(content.trim()).unwrap();
    assert_eq!(v["specversion"], "1.0");
    let key_ref_str = v["data"]["key_ref"].as_str().unwrap();
    assert_eq!(key_ref_str.len(), 64, "key_ref must be 64-char SHA-256 hex");
}
