diff --git a/.env.example b/.env.example index 68e704b..2e67990 100644 --- a/.env.example +++ b/.env.example @@ -14,7 +14,7 @@ SYNC_INTERVAL_MINUTES=60 NOTIFICATION_RETENTION_DAYS=0 READING_HISTORY_RETENTION_DAYS=0 AUDIT_RETENTION_DAYS=365 -# Optional Discord webhook URL or ntfy topic URL (for example https://ntfy.sh/my-private-topic). +# Optional Discord webhook URL or ntfy topic URL (for example https://ntfy.sh/my-private-topic). Failed sends retry up to 5 times. ALERT_WEBHOOK_URL= # Optional Bearer token for GET /api/metrics. Leave empty only when metrics are restricted by your network/proxy. METRICS_TOKEN= diff --git a/CHANGELOG.md b/CHANGELOG.md index 45e34d7..a2050ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ - Added Prometheus-compatible metrics and distinct liveness/readiness health probes. - Added SQLite FTS5 indexing for public-content search, maintained automatically as posts change. - Added a repeatable FTS search benchmark and documented PostgreSQL migration decision criteria. +- Added persistent alert delivery with HTTPS validation, exponential-backoff retries, and Prometheus delivery-state metrics. ### Fixed diff --git a/README.md b/README.md index 839096d..3efdf76 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,8 @@ docker compose logs -f web worker | `UPLOAD_ALLOWED_TYPES` | 逗號分隔的 Hub 附件 MIME 白名單。 | | `VIRUS_SCAN_URL` / `VIRUS_SCAN_REQUIRED` | 選用的 HTTP 掃毒服務；服務需回傳 `{ "clean": true }`。若 required 為 `1`，掃毒不可用時拒絕上傳。 | | `SYNC_INTERVAL_MINUTES` | 背景校正同步的間隔，預設 60 分鐘。 | +| `ALERT_WEBHOOK_URL` | 選填的 HTTPS Discord webhook 或 ntfy topic；同步及 webhook 異常會保存後投遞，失敗最多重試 5 次。 | +| `METRICS_TOKEN` | 選填的 `/api/metrics` Bearer Token；未設定時務必由網路／反向代理限制存取。 | | `SEED_MEMOS_*` | 選填；首次啟動時自動建立管理員的第一個 Memos 來源。 | ## 系統架構 @@ -88,7 +90,7 @@ Web 接收使用者操作和 webhook，將同步需求寫入 SQLite 的 `sync_jo ## 正式營運與監控 - `GET /api/health`：供反向代理或監控工具檢查服務與 SQLite 狀態，也會回傳失敗同步工作數與版本。 -- Web、Worker 的事件輸出為 JSON；同步錯誤同時保存於管理頁的「最近系統錯誤」。 +- Web、Worker 的事件輸出為 JSON；同步錯誤同時保存於管理頁的「最近系統錯誤」。告警 webhook 會以 SQLite 佇列投遞、指數退避重試五次，並以 `mebbling_alert_deliveries` metrics 暴露狀態。 - 登入在 15 分鐘內最多嘗試 8 次；webhook 與登入限流資料存於 SQLite，同一份資料庫的多個 Web 容器會共用計數。 - 所有會改變帳號或內容的瀏覽器 POST 都檢查 `Origin`，Webhook 則使用密鑰 URL 驗證，不適用此規則。 - Gitea Actions 工作流程會在推送／標籤時執行型別檢查、測試與 Docker 建置；若設定 `DEPLOY_WEBHOOK_URL` secret，建立 `v*` tag 時會通知部署端。 diff --git a/lib/alerts.ts b/lib/alerts.ts index c87e6b0..06e27c2 100644 --- a/lib/alerts.ts +++ b/lib/alerts.ts @@ -5,7 +5,34 @@ function allowed(bucket: string, seconds = 3600) { if (row && row.reset_at > now) return false; db.prepare("INSERT INTO rate_limits(bucket,count,reset_at) VALUES(?,1,?) ON CONFLICT(bucket) DO UPDATE SET count=count+1,reset_at=excluded.reset_at").run(bucket, now + seconds); return true; } -export async function sendAlert(bucket: string, title: string, message: string) { - const url = process.env.ALERT_WEBHOOK_URL?.trim(); if (!url || !allowed(`alert:${bucket}`)) return false; - try { const isNtfy = /(^|\.)ntfy\.sh\//.test(new URL(url).hostname + new URL(url).pathname); const response = await fetch(url, isNtfy ? { method: "POST", headers: { Title: title, Priority: "high" }, body: message, signal: AbortSignal.timeout(10_000) } : { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ content: `**${title}**\n${message}` }), signal: AbortSignal.timeout(10_000) }); if (!response.ok) throw new Error(`Alert ${response.status}`); return true; } catch { return false; } +export function sendAlert(bucket: string, title: string, message: string) { + if (!process.env.ALERT_WEBHOOK_URL?.trim() || !allowed(`alert:${bucket}`)) return false; + db.prepare("INSERT INTO alert_deliveries(bucket,title,message) VALUES(?,?,?)").run(bucket, title, message); + return true; +} + +type Alert = { id: number; title: string; message: string; attempts: number }; + +export async function deliverNextAlert() { + const candidate = db.prepare("SELECT id FROM alert_deliveries WHERE status='queued' AND run_after<=CURRENT_TIMESTAMP ORDER BY id LIMIT 1").get() as { id: number } | undefined; + if (!candidate) return false; + const claimed = db.prepare("UPDATE alert_deliveries SET status='running',attempts=attempts+1,started_at=CURRENT_TIMESTAMP WHERE id=? AND status='queued'").run(candidate.id); + if (!claimed.changes) return false; + const alert = db.prepare("SELECT id,title,message,attempts FROM alert_deliveries WHERE id=?").get(candidate.id) as Alert; + const url = process.env.ALERT_WEBHOOK_URL?.trim(); + try { + if (!url) throw new Error("ALERT_WEBHOOK_URL is not configured"); + const target = new URL(url); + if (target.protocol !== "https:") throw new Error("Alert webhook must use HTTPS"); + const isNtfy = /(^|\.)ntfy\.sh$/.test(target.hostname); + const response = await fetch(target, isNtfy ? { method: "POST", headers: { Title: alert.title, Priority: "high" }, body: alert.message, signal: AbortSignal.timeout(10_000) } : { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ content: `**${alert.title}**\n${alert.message}` }), signal: AbortSignal.timeout(10_000) }); + if (!response.ok) throw new Error(`Alert endpoint returned ${response.status}`); + db.prepare("UPDATE alert_deliveries SET status='done',finished_at=CURRENT_TIMESTAMP,last_error=NULL WHERE id=?").run(alert.id); + } catch (error) { + const message = error instanceof Error ? error.message : "Alert delivery failed"; + const exhausted = alert.attempts >= 5; + const delayMinutes = Math.min(60, 2 ** Math.max(0, alert.attempts - 1)); + db.prepare("UPDATE alert_deliveries SET status=?,last_error=?,finished_at=CASE WHEN ? THEN CURRENT_TIMESTAMP ELSE NULL END,run_after=CASE WHEN ? THEN run_after ELSE datetime('now', ?) END WHERE id=?").run(exhausted ? "failed" : "queued", message, exhausted ? 1 : 0, exhausted ? 1 : 0, `+${delayMinutes} minutes`, alert.id); + } + return true; } diff --git a/lib/db.ts b/lib/db.ts index e5f1ad5..1df6daf 100644 --- a/lib/db.ts +++ b/lib/db.ts @@ -90,6 +90,12 @@ CREATE TABLE IF NOT EXISTS error_events ( id INTEGER PRIMARY KEY, scope TEXT NOT NULL, message TEXT NOT NULL, context_json TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); +CREATE TABLE IF NOT EXISTS alert_deliveries ( + id INTEGER PRIMARY KEY, bucket TEXT NOT NULL, title TEXT NOT NULL, message TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', attempts INTEGER NOT NULL DEFAULT 0, last_error TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, started_at TEXT, finished_at TEXT, + run_after TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); CREATE TABLE IF NOT EXISTS source_invites ( id INTEGER PRIMARY KEY, source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE, token_hash TEXT UNIQUE NOT NULL, role TEXT NOT NULL DEFAULT 'viewer', expires_at TEXT NOT NULL, @@ -172,6 +178,8 @@ const hasFtsMigration = db.prepare("SELECT 1 FROM schema_migrations WHERE versio if (!hasFtsMigration) { db.prepare("INSERT INTO posts_fts(rowid,content,tags) SELECT id,content,tags_json FROM posts").run(); db.prepare("INSERT INTO schema_migrations(version) VALUES(39)").run(); } db.exec("CREATE INDEX IF NOT EXISTS posts_source_visibility_idx ON posts(source_id,visibility,hidden); CREATE INDEX IF NOT EXISTS posts_author_visibility_idx ON posts(author_id,visibility,hidden);"); db.prepare("INSERT OR IGNORE INTO schema_migrations(version) VALUES(40)").run(); +db.exec("CREATE INDEX IF NOT EXISTS alert_deliveries_status_idx ON alert_deliveries(status,run_after)"); +db.prepare("INSERT OR IGNORE INTO schema_migrations(version) VALUES(41)").run(); const admin = process.env.ADMIN_USERNAME; const adminPassword = process.env.ADMIN_PASSWORD; diff --git a/lib/metrics.ts b/lib/metrics.ts index e95fb05..6dbf074 100644 --- a/lib/metrics.ts +++ b/lib/metrics.ts @@ -4,8 +4,9 @@ const line = (name: string, value: number, labels = "") => `${name}${labels ? `{ export function prometheusMetrics() { const count = (sql: string) => Number((db.prepare(sql).get() as { count: number }).count); const jobs = db.prepare("SELECT status,count(*) AS count FROM sync_jobs GROUP BY status").all() as { status: string; count: number }[]; + const alerts = db.prepare("SELECT status,count(*) AS count FROM alert_deliveries GROUP BY status").all() as { status: string; count: number }[]; const queuedAge = Number((db.prepare("SELECT COALESCE(MAX(strftime('%s','now')-strftime('%s',created_at)),0) AS age FROM sync_jobs WHERE status='queued'").get() as { age: number }).age); const signed = count("SELECT count(*) AS count FROM sources WHERE is_enabled=1 AND webhook_mode='signed'"); const recentWebhook = count("SELECT count(*) AS count FROM sources WHERE is_enabled=1 AND webhook_mode='signed' AND last_webhook_at>=datetime('now','-24 hours')"); const cacheRows = db.prepare("SELECT attachments_json FROM posts").all() as { attachments_json: string }[]; let cacheBytes = 0; for (const row of cacheRows) { try { cacheBytes += (JSON.parse(row.attachments_json) as { url?: string; size?: number }[]).filter((item) => item.url?.startsWith("/uploads/cache/")).reduce((sum, item) => sum + Number(item.size || 0), 0); } catch {} } - return ["# HELP mebbling_sources_enabled Number of enabled sources", "# TYPE mebbling_sources_enabled gauge", line("mebbling_sources_enabled", count("SELECT count(*) AS count FROM sources WHERE is_enabled=1")), "# HELP mebbling_posts_public Number of visible public posts", "# TYPE mebbling_posts_public gauge", line("mebbling_posts_public", count("SELECT count(*) AS count FROM posts WHERE visibility='PUBLIC' AND hidden=0")), "# HELP mebbling_sync_jobs Number of sync jobs by status", "# TYPE mebbling_sync_jobs gauge", ...jobs.map((job) => line("mebbling_sync_jobs", job.count, `status=\"${job.status.replaceAll('"', '')}\"`)), line("mebbling_sync_queue_oldest_seconds", queuedAge), line("mebbling_signed_webhooks", signed), line("mebbling_signed_webhooks_recent_24h", recentWebhook), line("mebbling_attachment_cache_bytes", cacheBytes)].join("\n") + "\n"; + return ["# HELP mebbling_sources_enabled Number of enabled sources", "# TYPE mebbling_sources_enabled gauge", line("mebbling_sources_enabled", count("SELECT count(*) AS count FROM sources WHERE is_enabled=1")), "# HELP mebbling_posts_public Number of visible public posts", "# TYPE mebbling_posts_public gauge", line("mebbling_posts_public", count("SELECT count(*) AS count FROM posts WHERE visibility='PUBLIC' AND hidden=0")), "# HELP mebbling_sync_jobs Number of sync jobs by status", "# TYPE mebbling_sync_jobs gauge", ...jobs.map((job) => line("mebbling_sync_jobs", job.count, `status=\"${job.status.replaceAll('"', '')}\"`)), "# HELP mebbling_alert_deliveries Alert deliveries by status", "# TYPE mebbling_alert_deliveries gauge", ...alerts.map((alert) => line("mebbling_alert_deliveries", alert.count, `status=\"${alert.status.replaceAll('"', '')}\"`)), line("mebbling_sync_queue_oldest_seconds", queuedAge), line("mebbling_signed_webhooks", signed), line("mebbling_signed_webhooks_recent_24h", recentWebhook), line("mebbling_attachment_cache_bytes", cacheBytes)].join("\n") + "\n"; } diff --git a/tests/alerts.test.ts b/tests/alerts.test.ts new file mode 100644 index 0000000..e38fed5 --- /dev/null +++ b/tests/alerts.test.ts @@ -0,0 +1,26 @@ +import assert from "node:assert/strict"; +import { after, test } from "node:test"; +import { randomUUID } from "node:crypto"; +import { rmSync } from "node:fs"; + +const databasePath = `/tmp/mebbling-alerts-${randomUUID()}.db`; +process.env.DATABASE_PATH = databasePath; +process.env.ALERT_WEBHOOK_URL = "https://alerts.example.test/hook"; +delete process.env.HUB_BUILD; + +let database: typeof import("../lib/db").db | undefined; + +after(() => { database?.close(); rmSync(databasePath, { force: true }); rmSync(`${databasePath}-wal`, { force: true }); rmSync(`${databasePath}-shm`, { force: true }); delete process.env.ALERT_WEBHOOK_URL; }); + +test("persists failed alerts for exponential-backoff retry", async () => { + const { db } = await import("../lib/db"); database = db; + const { deliverNextAlert, sendAlert } = await import("../lib/alerts"); + assert.equal(sendAlert("test", "Test alert", "delivery should retry"), true); + const originalFetch = globalThis.fetch; + globalThis.fetch = async () => new Response("unavailable", { status: 503 }); + try { assert.equal(await deliverNextAlert(), true); } finally { globalThis.fetch = originalFetch; } + const delivery = db.prepare("SELECT status,attempts,last_error FROM alert_deliveries").get() as { status: string; attempts: number; last_error: string }; + assert.equal(delivery.status, "queued"); + assert.equal(delivery.attempts, 1); + assert.match(delivery.last_error, /503/); +}); diff --git a/tests/sync.test.ts b/tests/sync.test.ts index ce08ae1..49f186b 100644 --- a/tests/sync.test.ts +++ b/tests/sync.test.ts @@ -16,7 +16,7 @@ test("applies tracked migrations and deduplicates active pull jobs", async () => const { queuePull } = await import("../lib/sync"); const { notify } = await import("../lib/notifications"); const migrations = db.prepare("SELECT version FROM schema_migrations ORDER BY version").all() as { version: number }[]; - assert.deepEqual(migrations.map((item) => item.version), Array.from({ length: 40 }, (_, index) => index + 1)); + assert.deepEqual(migrations.map((item) => item.version), Array.from({ length: 41 }, (_, index) => index + 1)); const userId = Number(db.prepare("INSERT INTO users(username,password_hash) VALUES('sync-test','hash')").run().lastInsertRowid); const sourceId = Number(db.prepare("INSERT INTO sources(user_id,name,base_url,token_encrypted,is_enabled) VALUES(?,?,?,?,1)").run(userId, "Test", "https://example.test", "encrypted").lastInsertRowid); assert.equal(queuePull(sourceId, "manual"), true); diff --git a/worker/index.ts b/worker/index.ts index 4d81088..6a1e1a5 100644 --- a/worker/index.ts +++ b/worker/index.ts @@ -6,7 +6,7 @@ import { decrypt } from "../lib/crypto"; import { createMemo, createRemoteFile, getMemosIdentity, listMemos, memoUrl, setMemoAttachments } from "../lib/memos"; import { recordError } from "../lib/observability"; import { fetchRss } from "../lib/rss"; -import { sendAlert } from "../lib/alerts"; +import { deliverNextAlert, sendAlert } from "../lib/alerts"; type Source = { id: number; user_id: number; base_url: string; token_encrypted: string; remote_user: string | null; integration_type: "memos" | "rss"; rss_feed_url: string | null; is_enabled: number; sync_tags_json: string; sync_from: string | null; sync_to: string | null; sync_attachment_mode: "all" | "images" | "none"; attachment_storage_mode: "remote" | "images" | "all"; attachment_cache_limit_bytes: number; attachment_archive_after_days: number | null; sync_batch_size: number; sync_max_posts: number | null; sync_cursor: string | null; sync_imported_count: number; sync_run_id: string | null }; type Job = { id: number; source_id: number; kind: "pull" | "push"; payload_json: string | null; attempts: number }; @@ -122,4 +122,4 @@ function webhookHealth() { for (const source of stale) void sendAlert(`webhook:${source.id}`, "Mebbling Webhook 未收到事件", `來源「${source.name}」超過 7 天未收到 Webhook；目前仍會以定期 API 同步校正。`); } -let lastRetention = 0, lastWebhookHealth = 0; setInterval(() => { schedule(); if (Date.now() - lastRetention > 86_400_000) { retention(); lastRetention = Date.now(); } if (Date.now() - lastWebhookHealth > 3_600_000) { webhookHealth(); lastWebhookHealth = Date.now(); } void run(); }, 5000); schedule(); retention(); webhookHealth(); void run(); +let lastRetention = 0, lastWebhookHealth = 0; setInterval(() => { schedule(); if (Date.now() - lastRetention > 86_400_000) { retention(); lastRetention = Date.now(); } if (Date.now() - lastWebhookHealth > 3_600_000) { webhookHealth(); lastWebhookHealth = Date.now(); } void run(); void deliverNextAlert(); }, 5000); schedule(); retention(); webhookHealth(); void run(); void deliverNextAlert();