OpenPXE/crates/core/src/log_bus.rs

//! In-process log bus.
//!
//! The web UI's Terminal tab streams live server logs over SSE. To feed it
//! we install a `tracing_subscriber::Layer` that captures formatted lines
//! and pushes them onto:
//!
//!   1. A bounded `tokio::sync::broadcast` channel for live subscribers.
//!   2. A small in-memory ring buffer (default 500 lines) so a UI that
//!      connects mid-session sees recent context, not a blank pane.
//!
//! No file logging happens here — Docker/OpenShift already capture stdout.
//! This is purely an extra fan-out for the UI.

use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::sync::Arc;
use time::OffsetDateTime;
use tokio::sync::broadcast;
use tracing::field::{Field, Visit};
use tracing::{Event, Level, Subscriber};
use tracing_subscriber::layer::Context;
use tracing_subscriber::registry::LookupSpan;
use tracing_subscriber::Layer;

/// One captured log line. Cheap to clone (small struct, short strings).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LogLine {
    #[serde(with = "time::serde::rfc3339")]
    pub timestamp: OffsetDateTime,
    /// Lowercase: `error`, `warn`, `info`, `debug`, `trace`.
    pub level: String,
    pub target: String,
    pub message: String,
}

impl LogLine {
    /// Compact one-line "tail -f"-style render.
    #[must_use]
    pub fn render(&self) -> String {
        // 2026-04-29T12:34:56Z [info]  openpxe::http: HTTP listening on 0.0.0.0:80
        let ts = self
            .timestamp
            .format(&time::format_description::well_known::Rfc3339)
            .unwrap_or_else(|_| "?".into());
        format!(
            "{ts} [{:>5}]  {}: {}",
            self.level, self.target, self.message
        )
    }
}

#[derive(Debug)]
pub struct LogBus {
    tx: broadcast::Sender<LogLine>,
    buf: Mutex<VecDeque<LogLine>>,
    cap: usize,
}

impl LogBus {
    #[must_use]
    pub fn new(capacity: usize) -> Arc<Self> {
        // 256 = upper bound on concurrent live subscribers' lag tolerance.
        // If a slow client falls behind it'll get a Lagged error and skip
        // ahead — which is what we want for a live tail.
        let (tx, _) = broadcast::channel(256);
        Arc::new(Self {
            tx,
            buf: Mutex::new(VecDeque::with_capacity(capacity)),
            cap: capacity,
        })
    }

    /// Subscribe to new log lines as they're emitted.
    #[must_use]
    pub fn subscribe(&self) -> broadcast::Receiver<LogLine> {
        self.tx.subscribe()
    }

    /// Snapshot of the recent ring buffer (oldest → newest).
    #[must_use]
    pub fn recent(&self) -> Vec<LogLine> {
        self.buf.lock().iter().cloned().collect()
    }

    /// Drop everything in the recent ring buffer.
    pub fn clear(&self) {
        self.buf.lock().clear();
    }

    /// Manually push a synthetic log line (used by the terminal-command
    /// handler so operator commands appear inline in the live tail).
    pub fn push(&self, level: &str, target: &str, message: impl Into<String>) {
        let line = LogLine {
            timestamp: OffsetDateTime::now_utc(),
            level: level.to_string(),
            target: target.to_string(),
            message: message.into(),
        };
        self.record(line);
    }

    fn record(&self, line: LogLine) {
        {
            let mut g = self.buf.lock();
            if g.len() == self.cap {
                g.pop_front();
            }
            g.push_back(line.clone());
        }
        // Send errors are fine — just means no live subscribers right now.
        let _ = self.tx.send(line);
    }
}

/// `tracing_subscriber::Layer` that funnels every event into the LogBus.
///
/// Install once in `main` alongside the existing `fmt::layer()` so console
/// output and the UI tail see the same stream.
pub struct LogBusLayer {
    bus: Arc<LogBus>,
}

impl LogBusLayer {
    #[must_use]
    pub fn new(bus: Arc<LogBus>) -> Self {
        Self { bus }
    }
}

impl<S> Layer<S> for LogBusLayer
where
    S: Subscriber + for<'a> LookupSpan<'a>,
{
    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
        let meta = event.metadata();
        let level = match *meta.level() {
            Level::ERROR => "error",
            Level::WARN => "warn",
            Level::INFO => "info",
            Level::DEBUG => "debug",
            Level::TRACE => "trace",
        };
        let mut visitor = MessageVisitor::default();
        event.record(&mut visitor);
        let line = LogLine {
            timestamp: OffsetDateTime::now_utc(),
            level: level.to_string(),
            target: meta.target().to_string(),
            message: visitor.message,
        };
        self.bus.record(line);
    }
}

#[derive(Default)]
struct MessageVisitor {
    message: String,
}

impl Visit for MessageVisitor {
    fn record_str(&mut self, field: &Field, value: &str) {
        if field.name() == "message" {
            self.message = value.to_string();
        }
    }
    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
        if field.name() == "message" {
            self.message = format!("{value:?}").trim_matches('"').to_string();
        }
    }
}

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

    #[tokio::test(flavor = "current_thread")]
    async fn push_and_recent() {
        let bus = LogBus::new(3);
        bus.push("info", "test", "first");
        bus.push("info", "test", "second");
        bus.push("info", "test", "third");
        bus.push("info", "test", "fourth");
        let r = bus.recent();
        assert_eq!(r.len(), 3);
        assert_eq!(r[0].message, "second");
        assert_eq!(r[2].message, "fourth");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn subscribe_sees_new_lines() {
        let bus = LogBus::new(8);
        let mut rx = bus.subscribe();
        bus.push("info", "test", "live");
        let l = rx.recv().await.unwrap();
        assert_eq!(l.message, "live");
        assert_eq!(l.level, "info");
    }
}