feat: retry failed alert deliveries

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;
 }

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;

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";
 }