Name update

crates/core/src/log_bus.rs

@@ -0,0 +1,200 @@
+//! 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]  pxeforge::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");
+    }
+}