wsb 0.124.3689

use anyhow::{Context, Result};
use chrono::Utc;
use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, OnceLock};
use tokio::sync::{broadcast, Mutex};
use tracing::Subscriber;
use tracing_appender::non_blocking::{NonBlocking, WorkerGuard};
use tracing_subscriber::{fmt, fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer};
use tracing_subscriber::layer::Context as LayerContext;
use tracing_subscriber::registry::LookupSpan;

static LOGGING_INITIALIZED: AtomicBool = AtomicBool::new(false);

/// Global broadcast sender — set once by `set_log_broadcast` so the
/// `BroadcastLayer` (installed at logging init time) can forward events
/// even if the channel is created after tracing is initialized.
static LOG_BROADCAST_TX: OnceLock<broadcast::Sender<LogLine>> = OnceLock::new();
/// Global ring buffer — set alongside the broadcast sender.
static LOG_RING_BUFFER: OnceLock<LogRingBuffer> = OnceLock::new();

/// Register a broadcast sender and ring buffer so log events are forwarded to SSE clients.
/// Must be called before the web server starts accepting connections.
pub fn set_log_broadcast(tx: broadcast::Sender<LogLine>, ring: LogRingBuffer) {
    let _ = LOG_BROADCAST_TX.set(tx);
    let _ = LOG_RING_BUFFER.set(ring);
}

/// A single log line broadcast to connected SSE clients.
#[derive(Clone, Debug, serde::Serialize)]
pub struct LogLine {
    pub ts: String,
    pub level: String,
    pub target: String,
    pub message: String,
}

// ---------------------------------------------------------------------------
// LogRingBuffer — recent-history buffer for new SSE clients
// ---------------------------------------------------------------------------

/// Number of recent log lines to replay to a newly-connected SSE client.
const RING_CAPACITY: usize = 200;

/// A thread-safe ring buffer of the most recent log lines.
///
/// New SSE clients receive the current contents as a `history` batch event
/// before the live stream begins, so they immediately see context instead of
/// a blank panel.
#[derive(Clone)]
pub struct LogRingBuffer {
    inner: Arc<Mutex<VecDeque<LogLine>>>,
}

impl LogRingBuffer {
    pub fn new() -> Self {
        Self {
            inner: Arc::new(Mutex::new(VecDeque::with_capacity(RING_CAPACITY))),
        }
    }

    /// Append a line, evicting the oldest entry when at capacity.
    pub async fn push(&self, line: LogLine) {
        let mut buf = self.inner.lock().await;
        if buf.len() >= RING_CAPACITY {
            buf.pop_front();
        }
        buf.push_back(line);
    }

    /// Return a snapshot of all buffered lines in chronological order.
    pub async fn snapshot(&self) -> Vec<LogLine> {
        self.inner.lock().await.iter().cloned().collect()
    }
}

/// A tracing layer that forwards every event to the global broadcast channel.
/// The channel may be registered after this layer is installed via `set_log_broadcast`.
pub struct BroadcastLayer;

// Targets suppressed from the SSE broadcast (output panel).
// These are third-party crates that produce high-frequency internal chatter
// irrelevant to application-level diagnostics.
const SUPPRESSED_TARGETS: &[&str] = &[
    // HTTP / async runtime plumbing
    "hyper", "h2", "tower", "axum", "tokio", "mio", "want", "rustls", "tungstenite",
    // HTTP client
    "reqwest",
    // Database query traces
    "sqlx",
    // File-watching infrastructure
    "notify", "notify_debouncer_mini",
    // Filesystem walk / gitignore parsing
    "ignore", "globset",
];

impl<S: Subscriber + for<'a> LookupSpan<'a>> Layer<S> for BroadcastLayer {
    /// Short-circuit at the callsite level so suppressed crates incur zero cost:
    /// the tracing infrastructure skips field recording entirely.
    fn enabled(&self, metadata: &tracing::Metadata<'_>, _ctx: LayerContext<'_, S>) -> bool {
        let target = metadata.target();
        !SUPPRESSED_TARGETS.iter().any(|t| target.starts_with(t))
    }

    fn on_event(&self, event: &tracing::Event<'_>, _ctx: LayerContext<'_, S>) {
        // No-op until a sender is registered.
        let Some(tx) = LOG_BROADCAST_TX.get() else { return };

        let target = event.metadata().target();

        // Guard again in case enabled() wasn't consulted (e.g. direct on_event call in tests).
        if SUPPRESSED_TARGETS.iter().any(|t| target.starts_with(t)) {
            return;
        }

        let level = event.metadata().level().to_string();

        let mut visitor = MessageVisitor { message: String::new() };
        event.record(&mut visitor);

        let line = LogLine {
            ts: Utc::now().format("%H:%M:%S%.3f").to_string(),
            level,
            target: target.to_string(),
            message: visitor.message,
        };

        // Write to ring buffer using try_lock so we never block the tracing caller.
        // If the lock is contended this event is silently skipped from history
        // (the live broadcast still delivers it to connected clients).
        if let Some(ring) = LOG_RING_BUFFER.get() {
            let inner = &ring.inner;
            if let Ok(mut buf) = inner.try_lock() {
                if buf.len() >= RING_CAPACITY {
                    buf.pop_front();
                }
                buf.push_back(line.clone());
                drop(buf); // explicit drop before `inner` borrow ends
            }
        }

        // Broadcast to live SSE clients: if the channel is full, drop rather than block.
        let _ = tx.send(line);
    }
}

struct MessageVisitor {
    message: String,
}

impl tracing::field::Visit for MessageVisitor {
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
        if field.name() == "message" {
            self.message = value.to_string();
        } else if !self.message.is_empty() {
            self.message.push_str(&format!(" {}={}", field.name(), value));
        } else {
            self.message = format!("{}={}", field.name(), value);
        }
    }

    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
        if field.name() == "message" {
            self.message = format!("{:?}", value);
        } else if !self.message.is_empty() {
            self.message.push_str(&format!(" {}={:?}", field.name(), value));
        } else {
            self.message = format!("{}={:?}", field.name(), value);
        }
    }
}

/// Holds the file-appender worker guard for the lifetime of the process.
/// Dropping it before process exit flushes and closes the log file.
pub struct LoggingGuard {
    _file_guard: Option<WorkerGuard>,
}

/// Third-party crates that produce high-volume noise at their default levels.
/// These are silenced by embedding them directly in the filter parse string
/// so they cannot be overridden by RUST_LOG or post-construction add_directive
/// calls (which can lose to a higher-priority default).
///
/// Format: comma-separated EnvFilter directives, suitable for appending to
/// a base directive string like "debug,sqlx=off,mio=off,...".
const NOISE_FILTER: &str = "\
    ,sqlx=off\
    ,hyper=warn\
    ,h2=warn\
    ,tower=warn\
    ,mio=off\
    ,want=off\
    ,rustls=warn\
    ,reqwest=warn\
    ,notify=warn\
    ,notify_debouncer_mini=warn\
    ,ignore=warn\
    ,globset=warn\
";

/// Open a per-run log file at `<log_dir>/wsb-<timestamp>.log`.
///
/// Returns `(NonBlocking writer, WorkerGuard)` — the guard must be kept alive
/// until process exit to flush the async writer.
fn open_run_log_file(log_dir: &std::path::Path) -> Result<(NonBlocking, WorkerGuard)> {
    std::fs::create_dir_all(log_dir)
        .with_context(|| format!("failed to create log directory {}", log_dir.display()))?;

    let ts = Utc::now().format("%Y-%m-%dT%H-%M-%S");
    let filename = format!("wsb-{ts}.log");
    let path = log_dir.join(&filename);

    let file = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
        .with_context(|| format!("failed to open log file {}", path.display()))?;

    Ok(tracing_appender::non_blocking(file))
}

/// Initialize tracing for the process.
///
/// - Console: stderr, WARN by default; DEBUG when `debug_mode` is true.
///   Respects `RUST_LOG` env var to override.
/// - File: `<workspace>/.wsb/logs/wsb-<timestamp>.log`, DEBUG level.
///   Only created when `debug_mode` is true **and** a `.wsb` workspace is found.
///
/// Returns a guard that must be kept alive until process exit.
pub fn setup_logging(debug_mode: bool) -> Result<LoggingGuard> {
    if LOGGING_INITIALIZED.swap(true, Ordering::SeqCst) {
        return Ok(LoggingGuard { _file_guard: None });
    }

    let workspace_root = detect_workspace_root();

    let console_default = if debug_mode {
        tracing::Level::DEBUG
    } else {
        tracing::Level::WARN
    };

    // Console filter: respect RUST_LOG but always apply noise suppressions.
    // We build the directive string as "<default><noise>" so per-crate overrides
    // are part of the same parse expression and cannot be beaten by the default.
    let console_base = if debug_mode { "debug" } else { "warn" };
    let console_spec = format!("{}{}", console_base, NOISE_FILTER);
    // RUST_LOG still wins when set — use from_env_lossy which falls back to our
    // spec only when RUST_LOG is absent.
    let console_filter = EnvFilter::try_from_env("RUST_LOG")
        .unwrap_or_else(|_| EnvFilter::new(&console_spec));

    let console_layer = fmt::layer()
        .with_writer(std::io::stderr)
        .with_ansi(true)
        .with_target(debug_mode)
        .with_span_events(FmtSpan::NONE)
        .event_format(fmt::format().compact())
        .with_filter(console_filter);

    // File log: only in debug mode, one timestamped file per run.
    let file_guard = if debug_mode {
        if let Some(ref root) = workspace_root {
            let log_dir = root.join(".wsb").join("logs");
            let (non_blocking, guard) = open_run_log_file(&log_dir)?;

            // File always logs at DEBUG for our own code, with noise suppressed.
            let file_spec = format!("debug{}", NOISE_FILTER);
            let file_filter = EnvFilter::new(&file_spec);

            let file_layer = fmt::layer()
                .with_writer(non_blocking)
                .with_ansi(false)
                .with_target(true)
                .with_filter(file_filter);

            tracing_subscriber::registry()
                .with(console_layer)
                .with(file_layer)
                .with(BroadcastLayer)
                .try_init()
                .map_err(|e| anyhow::anyhow!("Failed to initialize tracing: {}", e))?;

            tracing::debug!(log_dir = %log_dir.display(), "debug logging active");
            Some(guard)
        } else {
            tracing_subscriber::registry()
                .with(console_layer)
                .with(BroadcastLayer)
                .try_init()
                .map_err(|e| anyhow::anyhow!("Failed to initialize tracing: {}", e))?;

            tracing::debug!("debug logging active (no .wsb workspace — file log skipped)");
            None
        }
    } else {
        tracing_subscriber::registry()
            .with(console_layer)
            .with(BroadcastLayer)
            .try_init()
            .map_err(|e| anyhow::anyhow!("Failed to initialize tracing: {}", e))?;
        None
    };

    Ok(LoggingGuard { _file_guard: file_guard })
}

/// Walk up from cwd looking for a .wsb directory.
pub fn detect_workspace_root() -> Option<PathBuf> {
    let current_dir = std::env::current_dir().ok()?;
    let mut dir = current_dir.as_path();
    loop {
        if dir.join(".wsb").exists() {
            return Some(dir.to_path_buf());
        }
        dir = dir.parent()?;
    }
}

// ---------------------------------------------------------------------------
// Structured event helpers — thin wrappers for common operation patterns.
// ---------------------------------------------------------------------------

pub fn log_operation_start(operation: &str, details: &str) {
    tracing::info!(operation, details, "starting");
}

pub fn log_operation_complete(operation: &str, duration: std::time::Duration) {
    tracing::info!(operation, elapsed_secs = duration.as_secs_f64(), "completed");
}

pub fn log_operation_error(operation: &str, error: &anyhow::Error) {
    tracing::error!(operation, error = %error, "failed");
    let mut cause = error.source();
    while let Some(err) = cause {
        tracing::debug!(caused_by = %err);
        cause = err.source();
    }
}

pub fn log_warning(context: &str, message: &str) {
    tracing::warn!(context, message);
}

pub fn log_version_info(version: &str, git_hash: Option<&str>) {
    match git_hash {
        Some(hash) => tracing::info!(version, git_commit = hash, "workspace starting"),
        None => tracing::info!(version, "workspace starting"),
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

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

    fn make_line(msg: &str) -> LogLine {
        LogLine {
            ts: "00:00:00.000".to_string(),
            level: "INFO".to_string(),
            target: "test".to_string(),
            message: msg.to_string(),
        }
    }

    /// A freshly-created ring buffer is empty.
    #[tokio::test]
    async fn ring_buffer_starts_empty() {
        let buf = LogRingBuffer::new();
        assert!(buf.snapshot().await.is_empty());
    }

    /// Lines pushed below capacity are all returned in order.
    #[tokio::test]
    async fn ring_buffer_preserves_order_below_capacity() {
        let buf = LogRingBuffer::new();
        for i in 0..10 {
            buf.push(make_line(&format!("msg-{i}"))).await;
        }
        let snap = buf.snapshot().await;
        assert_eq!(snap.len(), 10);
        for (i, line) in snap.iter().enumerate() {
            assert_eq!(line.message, format!("msg-{i}"));
        }
    }

    /// When capacity is exceeded, the oldest lines are evicted and the total
    /// count stays at RING_CAPACITY.
    #[tokio::test]
    async fn ring_buffer_evicts_oldest_at_capacity() {
        let buf = LogRingBuffer::new();
        let overflow = 50;
        for i in 0..(RING_CAPACITY + overflow) {
            buf.push(make_line(&format!("msg-{i}"))).await;
        }
        let snap = buf.snapshot().await;
        assert_eq!(
            snap.len(),
            RING_CAPACITY,
            "buffer must cap at RING_CAPACITY={RING_CAPACITY}"
        );
        // The first line in the snapshot is the (overflow)th message.
        assert_eq!(snap[0].message, format!("msg-{overflow}"));
        // The last line is the final message pushed.
        assert_eq!(
            snap[RING_CAPACITY - 1].message,
            format!("msg-{}", RING_CAPACITY + overflow - 1)
        );
    }

    /// `snapshot()` returns a cloned Vec; subsequent mutations do not affect
    /// the previously returned snapshot.
    #[tokio::test]
    async fn ring_buffer_snapshot_is_independent_clone() {
        let buf = LogRingBuffer::new();
        buf.push(make_line("first")).await;

        let snap1 = buf.snapshot().await;
        buf.push(make_line("second")).await;

        let snap2 = buf.snapshot().await;

        assert_eq!(snap1.len(), 1, "first snapshot must have 1 line");
        assert_eq!(snap2.len(), 2, "second snapshot must have 2 lines");
        assert_eq!(snap1[0].message, "first");
        assert_eq!(snap2[1].message, "second");
    }
}