yggdrasil/api/logs/
sse.rs

1//! 运行日志实时流 SSE 端点：`GET /api/logs/stream?levels=ERROR,WARN&target=X&q=Y`。
2//!
3//! 鉴权走 [`crate::api::auth::require_admin_from_headers`]：从 cookie 取 session，校验管理员。
4//! 事件类型：
5//! - `log`：JSON [`LogEntry`]（实时事件尚未落库，`id` 恒为 0，不可用作游标）；
6//! - `gap`：broadcast 通道满导致本连接丢事件（`Lagged`），data 为丢失条数文本。
7//!
8//! 过滤按连接参数在服务端完成：levels 逗号分隔（白名单校验，非法级别静默忽略，
9//! 空 = 全部）；target 精确匹配；q 为 message 大小写不敏感子串。
10//! keep-alive comment 每 15s 一次，防反向代理超时关闭空闲连接。
11
12use std::convert::Infallible;
13use std::time::Duration;
14
15use axum::extract::Query;
16use axum::http::{HeaderMap, StatusCode};
17use axum::response::sse::{Event, KeepAlive, Sse};
18use futures::StreamExt;
19use serde::Deserialize;
20use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
21use tokio_stream::wrappers::BroadcastStream;
22
23use crate::api::logs::capture::{self, LogRecord};
24use crate::models::log::LogEntry;
25
26/// SSE 查询参数：`?levels=ERROR,WARN&target=X&q=Y`（全部可选）。
27#[derive(Deserialize)]
28pub struct LogStreamQuery {
29    pub levels: Option<String>,
30    pub target: Option<String>,
31    pub q: Option<String>,
32}
33
34/// SSE handler：cookie admin 鉴权后 subscribe 广播通道，按连接参数过滤推送。
35pub async fn log_stream(
36    headers: HeaderMap,
37    Query(q): Query<LogStreamQuery>,
38) -> Result<Sse<impl futures::Stream<Item = Result<Event, Infallible>>>, (StatusCode, String)> {
39    // 鉴权：cookie → session → admin
40    crate::api::auth::require_admin_from_headers(&headers)
41        .await
42        .map_err(|(status, msg)| (status, msg.to_string()))?;
43
44    // levels：逗号分隔、trim 大写、白名单过滤（非法级别静默忽略——EventSource
45    // 读不到 400 响应体，容错优于报错）；空 vec = 全部级别。
46    let levels: Vec<String> = q
47        .levels
48        .unwrap_or_default()
49        .split(',')
50        .map(|l| l.trim().to_uppercase())
51        .filter(|l| super::VALID_LEVELS.contains(&l.as_str()))
52        .collect();
53    let target = q
54        .target
55        .map(|t| t.trim().to_string())
56        .filter(|t| !t.is_empty());
57    // q 预转小写，匹配时对 message 做大小写不敏感子串（语义对齐 SQL ILIKE）。
58    let needle =
59        q.q.map(|s| s.trim().to_lowercase())
60            .filter(|s| !s.is_empty());
61
62    let stream = BroadcastStream::new(capture::subscribe_live()).filter_map(move |item| {
63        std::future::ready(
64            match item {
65                Ok(record) => {
66                    if !matches_filters(&record, &levels, &target, &needle) {
67                        None
68                    } else {
69                        let entry = LogEntry {
70                            // 实时事件尚未落库，无数据库 id；前端勿用它做游标。
71                            id: 0,
72                            ts: record.ts,
73                            level: record.level,
74                            target: record.target,
75                            message: record.message,
76                        };
77                        Some(
78                            Event::default()
79                                .event("log")
80                                .json_data(entry)
81                                .unwrap_or_else(|_| Event::default().event("log").data("{}")),
82                        )
83                    }
84                }
85                // 消费太慢被广播通道覆盖：通知前端有缺口（应重新拉历史页兜底）。
86                Err(BroadcastStreamRecvError::Lagged(n)) => Some(
87                    Event::default()
88                        .event("gap")
89                        .data(format!("missed {n} log events")),
90                ),
91            }
92            .map(Ok),
93        )
94    });
95
96    // 退出时主动结束，否则这条永不结束的响应会拖住优雅关闭。
97    let stream = stream.take_until(crate::shutdown::requested());
98    Ok(Sse::new(stream).keep_alive(
99        KeepAlive::new()
100            .interval(Duration::from_secs(15))
101            .text("keep-alive"),
102    ))
103}
104
105/// 按连接参数过滤单条记录。
106fn matches_filters(
107    record: &LogRecord,
108    levels: &[String],
109    target: &Option<String>,
110    needle: &Option<String>,
111) -> bool {
112    if !levels.is_empty() && !levels.iter().any(|l| l == &record.level) {
113        return false;
114    }
115    if let Some(t) = target {
116        if &record.target != t {
117            return false;
118        }
119    }
120    if let Some(n) = needle {
121        if !record.message.to_lowercase().contains(n.as_str()) {
122            return false;
123        }
124    }
125    true
126}
127
128#[cfg(all(test, feature = "server"))]
129mod tests {
130    use super::*;
131
132    fn record(level: &str, target: &str, message: &str) -> LogRecord {
133        LogRecord {
134            ts: chrono::Utc::now(),
135            level: level.to_string(),
136            target: target.to_string(),
137            message: message.to_string(),
138        }
139    }
140
141    #[test]
142    fn empty_filters_match_everything() {
143        let r = record("INFO", "yggdrasil::api::posts", "hello");
144        assert!(matches_filters(&r, &[], &None, &None));
145    }
146
147    #[test]
148    fn level_filter_matches_whitelist() {
149        let r = record("WARN", "t", "m");
150        let levels = vec!["ERROR".to_string(), "WARN".to_string()];
151        assert!(matches_filters(&r, &levels, &None, &None));
152        let levels = vec!["ERROR".to_string()];
153        assert!(!matches_filters(&r, &levels, &None, &None));
154    }
155
156    #[test]
157    fn target_filter_is_exact() {
158        let r = record("INFO", "yggdrasil::api::posts", "m");
159        assert!(matches_filters(
160            &r,
161            &[],
162            &Some("yggdrasil::api::posts".to_string()),
163            &None
164        ));
165        assert!(!matches_filters(
166            &r,
167            &[],
168            &Some("yggdrasil::api".to_string()),
169            &None
170        ));
171    }
172
173    #[test]
174    fn query_filter_is_case_insensitive_substring() {
175        let r = record("INFO", "t", "Database Migration Failed");
176        assert!(matches_filters(
177            &r,
178            &[],
179            &None,
180            &Some("migration failed".to_string())
181        ));
182        assert!(!matches_filters(
183            &r,
184            &[],
185            &None,
186            &Some("backup".to_string())
187        ));
188    }
189}
yggdrasil/api/logs/sse.rs

yggdrasil/api/logs/
sse.rs
