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 {