yggdrasil/api/logs/
sse.rs

1//! 运行日志实时流 SSE 端点：`GET /api/logs/stream?levels=ERROR,WARN&target=X&q=Y`。
2//!
3//! 鉴权镜像 [`crate::api::database::export`]：从 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::auth::session::parse_session_token;
25use crate::models::log::LogEntry;
26
27/// SSE 查询参数：`?levels=ERROR,WARN&target=X&q=Y`（全部可选）。
28#[derive(Deserialize)]
29pub struct LogStreamQuery {
30    pub levels: Option<String>,
31    pub target: Option<String>,
32    pub q: Option<String>,
33}
34
35/// SSE handler：cookie admin 鉴权后 subscribe 广播通道，按连接参数过滤推送。
36pub async fn log_stream(
37    headers: HeaderMap,
38    Query(q): Query<LogStreamQuery>,
39) -> Result<Sse<impl futures::Stream<Item = Result<Event, Infallible>>>, (StatusCode, String)> {
40    // 鉴权：cookie → session → admin（镜像 export.rs）
41    let cookie_header = headers
42        .get("cookie")
43        .and_then(|h| h.to_str().ok())
44        .unwrap_or("");
45    let token = parse_session_token(cookie_header).map(str::to_string);
46    let token = match token {
47        Some(t) => t,
48        None => return Err((StatusCode::UNAUTHORIZED, "未登录".to_string())),
49    };
50    let user = match crate::api::auth::get_user_by_token(&token).await {
51        Ok(Some(u)) => u,
52        _ => return Err((StatusCode::UNAUTHORIZED, "会话已过期".to_string())),
53    };
54    if user.role != crate::models::user::UserRole::Admin {
55        return Err((StatusCode::FORBIDDEN, "权限不足".to_string()));
56    }
57
58    // levels：逗号分隔、trim 大写、白名单过滤（非法级别静默忽略——EventSource
59    // 读不到 400 响应体，容错优于报错）；空 vec = 全部级别。
60    let levels: Vec<String> = q
61        .levels
62        .unwrap_or_default()
63        .split(',')
64        .map(|l| l.trim().to_uppercase())
65        .filter(|l| super::VALID_LEVELS.contains(&l.as_str()))
66        .collect();
67    let target = q
68        .target
69        .map(|t| t.trim().to_string())
70        .filter(|t| !t.is_empty());
71    // q 预转小写，匹配时对 message 做大小写不敏感子串（语义对齐 SQL ILIKE）。
72    let needle =
73        q.q.map(|s| s.trim().to_lowercase())
74            .filter(|s| !s.is_empty());
75
76    let stream = BroadcastStream::new(capture::subscribe_live()).filter_map(move |item| {
77        std::future::ready(
78            match item {
79                Ok(record) => {
80                    if !matches_filters(&record, &levels, &target, &needle) {
81                        None
82                    } else {
83                        let entry = LogEntry {
84                            // 实时事件尚未落库，无数据库 id；前端勿用它做游标。
85                            id: 0,
86                            ts: record.ts,
87                            level: record.level,
88                            target: record.target,
89                            message: record.message,
90                        };
91                        Some(
92                            Event::default()
93                                .event("log")
94                                .json_data(entry)
95                                .unwrap_or_else(|_| Event::default().event("log").data("{}")),
96                        )
97                    }
98                }
99                // 消费太慢被广播通道覆盖：通知前端有缺口（应重新拉历史页兜底）。
100                Err(BroadcastStreamRecvError::Lagged(n)) => Some(
101                    Event::default()
102                        .event("gap")
103                        .data(format!("missed {n} log events")),
104                ),
105            }
106            .map(Ok),
107        )
108    });
109
110    Ok(Sse::new(stream).keep_alive(
111        KeepAlive::new()
112            .interval(Duration::from_secs(15))
113            .text("keep-alive"),
114    ))
115}
116
117/// 按连接参数过滤单条记录。
118fn matches_filters(
119    record: &LogRecord,
120    levels: &[String],
121    target: &Option<String>,
122    needle: &Option<String>,
123) -> bool {
124    if !levels.is_empty() && !levels.iter().any(|l| l == &record.level) {
125        return false;
126    }
127    if let Some(t) = target {
128        if &record.target != t {
129            return false;
130        }
131    }
132    if let Some(n) = needle {
133        if !record.message.to_lowercase().contains(n.as_str()) {
134            return false;
135        }
136    }
137    true
138}
139
140#[cfg(all(test, feature = "server"))]
141mod tests {
142    use super::*;
143
144    fn record(level: &str, target: &str, message: &str) -> LogRecord {
145        LogRecord {
146            ts: chrono::Utc::now(),
147            level: level.to_string(),
148            target: target.to_string(),
149            message: message.to_string(),
150        }
151    }
152
153    #[test]
154    fn empty_filters_match_everything() {
155        let r = record("INFO", "yggdrasil::api::posts", "hello");
156        assert!(matches_filters(&r, &[], &None, &None));
157    }
158
159    #[test]
160    fn level_filter_matches_whitelist() {
161        let r = record("WARN", "t", "m");
162        let levels = vec!["ERROR".to_string(), "WARN".to_string()];
163        assert!(matches_filters(&r, &levels, &None, &None));
164        let levels = vec!["ERROR".to_string()];
165        assert!(!matches_filters(&r, &levels, &None, &None));
166    }
167
168    #[test]
169    fn target_filter_is_exact() {
170        let r = record("INFO", "yggdrasil::api::posts", "m");
171        assert!(matches_filters(
172            &r,
173            &[],
174            &Some("yggdrasil::api::posts".to_string()),
175            &None
176        ));
177        assert!(!matches_filters(
178            &r,
179            &[],
180            &Some("yggdrasil::api".to_string()),
181            &None
182        ));
183    }
184
185    #[test]
186    fn query_filter_is_case_insensitive_substring() {
187        let r = record("INFO", "t", "Database Migration Failed");
188        assert!(matches_filters(
189            &r,
190            &[],
191            &None,
192            &Some("migration failed".to_string())
193        ));
194        assert!(!matches_filters(
195            &r,
196            &[],
197            &None,
198            &Some("backup".to_string())
199        ));
200    }
201}
yggdrasil/api/logs/sse.rs

yggdrasil/api/logs/
sse.rs
