yggdrasil/api/logs/
mod.rs

1//! 运行日志查看器接口。
2//!
3//! - 本文件：5 个 Dioxus server function（查询 / 导出 / target 列表 / 设置读写），
4//!   双目标编译，DB 逻辑 gate 在 `#[cfg(feature = "server")]` 块内。
5//! - [`crate::api::logs::capture`]：进程内 tracing Layer（日志捕获 mpsc/broadcast），server-only。
6//! - [`crate::api::logs::sse`]：`GET /api/logs/stream` 实时流 axum handler，server-only。
7//!
8//! 数据源：logs 表（迁移 025），由 [`crate::tasks::log_writer`] 批量写入、
9//! [`crate::tasks::log_purge`] 按保留策略裁剪。
10
11// 与 settings 模块一致：Dioxus `#[server]` 宏触发 deprecated/unit 提示，按项目惯例放行。
12#![allow(clippy::unused_unit, deprecated)]
13
14use dioxus::prelude::*;
15
16#[cfg(feature = "server")]
17use crate::api::auth::get_current_admin_user;
18#[cfg(feature = "server")]
19use crate::api::error::AppError;
20#[cfg(feature = "server")]
21use crate::db::pool::get_conn;
22// LogEntry 仅在 server 端的行映射里构造，wasm 端无人消费。
23#[cfg(feature = "server")]
24use crate::models::log::LogEntry;
25use crate::models::log::{LogSettings, LogsPage};
26
27#[cfg(feature = "server")]
28pub mod capture;
29#[cfg(feature = "server")]
30pub mod sse;
31
32/// 日志级别白名单（大写）。levels 空 vec = 全部级别。
33#[cfg(feature = "server")]
34pub(crate) const VALID_LEVELS: [&str; 5] = ["ERROR", "WARN", "INFO", "DEBUG", "TRACE"];
35
36/// `get_logs` 单页条数上限。
37#[cfg(feature = "server")]
38const MAX_PAGE_LIMIT: i32 = 500;
39
40/// `export_logs` 总行数上限。
41#[cfg(feature = "server")]
42const EXPORT_MAX_ROWS: i64 = 10_000;
43
44/// 规范化后的日志筛选条件（levels 已白名单校验，target/query 已 trim）。
45#[cfg(feature = "server")]
46struct LogFilter {
47    levels: Vec<String>,
48    target: Option<String>,
49    query: Option<String>,
50}
51
52#[cfg(feature = "server")]
53impl LogFilter {
54    /// 规范化 + 白名单校验：级别 trim 后大写，不在白名单内返回 BadRequest。
55    fn new(
56        levels: Vec<String>,
57        target: Option<String>,
58        query: Option<String>,
59    ) -> Result<Self, AppError> {
60        let levels: Vec<String> = levels
61            .iter()
62            .map(|l| l.trim().to_uppercase())
63            .filter(|l| !l.is_empty())
64            .collect();
65        for l in &levels {
66            if !VALID_LEVELS.contains(&l.as_str()) {
67                return Err(AppError::BadRequest(format!("非法日志级别: {l}")));
68            }
69        }
70        Ok(Self {
71            levels,
72            target: target
73                .map(|t| t.trim().to_string())
74                .filter(|t| !t.is_empty()),
75            query: query
76                .map(|q| q.trim().to_string())
77                .filter(|q| !q.is_empty()),
78        })
79    }
80
81    /// 生成 WHERE 片段与参数列表（参数按出现顺序编号，$1 起；全部参数化，无拼接值）。
82    /// 返回 (conditions, params)，调用方追加自己的条件后 join 成 WHERE。
83    fn conditions(&self) -> (Vec<String>, Vec<&(dyn tokio_postgres::types::ToSql + Sync)>) {
84        let mut conditions: Vec<String> = Vec::new();
85        let mut params: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = Vec::new();
86        if !self.levels.is_empty() {
87            params.push(&self.levels);
88            conditions.push(format!("level = ANY(${})", params.len()));
89        }
90        if let Some(t) = &self.target {
91            params.push(t);
92            conditions.push(format!("target = ${}", params.len()));
93        }
94        if let Some(q) = &self.query {
95            params.push(q);
96            conditions.push(format!("message ILIKE '%' || ${} || '%'", params.len()));
97        }
98        (conditions, params)
99    }
100}
101
102/// 把 conditions 拼成 WHERE 子句（无条件时为空串）。
103#[cfg(feature = "server")]
104fn where_clause(conditions: &[String]) -> String {
105    if conditions.is_empty() {
106        String::new()
107    } else {
108        format!("WHERE {}", conditions.join(" AND "))
109    }
110}
111
112/// 分页查询运行日志（id DESC 游标分页）。
113///
114/// - `levels`：级别白名单过滤，空 vec = 全部；
115/// - `target`：精确匹配 tracing target；
116/// - `query`：message 子串匹配（参数化 ILIKE）；
117/// - `before_id`：游标（`WHERE id < before_id`），None = 从最新开始；
118/// - `limit`：clamp 到 1..=500。
119#[server(GetLogs, "/api")]
120pub async fn get_logs(
121    levels: Vec<String>,
122    target: Option<String>,
123    query: Option<String>,
124    before_id: Option<i64>,
125    limit: i32,
126) -> Result<LogsPage, ServerFnError> {
127    let _user = get_current_admin_user().await?;
128
129    #[cfg(feature = "server")]
130    {
131        let filter = LogFilter::new(levels, target, query)?;
132        let limit = limit.clamp(1, MAX_PAGE_LIMIT) as i64;
133        let client = get_conn().await.map_err(AppError::db_conn)?;
134
135        let (mut conditions, mut params) = filter.conditions();
136        if let Some(before) = &before_id {
137            params.push(before);
138            conditions.push(format!("id < ${}", params.len()));
139        }
140        params.push(&limit);
141        let limit_idx = params.len();
142
143        let rows = client
144            .query(
145                &format!(
146                    "SELECT id, ts, level, target, message FROM logs {} \
147                     ORDER BY id DESC LIMIT ${limit_idx}",
148                    where_clause(&conditions)
149                ),
150                &params,
151            )
152            .await
153            .map_err(AppError::query)?;
154
155        let entries: Vec<LogEntry> = rows
156            .iter()
157            .map(|r| LogEntry {
158                id: r.get(0),
159                ts: r.get(1),
160                level: r.get(2),
161                target: r.get(3),
162                message: r.get(4),
163            })
164            .collect();
165
166        // 满页才给游标：最后一行的 id 即下一页的 before_id。
167        let next_cursor = if entries.len() as i64 == limit {
168            entries.last().map(|e| e.id)
169        } else {
170            None
171        };
172
173        Ok(LogsPage {
174            entries,
175            next_cursor,
176            dropped: capture::dropped_count(),
177        })
178    }
179
180    #[cfg(not(feature = "server"))]
181    {
182        Ok(LogsPage {
183            entries: Vec::new(),
184            next_cursor: None,
185            dropped: 0,
186        })
187    }
188}
189
190/// 导出运行日志为纯文本（每行 `[RFC3339] LEVEL target: message`）。
191///
192/// 与 `get_logs` 同一筛选语义，按 id ASC（时间正序），上限 10000 行。
193#[server(ExportLogs, "/api")]
194pub async fn export_logs(
195    levels: Vec<String>,
196    target: Option<String>,
197    query: Option<String>,
198) -> Result<String, ServerFnError> {
199    let _user = get_current_admin_user().await?;
200
201    #[cfg(feature = "server")]
202    {
203        use std::fmt::Write as _;
204
205        let filter = LogFilter::new(levels, target, query)?;
206        let client = get_conn().await.map_err(AppError::db_conn)?;
207        let (conditions, params) = filter.conditions();
208
209        let rows = client
210            .query(
211                &format!(
212                    "SELECT id, ts, level, target, message FROM logs {} \
213                     ORDER BY id ASC LIMIT {EXPORT_MAX_ROWS}",
214                    where_clause(&conditions)
215                ),
216                &params,
217            )
218            .await
219            .map_err(AppError::query)?;
220
221        let mut out = String::new();
222        for r in &rows {
223            let ts: chrono::DateTime<chrono::Utc> = r.get(1);
224            let level: String = r.get(2);
225            let target: String = r.get(3);
226            let message: String = r.get(4);
227            // 写 String 不会失败，忽略 fmt::Result。
228            let _ = writeln!(
229                out,
230                "[{}] {} {}: {}",
231                ts.to_rfc3339(),
232                level,
233                target,
234                message
235            );
236        }
237        Ok(out)
238    }
239
240    #[cfg(not(feature = "server"))]
241    {
242        Ok(String::new())
243    }
244}
245
246/// 日志 target 去重列表（前端筛选项）。
247///
248/// 走 moka 缓存（60s TTL）：DISTINCT 全表扫在日志表上不值得每开页面跑一次。
249#[server(GetLogTargets, "/api")]
250pub async fn get_log_targets() -> Result<Vec<String>, ServerFnError> {
251    let _user = get_current_admin_user().await?;
252
253    #[cfg(feature = "server")]
254    {
255        if let Some(cached) = crate::cache::get_log_targets().await {
256            return Ok(cached);
257        }
258        let client = get_conn().await.map_err(AppError::db_conn)?;
259        let rows = client
260            .query("SELECT DISTINCT target FROM logs ORDER BY target", &[])
261            .await
262            .map_err(AppError::query)?;
263        let targets: Vec<String> = rows.iter().map(|r| r.get(0)).collect();
264        crate::cache::set_log_targets(targets.clone()).await;
265        Ok(targets)
266    }
267
268    #[cfg(not(feature = "server"))]
269    {
270        Ok(Vec::new())
271    }
272}
273
274/// 读取日志查看器配置。
275///
276/// settings 表缺失键时回退到默认值，保证向后兼容（trash 模式，不做 env 播种）。
277#[server(GetLogSettings, "/api")]
278pub async fn get_log_settings() -> Result<LogSettings, ServerFnError> {
279    let _user = get_current_admin_user().await?;
280
281    #[cfg(feature = "server")]
282    {
283        let client = get_conn().await.map_err(AppError::db_conn)?;
284
285        let retention_days: i32 = client
286            .query_opt(
287                "SELECT value FROM settings WHERE key = 'logs_retention_days'",
288                &[],
289            )
290            .await
291            .map_err(AppError::query)?
292            .and_then(|r| r.get::<_, String>("value").parse().ok())
293            .unwrap_or(crate::models::log::DEFAULT_LOGS_RETENTION_DAYS);
294
295        let max_rows: i32 = client
296            .query_opt(
297                "SELECT value FROM settings WHERE key = 'logs_max_rows'",
298                &[],
299            )
300            .await
301            .map_err(AppError::query)?
302            .and_then(|r| r.get::<_, String>("value").parse().ok())
303            .unwrap_or(crate::models::log::DEFAULT_LOGS_MAX_ROWS);
304
305        Ok(LogSettings {
306            retention_days: LogSettings::clamp_retention(retention_days),
307            max_rows: LogSettings::clamp_max_rows(max_rows),
308        })
309    }
310
311    #[cfg(not(feature = "server"))]
312    {
313        Ok(LogSettings::default())
314    }
315}
316
317/// 更新日志查看器配置。
318///
319/// 两个值都会被 clamp 到合法范围后写入。
320#[server(UpdateLogSettings, "/api")]
321pub async fn update_log_settings(
322    retention_days: i32,
323    max_rows: i32,
324) -> Result<LogSettings, ServerFnError> {
325    let _user = get_current_admin_user().await?;
326
327    let retention_days = LogSettings::clamp_retention(retention_days);
328    let max_rows = LogSettings::clamp_max_rows(max_rows);
329
330    #[cfg(feature = "server")]
331    {
332        let client = get_conn().await.map_err(AppError::db_conn)?;
333
334        client
335            .execute(
336                "INSERT INTO settings (key, value, updated_at) VALUES ('logs_retention_days', $1, NOW())
337                 ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()",
338                &[&retention_days.to_string()],
339            )
340            .await
341            .map_err(AppError::query)?;
342
343        client
344            .execute(
345                "INSERT INTO settings (key, value, updated_at) VALUES ('logs_max_rows', $1, NOW())
346                 ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()",
347                &[&max_rows.to_string()],
348            )
349            .await
350            .map_err(AppError::query)?;
351
352        // target 在 capture 排除名单内：此日志只到控制台，不进查看器管道。
353        tracing::info!(
354            "Log settings updated: retention_days={}, max_rows={}",
355            retention_days,
356            max_rows
357        );
358    }
359
360    Ok(LogSettings {
361        retention_days,
362        max_rows,
363    })
364}
yggdrasil/api/logs/mod.rs

yggdrasil/api/logs/
mod.rs
