wsb 0.258.6441

use crate::prelude::*;
use chrono::Utc;
use std::collections::VecDeque;
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::layer::Context as LayerContext;
use tracing_subscriber::registry::LookupSpan;
use tracing_subscriber::{
    fmt, fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer,
};

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

/// The process's log stream: the live channel and the recent history.
static LOG_STREAM: OnceLock<(broadcast::Sender<LogLine>, LogRingBuffer)> = OnceLock::new();

/// How many lines a slow reader of the live channel may fall behind.
const LOG_CHANNEL_CAPACITY: usize = 1024;

/// Where tracing's console output goes: the sink, which decides whether that
/// is a terminal or a record in the packaged stream.
#[derive(Clone, Copy)]
struct ConsoleWriter;

impl std::io::Write for ConsoleWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        // Lossy on purpose: a log line with a stray byte is still a log line,
        // and refusing it here would drop the diagnostic being written.
        crate::output::write_err(&String::from_utf8_lossy(buf));
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for ConsoleWriter {
    type Writer = ConsoleWriter;

    fn make_writer(&'a self) -> Self::Writer {
        *self
    }
}

/// The instant a record was written, as RFC 3339 in UTC.
///
/// One clock for every JSONL line, so a consumer can order lines
/// from two runs against each other without knowing where either ran.
pub fn rfc3339_now() -> String {
    Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
}

/// The process's log stream, which [`BroadcastLayer`] fills: a channel for
/// pages reading it live, and the recent lines for one that connects later.
///
/// One per process, made the first time anything asks — the layer or a
/// reader, whichever comes first. The log is the PROCESS's: a daemon serving
/// several worktrees writes one log, and every page reads that one. It was a
/// pair the first worktree opened handed to the layer, once; every worktree
/// after made its own, which the layer never wrote to, and their Output
/// panels waited for lines for ever.
pub fn log_stream() -> (broadcast::Sender<LogLine>, LogRingBuffer) {
    let (tx, ring) = LOG_STREAM
        .get_or_init(|| (broadcast::channel(LOG_CHANNEL_CAPACITY).0, LogRingBuffer::new()));
    (tx.clone(), ring.clone())
}

/// 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 writes every event to the process's [`log_stream`].
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
    "rusqlite",
    // 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>) {
        let (tx, ring) = log_stream();
        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 Ok(mut buf) = ring.inner.try_lock() {
            if buf.len() >= RING_CAPACITY {
                buf.pop_front();
            }
            buf.push_back(line.clone());
        }

        // 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,hyper=warn,mio=off,...".
const NOISE_FILTER: &str = "\
    ,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/disposable/logs/daemon/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));

    // The console writer goes through the sink rather than straight to stderr:
    // under `--jsonl` a warning written directly would be the one
    // line of the run that is not a record, arriving in the middle of a stream
    // a consumer is parsing. In human mode it reaches the same stderr it
    // always did.
    let console_layer = fmt::layer()
        .with_writer(ConsoleWriter)
        .with_ansi(!crate::output::is_jsonl())
        .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 {
            // The `daemon` stream of the workspace's logs root, so this
            // daemon's own output is bounded by the same limit as everything
            // else it writes. It used to sit in `.wsb/logs`, outside every
            // policy that names a directory — which is how a log nobody had
            // ever looked at grew for a year.
            let log_dir = crate::logs::stream_dir(
                &crate::logs::root(root),
                crate::logs::Stream::Daemon,
            );
            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()?;
    workspace_root_from(&current_dir)
}

/// The workspace `start` is in: the superproject, when `start` is inside one
/// that wsb manages, and otherwise the nearest directory wsb is set up in.
///
/// Every repository of a workspace carries a `.wsb` of its own — its version
/// state and templates — so the nearest one is the repository's, not the
/// workspace's. Taking it put the log of every command run inside a member in
/// that member: the `wsb commit` a ship runs in `tagged-urn/` wrote
/// `tagged-urn/.wsb/disposable/logs/…`, where neither the ship's own log nor
/// the workbench's activity list could see it. A repository that is not part
/// of a workspace has no superproject, and its own `.wsb` is the workspace.
///
/// Only a directory wsb is set up in (`.wsb/state.json`) counts. Any
/// ancestor holding a `.wsb` folder did: a command run anywhere under the
/// home directory logged into `~/.wsb/disposable/` — the machine's old state
/// folder — and that stray folder then stopped the machine's state being
/// moved out of it. A command run in no worktree writes no transcript.
pub fn workspace_root_from(start: &Path) -> Option<PathBuf> {
    if let Some(super_root) = crate::workspace::find_super_root(start) {
        if super_root.join(".wsb").exists() {
            return Some(super_root);
        }
    }
    crate::workspace::find_project_root(start)
}

// ---------------------------------------------------------------------------
// 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::*;

    /// TEST12361: a command run inside a member of the workspace logs to the
    /// workspace, not to the member's own `.wsb`; a directory outside any
    /// workspace keeps its own.
    ///
    /// The `wsb commit` a ship ran in `tagged-urn/` wrote its log into
    /// `tagged-urn/.wsb/disposable/logs`, where nothing that lists logs looks.
    ///
    /// A fixture of its own: a workspace repository holding a member
    /// repository, each with an `.wsb`. It read the real checkout's, which a
    /// guest is handed without — `.wsb` is never copied — so in every guest the
    /// member had no `.wsb` to be mistaken for the workspace's.
    #[test]
    fn test12361_a_member_logs_to_its_workspace() {
        let tree = tempfile::tempdir().unwrap();
        let workspace = crate::platform::canonicalize(&tree.path()).unwrap();
        let member = workspace.join("tagged-urn");
        let plain = workspace.join("wsb");
        for repository in [&workspace, &member] {
            std::fs::create_dir_all(repository.join(".wsb")).unwrap();
            let status = crate::git_backend::git_command()
                .args(["init", "--quiet"])
                .current_dir(repository)
                .status()
                .unwrap();
            assert!(status.success(), "git init {}", repository.display());
        }
        std::fs::create_dir_all(member.join("tagged-urn-rs")).unwrap();
        std::fs::create_dir_all(&plain).unwrap();
        assert!(member.join(".wsb").is_dir(), "the member has its own .wsb");

        assert_eq!(workspace_root_from(&member.join("tagged-urn-rs")), Some(workspace.clone()));
        assert_eq!(workspace_root_from(&member), Some(workspace.clone()));
        assert_eq!(workspace_root_from(&plain), Some(workspace));

        let alone = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(alone.path().join(".wsb")).unwrap();
        std::fs::create_dir_all(alone.path().join("inner")).unwrap();
        // A `.wsb` folder alone is not a workspace: the machine's old state
        // folder in the home directory is one, and a command run under it
        // must not log into it.
        assert_eq!(workspace_root_from(&alone.path().join("inner")), None);
        std::fs::write(alone.path().join(".wsb/state.json"), "{}").unwrap();
        assert_eq!(workspace_root_from(&alone.path().join("inner")), Some(alone.path().to_path_buf()));
    }

    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(),
        }
    }

    /// TEST9045: what the layer is given reaches the stream a page reads, live
    /// and as history.
    ///
    /// The layer and the stream were two halves joined by a setter, and the
    /// seam failed twice: once the setter was never called, and the Output
    /// panel waited for lines for ever; then it was called once per worktree
    /// into a set-once slot, and every worktree after the first read a stream
    /// nothing wrote to. Now there is one stream, so the test is the whole
    /// path: an event in, the same line out of a reader taken beforehand, and
    /// in the history a later reader gets.
    #[tokio::test]
    async fn test9045_an_event_reaches_the_log_stream() {
        let (tx, ring) = log_stream();
        let mut live = tx.subscribe();
        let marker = format!("log-stream-marker-{}", std::process::id());
        let subscriber = tracing_subscriber::registry().with(BroadcastLayer);
        tracing::subscriber::with_default(subscriber, || tracing::info!("{marker}"));

        // Bounded: a layer that writes nothing would otherwise leave this
        // waiting for ever rather than failing.
        let line = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                let line = live.recv().await.expect("the live channel delivers the event");
                if line.message == marker {
                    break line;
                }
            }
        })
        .await
        .expect("the event never reached the live channel");
        assert_eq!(line.level, "INFO");
        let (_, later) = log_stream();
        assert!(
            later.snapshot().await.iter().any(|line| line.message == marker),
            "a reader that connects after the event finds it in the history"
        );
        assert!(ring.snapshot().await.iter().any(|line| line.message == marker));
    }

    /// 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");
    }
}