From cb58b68932839b1dac9113c5490eea881352ce26 Mon Sep 17 00:00:00 2001 From: akn Date: Mon, 2 Nov 2026 14:01:33 +0900 Subject: [PATCH] feat: dedupe state notifications, 30-day default view and full history --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/index.html | 142 +++++++++++++++++++---- src/main.rs | 301 +++++++++++++++++++++++++++++++++++++++---------- 4 files changed, 366 insertions(+), 81 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 56e742e..ad1f010 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -936,7 +936,7 @@ dependencies = [ [[package]] name = "newsboard" -version = "0.1.0" +version = "0.2.0" dependencies = [ "axum", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 6f0598e..d17fb06 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "newsboard" -version = "0.1.0" +version = "0.2.0" edition = "2021" [dependencies] diff --git a/src/index.html b/src/index.html index f077f15..e738d54 100644 --- a/src/index.html +++ b/src/index.html @@ -27,7 +27,26 @@ h1 { .subtitle { color: #6c7086; font-size: 0.8rem; - margin-bottom: 2rem; + margin-bottom: 1rem; +} + +.toolbar { + display: flex; + align-items: center; + gap: 0.8rem; + margin-bottom: 1.5rem; + font-size: 0.75rem; + color: #6c7086; +} + +.history-link { + color: #89b4fa; + text-decoration: none; + border-bottom: 1px dashed #89b4fa; +} + +.history-link:hover { + color: #b4befe; } #notifications { @@ -55,6 +74,10 @@ h1 { .card.medium { border-left-color: #f9e2af; background: #2a281e; } .card.info { border-left-color: #89b4fa; background: #1e242a; } +/* 状态类卡片收敛后的样式 */ +.card.resolved { border-left-color: #a6e3a1; background: #1e2a24; opacity: 0.75; } +.badge.resolved { background: #a6e3a1; color: #1e1f2c; } + .card .meta { display: flex; align-items: center; @@ -175,6 +198,12 @@ h1 {

📬 NewsBoard

notification center · sugarsource.club
+
+ 近 30 天 + · + 查看完整历史 +
+
connecting... @@ -198,8 +227,12 @@ h1 { const el = document.getElementById('notifications'); const statusDot = document.getElementById('statusDot'); const statusText = document.getElementById('statusText'); -const MAX_VISIBLE = 200; -let hasHistory = false; +const scopeLabel = document.getElementById('scopeLabel'); +const historyToggle = document.getElementById('historyToggle'); +let showAll = false; +let maxVisible = 500; +// (topic + dedupe_key) -> card element，用于状态类消息就地更新 +const cards = new Map(); function severityClass(s) { const map = { critical: 'critical', high: 'high', medium: 'medium' }; @@ -235,27 +268,37 @@ function escapeHtml(text) { return d.innerHTML; } -function addCard(notif, animate) { - const card = document.createElement('div'); - card.className = `card ${severityClass(notif.severity)}`; +function cardKey(notif) { + return notif.dedupe_key ? notif.topic + '\u001f' + notif.dedupe_key : null; +} + +function isResolved(payload) { + try { + return JSON.parse(payload).resolved === true; + } catch { + return false; + } +} + +function renderCard(card, notif, animate) { + const sev = severityClass(notif.severity); + const resolved = isResolved(notif.payload); + card.className = `card ${sev}${resolved ? ' resolved' : ''}`; if (animate) card.classList.add('new-highlight'); const topicShort = notif.topic.replace(/^notify\//, ''); + const badge = resolved + ? 'resolved' + : `${notif.severity}`; card.innerHTML = `
- ${notif.severity} + ${badge} ${escapeHtml(topicShort)} ${formatTime(notif.created_at)}
${renderHighlightedJson(notif.payload)}
`; - - el.insertBefore(card, el.firstChild); - - while (el.children.length > MAX_VISIBLE) { - el.removeChild(el.lastChild); - } } function clearEmpty() { @@ -263,15 +306,68 @@ function clearEmpty() { if (empty) empty.remove(); } -async function loadHistory() { +function trim() { + while (el.children.length > maxVisible) { + const last = el.lastElementChild; + if (!last) break; + if (last.dataset && last.dataset.key) cards.delete(last.dataset.key); + el.removeChild(last); + } +} + +function addCard(notif, animate) { + clearEmpty(); + const key = cardKey(notif); + + // 状态类消息：已有同 key 卡片则就地更新并移到顶部 + if (key) { + const existing = cards.get(key); + if (existing) { + existing.classList.remove('new-highlight'); + renderCard(existing, notif, false); + if (animate) { + void existing.offsetWidth; + existing.classList.add('new-highlight'); + } + el.insertBefore(existing, el.firstChild); + trim(); + return; + } + } + + const card = document.createElement('div'); + renderCard(card, notif, animate); + if (key) { + card.dataset.key = key; + cards.set(key, card); + } + el.insertBefore(card, el.firstChild); + trim(); +} + +function resetList() { + el.innerHTML = ''; + cards.clear(); +} + +async function loadHistory(all) { + showAll = all; + maxVisible = all ? 5000 : 500; + scopeLabel.textContent = all ? '最近一年' : '近 30 天'; + historyToggle.textContent = all ? '只看近 30 天' : '查看完整历史'; + try { - const res = await fetch('/history'); + const res = await fetch('/history' + (all ? '?all=1' : '')); const data = await res.json(); - if (data.length > 0) { - clearEmpty(); - data.reverse().forEach(n => addCard(n, false)); + resetList(); + if (data.length === 0) { + el.innerHTML = '
暂无通知
'; + return; + } + // 服务端按时间倒序返回；从末尾开始 prepend，最终仍是倒序 + for (let i = data.length - 1; i >= 0; i--) { + addCard(data[i], false); } - hasHistory = true; } catch (e) { console.error('history fetch failed:', e); } @@ -288,7 +384,6 @@ function connectSSE() { evtSource.onmessage = (e) => { try { const notif = JSON.parse(e.data); - clearEmpty(); addCard(notif, true); } catch (err) { console.error('SSE parse error:', err); @@ -301,6 +396,11 @@ function connectSSE() { }; } +historyToggle.addEventListener('click', (e) => { + e.preventDefault(); + loadHistory(!showAll); +}); + const llmDot = document.getElementById('llmDot'); const llmText = document.getElementById('llmText'); @@ -323,7 +423,7 @@ async function pollLlm() { } } -loadHistory(); +loadHistory(false); connectSSE(); pollLlm(); setInterval(pollLlm, 15000); diff --git a/src/main.rs b/src/main.rs index 7186c40..7284ff5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,5 +1,5 @@ use axum::{ - extract::State, + extract::{Query, State}, response::{ sse::{Event, KeepAlive, Sse}, IntoResponse, @@ -7,7 +7,7 @@ use axum::{ routing::get, Router, }; -use chrono::Utc; +use chrono::{Duration, Utc}; use rumqttc::{AsyncClient, MqttOptions, Packet, QoS}; use rust_embed::RustEmbed; use serde::{Deserialize, Serialize}; @@ -16,13 +16,18 @@ use sqlx::SqlitePool; use std::convert::Infallible; use std::sync::Arc; use std::sync::RwLock; -use std::time::{Duration, Instant}; +use std::time::{Duration as StdDuration, Instant}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::sync::broadcast; use tokio_stream::StreamExt; use tokio_stream::wrappers::BroadcastStream; use tracing::{error, info, warn}; +/// 默认只展示最近 30 天，点「历史」查看完整记录 +const VISIBLE_DAYS: i64 = 30; +/// 数据库保留 1 年 +const RETENTION_DAYS: i64 = 366; + #[derive(RustEmbed)] #[folder = "src/"] struct Assets; @@ -34,6 +39,7 @@ struct Notification { payload: String, severity: String, created_at: String, + dedupe_key: Option, } #[derive(Clone, Serialize, Default)] @@ -50,6 +56,11 @@ struct AppState { llm: Arc>, } +#[derive(Deserialize)] +struct HistoryQuery { + all: Option, +} + fn parse_http_url(url: &str) -> Option<(String, u16, String)> { let rest = url.strip_prefix("http://")?; let (authority, path) = match rest.find('/') { @@ -81,7 +92,7 @@ async fn probe_llm(host: &str, port: u16, path: &str) -> bool { ) }; matches!( - tokio::time::timeout(Duration::from_secs(5), attempt).await, + tokio::time::timeout(StdDuration::from_secs(5), attempt).await, Ok(Ok(true)) ) } @@ -95,7 +106,7 @@ fn spawn_llm_probe(status: Arc>, url: String) { return; } }; - let mut ticker = tokio::time::interval(Duration::from_secs(15)); + let mut ticker = tokio::time::interval(StdDuration::from_secs(15)); loop { ticker.tick().await; let start = Instant::now(); @@ -112,31 +123,6 @@ fn spawn_llm_probe(status: Arc>, url: String) { }); } -async fn init_db(pool: &SqlitePool) -> sqlx::Result<()> { - sqlx::query( - "CREATE TABLE IF NOT EXISTS notifications ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - topic TEXT NOT NULL, - payload TEXT NOT NULL, - severity TEXT NOT NULL DEFAULT 'info', - created_at TEXT NOT NULL - )", - ) - .execute(pool) - .await?; - - // keep only latest 1000 - sqlx::query( - "DELETE FROM notifications WHERE id NOT IN ( - SELECT id FROM notifications ORDER BY id DESC LIMIT 1000 - )", - ) - .execute(pool) - .await?; - - Ok(()) -} - fn extract_severity(payload: &str) -> String { serde_json::from_str::(payload) .ok() @@ -144,27 +130,219 @@ fn extract_severity(payload: &str) -> String { .unwrap_or_else(|| "info".into()) } -async fn insert_notification( - pool: &SqlitePool, - topic: &str, - payload: &str, -) -> sqlx::Result { - let severity = extract_severity(payload); - let now = Utc::now().to_rfc3339(); +/// 状态类消息的去重键。 +/// +/// 优先使用发布端显式提供的 `dedupe_key`；旧消息没有该字段时按 topic + payload +/// 语义推导，这样迁移时能把历史上重复的心跳/状态归并成一条。 +/// 事件类消息（start/result/retry/autofix/reboot）不返回 key，继续逐条保留。 +/// 注意：`start`/`result` 的 payload 也带 `type:"monthly"`，所以必须结合 topic 判断， +/// 不能只看 payload 字段。 +fn extract_dedupe_key(topic: &str, payload: &str) -> Option { + let v: serde_json::Value = serde_json::from_str(payload).ok()?; + + if let Some(key) = v.get("dedupe_key").and_then(|x| x.as_str()) { + if !key.is_empty() { + return Some(key.to_string()); + } + } + + let typ = v.get("type").and_then(|x| x.as_str()).unwrap_or(""); + let is_announce = topic.ends_with("/announce"); + let is_poll = topic.ends_with("/poll"); + let is_result = topic.ends_with("/result"); + let is_waiting = topic.ends_with("/waiting"); + + if is_poll { + return Some("security-poll".into()); + } + if is_waiting || typ == "waiting-nixpkgs" { + return Some("waiting:nixpkgs".into()); + } + if is_announce { + if typ == "monthly" { + return Some("announce:monthly".into()); + } + if let Some(trigger) = v.get("trigger").and_then(|x| x.as_str()) { + match trigger { + "advisory" => { + let id = v + .get("ghsa_id") + .and_then(|x| x.as_str()) + .unwrap_or("unknown"); + return Some(format!("announce:advisory:{id}")); + } + "package-release" => return Some("announce:package-release".into()), + _ => {} + } + } + return None; + } + if is_result && v.get("pending").and_then(|x| x.as_bool()) == Some(true) { + return Some("pending:upstream".into()); + } + + None +} + +async fn prune_old(pool: &SqlitePool) -> sqlx::Result<()> { + let cutoff = (Utc::now() - Duration::days(RETENTION_DAYS)).to_rfc3339(); + sqlx::query("DELETE FROM notifications WHERE created_at < ?") + .bind(cutoff) + .execute(pool) + .await?; + Ok(()) +} + +async fn init_db(pool: &SqlitePool) -> sqlx::Result<()> { sqlx::query( - "INSERT INTO notifications (topic, payload, severity, created_at) VALUES (?, ?, ?, ?)", + "CREATE TABLE IF NOT EXISTS notifications ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + topic TEXT NOT NULL, + payload TEXT NOT NULL, + severity TEXT NOT NULL DEFAULT 'info', + created_at TEXT NOT NULL, + dedupe_key TEXT + )", ) - .bind(topic) - .bind(payload) - .bind(&severity) - .bind(&now) .execute(pool) .await?; - let id: (i64,) = sqlx::query_as("SELECT last_insert_rowid()") + // 先补列/合并历史数据，再建索引（旧库此时还没有 dedupe_key 列） + migrate_dedupe(pool).await?; + + // SQLite 唯一索引允许任意多个 NULL，所以事件类消息不受影响 + sqlx::query( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_notifications_dedupe + ON notifications(topic, dedupe_key)", + ) + .execute(pool) + .await?; + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_notifications_created + ON notifications(created_at DESC, id DESC)", + ) + .execute(pool) + .await?; + + prune_old(pool).await?; + Ok(()) +} + +/// 一次性迁移：补 dedupe_key 列，回填历史状态键并合并重复行。 +/// 用 PRAGMA user_version 保证只跑一次。 +async fn migrate_dedupe(pool: &SqlitePool) -> sqlx::Result<()> { + let (version,): (i64,) = sqlx::query_as("PRAGMA user_version") .fetch_one(pool) .await?; - Ok(id.0) + if version >= 1 { + return Ok(()); + } + + let columns: Vec<(i64, String)> = + sqlx::query_as("SELECT cid, name FROM pragma_table_info('notifications')") + .fetch_all(pool) + .await?; + if !columns.iter().any(|(_, name)| name == "dedupe_key") { + sqlx::query("ALTER TABLE notifications ADD COLUMN dedupe_key TEXT") + .execute(pool) + .await?; + } + + // 回填历史状态键 + let rows: Vec<(i64, String, String)> = + sqlx::query_as("SELECT id, topic, payload FROM notifications WHERE dedupe_key IS NULL") + .fetch_all(pool) + .await?; + for (id, topic, payload) in rows { + if let Some(key) = extract_dedupe_key(&topic, &payload) { + sqlx::query("UPDATE notifications SET dedupe_key = ? WHERE id = ?") + .bind(key) + .bind(id) + .execute(pool) + .await?; + } + } + + // 同一 (topic, dedupe_key) 只保留最新一条 + sqlx::query( + "DELETE FROM notifications + WHERE dedupe_key IS NOT NULL + AND id NOT IN ( + SELECT MAX(id) FROM notifications + WHERE dedupe_key IS NOT NULL + GROUP BY topic, dedupe_key + )", + ) + .execute(pool) + .await?; + + sqlx::query("PRAGMA user_version = 1") + .execute(pool) + .await?; + info!("dedupe migration completed"); + Ok(()) +} + +async fn store_notification( + pool: &SqlitePool, + topic: &str, + payload: &str, +) -> sqlx::Result { + let severity = extract_severity(payload); + let dedupe_key = extract_dedupe_key(topic, payload); + let now = Utc::now().to_rfc3339(); + + let id: i64 = if let Some(key) = &dedupe_key { + sqlx::query( + "INSERT INTO notifications (topic, payload, severity, created_at, dedupe_key) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(topic, dedupe_key) DO UPDATE SET + payload = excluded.payload, + severity = excluded.severity, + created_at = excluded.created_at", + ) + .bind(topic) + .bind(payload) + .bind(&severity) + .bind(&now) + .bind(key) + .execute(pool) + .await?; + + let (id,): (i64,) = + sqlx::query_as("SELECT id FROM notifications WHERE topic = ? AND dedupe_key = ?") + .bind(topic) + .bind(key) + .fetch_one(pool) + .await?; + id + } else { + sqlx::query( + "INSERT INTO notifications (topic, payload, severity, created_at) VALUES (?, ?, ?, ?)", + ) + .bind(topic) + .bind(payload) + .bind(&severity) + .bind(&now) + .execute(pool) + .await?; + + let (id,): (i64,) = sqlx::query_as("SELECT last_insert_rowid()") + .fetch_one(pool) + .await?; + id + }; + + let _ = prune_old(pool).await; + + Ok(Notification { + id, + topic: topic.to_string(), + payload: payload.to_string(), + severity, + created_at: now, + dedupe_key, + }) } async fn run_mqtt(pool: SqlitePool, tx: broadcast::Sender) { @@ -192,15 +370,8 @@ async fn run_mqtt(pool: SqlitePool, tx: broadcast::Sender) { let topic = publish.topic.clone(); let payload = String::from_utf8_lossy(&publish.payload).to_string(); - match insert_notification(&pool_clone, &topic, &payload).await { - Ok(id) => { - let notif = Notification { - id, - topic: topic.clone(), - payload: payload.clone(), - severity: extract_severity(&payload), - created_at: Utc::now().to_rfc3339(), - }; + match store_notification(&pool_clone, &topic, &payload).await { + Ok(notif) => { let json = serde_json::to_string(¬if).unwrap(); let _ = tx_clone.send(json); info!(topic = %topic, "notification stored"); @@ -234,10 +405,27 @@ async fn handle_root() -> impl IntoResponse { axum::response::Html(content) } -async fn handle_history(State(state): State>) -> impl IntoResponse { +async fn handle_history( + State(state): State>, + Query(query): Query, +) -> impl IntoResponse { + let all = matches!(query.all.as_deref(), Some("1") | Some("true")); + let (days, limit) = if all { + (RETENTION_DAYS, 5000) + } else { + (VISIBLE_DAYS, 500) + }; + let cutoff = (Utc::now() - Duration::days(days)).to_rfc3339(); + let rows = sqlx::query_as::<_, Notification>( - "SELECT id, topic, payload, severity, created_at FROM notifications ORDER BY id DESC LIMIT 200", + "SELECT id, topic, payload, severity, created_at, dedupe_key + FROM notifications + WHERE created_at >= ? + ORDER BY created_at DESC, id DESC + LIMIT ?", ) + .bind(cutoff) + .bind(limit) .fetch_all(&state.pool) .await .unwrap_or_default(); @@ -271,9 +459,6 @@ async fn main() { let db_path = std::env::var("NEWSBOARD_DB") .unwrap_or_else(|_| "/var/lib/newsboard/news.db".into()); - let db_path = std::env::var("NEWSBOARD_DB") - .unwrap_or_else(|_| "/var/lib/newsboard/news.db".into()); - let url = if db_path == ":memory:" || db_path == "" { "sqlite::memory:".to_string() } else {