//! The decline-row tracing capture (`w4-channel-not-held-rows`): a
//! minimal hand-rolled global [`tracing::Subscriber`] recording every
//! event's level + field pairs, the mapping binaries' decline-class
//! rows assert the [`crate`'s event vocabulary](alkssh's observe
//! module) against. The recorded test convention (see
//! `tests/observe.rs`'s module docs) applies verbatim: no
//! `tracing-subscriber` dependency (the hand-rolled shape needs none),
//! and the **global** default is the only capture shape that works
//! (the server side runs on spawned tasks, so a thread-local default
//! never sees their events.
//!
//! Rows install once per binary with [`install`], then match events by
//! field pairs ([`await_event`], so parallel rows never order against
//! each other) and pin the decline event's exact vocabulary with
//! [`assert_request_decline`] (the exact-field-set assertion is the
//! safety posture: no dynamic extra fields exist).

#![allow(dead_code)]

use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};

/// One captured event: the level and the field pairs, in emit order.
#[derive(Debug, Clone)]
pub struct Captured {
    pub level: String,
    pub fields: Vec<(String, String)>,
}

impl Captured {
    /// The field's value by name, the rows' matching input.
    pub fn field(&self, name: &str) -> Option<&str> {
        self.fields
            .iter()
            .find(|(n, _)| n == name)
            .map(|(_, v)| v.as_str())
    }

    fn has(&self, want: &[(&str, &str)]) -> bool {
        want.iter()
            .all(|(n, v)| self.field(n).is_some_and(|got| got == *v))
    }
}

struct FieldVisitor(Vec<(String, String)>);

impl tracing::field::Visit for FieldVisitor {
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
        self.0.push((field.name().to_owned(), value.to_owned()));
    }

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

#[derive(Default)]
struct GlobalCapture(Mutex<Vec<Captured>>);

impl tracing::Subscriber for GlobalCapture {
    fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool {
        true
    }

    fn new_span(&self, _span: &tracing::span::Attributes<'_>) -> tracing::Id {
        tracing::Id::from_u64(1)
    }

    fn record(&self, _span: &tracing::Id, _values: &tracing::span::Record<'_>) {}

    fn record_follows_from(&self, _span: &tracing::Id, _follows: &tracing::Id) {}

    fn event(&self, event: &tracing::Event<'_>) {
        let mut visitor = FieldVisitor(Vec::new());
        event.record(&mut visitor);
        self.0
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(Captured {
                level: event.metadata().level().to_string(),
                fields: visitor.0,
            });
    }

    fn enter(&self, _span: &tracing::Id) {}
    fn exit(&self, _span: &tracing::Id) {}
    fn clone_span(&self, _id: &tracing::Id) -> tracing::Id {
        tracing::Id::from_u64(1)
    }
}

/// The installed capture, one per binary (the [`OnceLock`] makes the
/// exactly-once shape explicit).
fn capture() -> &'static Arc<GlobalCapture> {
    static CAPTURE: OnceLock<Arc<GlobalCapture>> = OnceLock::new();
    CAPTURE.get_or_init(|| {
        let capture = Arc::new(GlobalCapture::default());
        tracing::subscriber::set_global_default(Arc::clone(&capture))
            .expect("the capture suite installs the global capture exactly once");
        capture
    })
}

/// Installs the global capture once per binary. Call it before a row's
/// first wire exchange, so no event is emitted below the
/// not-yet-installed global.
pub fn install() {
    let _ = capture();
}

/// The non-blocking peek; `Some` when a matching event is already on
/// record, `None` otherwise. The e2e rows poll this alongside their
/// bounded curls (the awaiting rows' `await_event` stays fixed-window;
/// a slow CI host turns the wait-then-window shape into a flake, the
/// poll-with-row-deadline shape does not).
pub fn captured_with(want: &[(&str, &str)]) -> Option<Captured> {
    let capture = capture();
    capture
        .0
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .clone()
        .into_iter()
        .find(|e| e.has(want))
}

/// Waits until an event matching every field pair appears, then
/// returns it (the spawned-task surfaces land their events
/// asynchronously (the deadline loop is the observe-suite convention).
pub async fn await_event(want: &[(&str, &str)]) -> Captured {
    let capture = capture();
    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        let events = capture.0.lock().unwrap_or_else(|e| e.into_inner()).clone();
        if let Some(event) = events.into_iter().find(|e| e.has(want)) {
            return event;
        }
        if Instant::now() > deadline {
            panic!("no tracing event matching {want:?} within the window");
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
}

/// The expected `feature_build` value, derived from `cfg!`
/// independently of the crate's match; divergence breaks this loudly.
pub fn expected_build() -> &'static str {
    match (
        cfg!(feature = "session"),
        cfg!(feature = "git"),
        cfg!(feature = "tunnel"),
        cfg!(feature = "sftp"),
    ) {
        (false, false, false, false) => "none",
        (true, false, false, false) => "session",
        (false, true, false, false) => "git",
        (false, false, true, false) => "tunnel",
        (false, false, false, true) => "sftp",
        (true, true, false, false) => "session+git",
        (true, false, true, false) => "session+tunnel",
        (true, false, false, true) => "session+sftp",
        (false, true, true, false) => "git+tunnel",
        (false, true, false, true) => "git+sftp",
        (false, false, true, true) => "tunnel+sftp",
        (true, true, true, false) => "session+git+tunnel",
        (true, true, false, true) => "session+git+sftp",
        (true, false, true, true) => "session+tunnel+sftp",
        (false, true, true, true) => "git+tunnel+sftp",
        (true, true, true, true) => "session+git+tunnel+sftp",
    }
}

/// The request-decline vocabulary assertion: a captured request-level
/// decline carries exactly the vocabulary (a sorted-name-set comparison,
/// nothing more; the dynamic-free shape is the safety posture), the
/// build self-describing, the message shape, and the `debug` level (a
/// decline is the door's normal, honest posture). `exec_route` present
/// iff the row was a routed exec.
pub fn assert_request_decline(event: &Captured, exec_route: bool) {
    assert_eq!(
        event.level, "DEBUG",
        "a decline emits at debug, the vocabulary's level posture"
    );
    assert_eq!(
        event.field("feature_build"),
        Some(expected_build()),
        "every event is self-describing per build"
    );
    assert_eq!(
        event.field("message"),
        Some("decline: channel request rejected"),
        "the request-level decline message shape"
    );
    let mut names: Vec<&str> = event.fields.iter().map(|(n, _)| n.as_str()).collect();
    names.sort_unstable();
    let mut expected: Vec<&str> = vec![
        "channel_id",
        "decline_class",
        "feature_build",
        "message",
        "reason",
        "row",
    ];
    if exec_route {
        expected.push("exec_route");
    }
    expected.sort_unstable();
    assert_eq!(
        names, expected,
        "a request decline carries exactly the vocabulary, nothing else"
    );
}