meow_api/
log_stream.rs

1use serde::ser::{SerializeStruct, Serializer};
2use serde::Serialize;
3use tokio::sync::broadcast;
4use tracing_subscriber::Layer;
5
6#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
7pub enum LogLevel {
8    Debug,
9    Info,
10    Warning,
11    Error,
12    Silent,
13}
14
15impl LogLevel {
16    pub fn as_str(self) -> &'static str {
17        match self {
18            LogLevel::Debug => "debug",
19            LogLevel::Info => "info",
20            LogLevel::Warning => "warning",
21            LogLevel::Error => "error",
22            LogLevel::Silent => "silent",
23        }
24    }
25}
26
27#[derive(Clone, Debug)]
28pub struct LogMessage {
29    pub level: LogLevel,
30    pub payload: String,
31    pub time: time::OffsetDateTime,
32}
33
34impl Serialize for LogMessage {
35    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
36        let mut m = s.serialize_struct("LogMessage", 3)?;
37        m.serialize_field("type", self.level.as_str())?;
38        m.serialize_field("payload", &self.payload)?;
39        let ts = self
40            .time
41            .format(&time::format_description::well_known::Rfc3339)
42            .unwrap_or_default();
43        m.serialize_field("time", &ts)?;
44        m.end()
45    }
46}
47
48pub struct LogBroadcastLayer {
49    pub tx: broadcast::Sender<LogMessage>,
50}
51
52impl<S: tracing::Subscriber> Layer<S> for LogBroadcastLayer {
53    fn on_event(
54        &self,
55        event: &tracing::Event<'_>,
56        _ctx: tracing_subscriber::layer::Context<'_, S>,
57    ) {
58        let level = match *event.metadata().level() {
59            tracing::Level::TRACE | tracing::Level::DEBUG => LogLevel::Debug,
60            tracing::Level::INFO => LogLevel::Info,
61            tracing::Level::WARN => LogLevel::Warning,
62            tracing::Level::ERROR => LogLevel::Error,
63        };
64        let mut visitor = MessageVisitor(String::new());
65        event.record(&mut visitor);
66        let msg = LogMessage {
67            level,
68            payload: visitor.0,
69            time: time::OffsetDateTime::now_utc(),
70        };
71        // Non-blocking; Err = no subscribers or channel full — both acceptable.
72        let _ = self.tx.send(msg);
73    }
74}
75
76struct MessageVisitor(String);
77
78impl tracing::field::Visit for MessageVisitor {
79    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
80        if field.name() == "message" {
81            self.0 = format!("{value:?}");
82        }
83    }
84
85    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
86        if field.name() == "message" {
87            self.0 = value.to_string();
88        }
89    }
90}
91
92pub fn parse_log_level(s: &str) -> LogLevel {
93    match s.to_ascii_lowercase().as_str() {
94        "debug" => LogLevel::Debug,
95        "warning" | "warn" => LogLevel::Warning,
96        "error" => LogLevel::Error,
97        "silent" => LogLevel::Silent,
98        // "info" and any unrecognised value default to Info.
99        _ => LogLevel::Info,
100    }
101}
102
103#[cfg(test)]
104mod tests {
105    use super::*;
106    use tracing::subscriber;
107    use tracing_subscriber::prelude::*;
108
109    #[test]
110    fn log_level_str_round_trip_via_parse() {
111        for (s, lvl) in [
112            ("debug", LogLevel::Debug),
113            ("info", LogLevel::Info),
114            ("warning", LogLevel::Warning),
115            ("error", LogLevel::Error),
116            ("silent", LogLevel::Silent),
117        ] {
118            assert_eq!(parse_log_level(s), lvl);
119            assert_eq!(lvl.as_str(), s);
120        }
121    }
122
123    #[test]
124    fn parse_log_level_accepts_warn_alias() {
125        assert_eq!(parse_log_level("warn"), LogLevel::Warning);
126        assert_eq!(parse_log_level("WARN"), LogLevel::Warning);
127    }
128
129    #[test]
130    fn parse_log_level_is_case_insensitive() {
131        assert_eq!(parse_log_level("DEBUG"), LogLevel::Debug);
132        assert_eq!(parse_log_level("Info"), LogLevel::Info);
133        assert_eq!(parse_log_level("Error"), LogLevel::Error);
134    }
135
136    #[test]
137    fn parse_log_level_unknown_input_defaults_to_info() {
138        // Documented behaviour: an unrecognised level → fall back to Info
139        // rather than rejecting the request.
140        assert_eq!(parse_log_level(""), LogLevel::Info);
141        assert_eq!(parse_log_level("nonsense"), LogLevel::Info);
142        assert_eq!(parse_log_level("trace"), LogLevel::Info);
143    }
144
145    #[test]
146    fn log_level_ord_matches_severity_increasing() {
147        // The WS handler filters with `level >= request_level`. The enum
148        // ordering must therefore be Debug < Info < Warning < Error < Silent
149        // (where Silent is the strictest filter — nothing passes).
150        assert!(LogLevel::Debug < LogLevel::Info);
151        assert!(LogLevel::Info < LogLevel::Warning);
152        assert!(LogLevel::Warning < LogLevel::Error);
153        assert!(LogLevel::Error < LogLevel::Silent);
154    }
155
156    #[test]
157    fn log_message_serializes_three_fields() {
158        // 2026-01-02 03:04:05 UTC = 1767322945 seconds since unix epoch.
159        let ts = time::OffsetDateTime::from_unix_timestamp(1_767_322_945).unwrap();
160        let msg = LogMessage {
161            level: LogLevel::Warning,
162            payload: "alerts: thing happened".into(),
163            time: ts,
164        };
165        let json = serde_json::to_value(&msg).unwrap();
166        assert_eq!(json["type"], "warning");
167        assert_eq!(json["payload"], "alerts: thing happened");
168        assert_eq!(json["time"], "2026-01-02T03:02:25Z");
169    }
170
171    #[test]
172    fn layer_forwards_string_message_event() {
173        let (tx, mut rx) = broadcast::channel(8);
174        let layer = LogBroadcastLayer { tx };
175        let registry = tracing_subscriber::registry().with(layer);
176        subscriber::with_default(registry, || {
177            tracing::warn!("hello-world");
178        });
179        let got = rx.try_recv().expect("event must be forwarded");
180        assert_eq!(got.level, LogLevel::Warning);
181        assert!(
182            got.payload.contains("hello-world"),
183            "payload: {}",
184            got.payload
185        );
186    }
187
188    #[test]
189    fn layer_maps_tracing_levels_to_log_levels() {
190        let (tx, mut rx) = broadcast::channel(16);
191        let layer = LogBroadcastLayer { tx };
192        let registry = tracing_subscriber::registry().with(layer);
193        subscriber::with_default(registry, || {
194            tracing::error!("e");
195            tracing::warn!("w");
196            tracing::info!("i");
197            tracing::debug!("d"); // collapses to Debug
198            tracing::trace!("t"); // collapses to Debug
199        });
200        let levels: Vec<LogLevel> = std::iter::from_fn(|| rx.try_recv().ok())
201            .map(|m| m.level)
202            .collect();
203        // Trace+debug both map to Debug; default filter may drop them at the
204        // subscriber level, so accept either {Error, Warning, Info} only or
205        // the full set.
206        assert!(levels.starts_with(&[LogLevel::Error, LogLevel::Warning, LogLevel::Info]));
207        for extra in &levels[3..] {
208            assert_eq!(*extra, LogLevel::Debug);
209        }
210    }
211
212    #[test]
213    fn layer_send_with_no_subscribers_does_not_panic() {
214        // Documented contract: a send Err (no subscribers / channel full) is
215        // acceptable — verify we don't regress to a panicking `.unwrap()`.
216        let (tx, rx) = broadcast::channel(1);
217        drop(rx);
218        let layer = LogBroadcastLayer { tx };
219        let registry = tracing_subscriber::registry().with(layer);
220        subscriber::with_default(registry, || {
221            tracing::info!("payload");
222            tracing::error!("oh no");
223        });
224        // No panic = pass.
225    }
226}
meow_api/log_stream.rs

meow_api/
log_stream.rs
