feat: dedupe state notifications, 30-day default view and full history

Cargo.lock

@@ -936,7 +936,7 @@ dependencies = [
 
 [[package]]
 name = "newsboard"
-version = "0.1.0"
+version = "0.2.0"
 dependencies = [
  "axum",
  "chrono",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "newsboard"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2021"
 
 [dependencies]

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 {
 <h1>📬 NewsBoard</h1>
 <div class="subtitle">notification center · sugarsource.club</div>
 
+<div class="toolbar">
+    <span id="scopeLabel">近 30 天</span>
+    <span class="status-sep">·</span>
+    <a href="#" class="history-link" id="historyToggle">查看完整历史</a>
+</div>
+
 <div class="status">
     <span class="status-dot" id="statusDot"></span>
     <span id="statusText">connecting...</span>
@@ -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
+        ? '<span class="badge resolved">resolved</span>'
+        : `<span class="badge ${sev}">${notif.severity}</span>`;
 
     card.innerHTML = `
         <div class="meta">
-            <span class="badge ${severityClass(notif.severity)}">${notif.severity}</span>
+            ${badge}
             <span class="topic">${escapeHtml(topicShort)}</span>
             <span class="time">${formatTime(notif.created_at)}</span>
         </div>
         <div class="payload">${renderHighlightedJson(notif.payload)}</div>
     `;
-
-    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 = '<div class="empty">暂无通知</div>';
+            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);

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<String>,
 }
 
 #[derive(Clone, Serialize, Default)]
@@ -50,6 +56,11 @@ struct AppState {
     llm: Arc<RwLock<LlmStatus>>,
 }
 
+#[derive(Deserialize)]
+struct HistoryQuery {
+    all: Option<String>,
+}
+
 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<RwLock<LlmStatus>>, 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<RwLock<LlmStatus>>, 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::<serde_json::Value>(payload)
         .ok()
@@ -144,13 +130,193 @@ fn extract_severity(payload: &str) -> String {
         .unwrap_or_else(|| "info".into())
 }
 
-async fn insert_notification(
+/// 状态类消息的去重键。
+///
+/// 优先使用发布端显式提供的 `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<String> {
+    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(
+        "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
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    // 先补列/合并历史数据，再建索引（旧库此时还没有 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?;
+    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<i64> {
+) -> sqlx::Result<Notification> {
     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 (?, ?, ?, ?)",
         )
@@ -161,10 +327,22 @@ async fn insert_notification(
         .execute(pool)
         .await?;
 
-    let id: (i64,) = sqlx::query_as("SELECT last_insert_rowid()")
+        let (id,): (i64,) = sqlx::query_as("SELECT last_insert_rowid()")
             .fetch_one(pool)
             .await?;
-    Ok(id.0)
+        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<String>) {
@@ -192,15 +370,8 @@ async fn run_mqtt(pool: SqlitePool, tx: broadcast::Sender<String>) {
                     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(&notif).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<Arc<AppState>>) -> impl IntoResponse {
+async fn handle_history(
+    State(state): State<Arc<AppState>>,
+    Query(query): Query<HistoryQuery>,
+) -> 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 {