feat(ops): generalize operator alerts and add the daily digest

.env.example

@@ -59,10 +59,13 @@ EMAIL_FROM_ADDRESS=bottleneck <noreply@bneck.com>
 
 REDIS_URL=redis://localhost:6379
 
-# Operator chat for webhook auto-disable / enqueue-failure alerts. Optional;
-# falls back to BEARER_ADMIN_TELEGRAM_ID. Alerting no-ops if neither is set or
-# outside production.
+# Operator chat (Bot API id; the bot must be a member) for alerts: webhook
+# endpoint auto-disabled, enqueue failed, Redis degraded, worker loop errors,
+# migration failed at boot, and the daily digest. Optional; falls back to
+# BEARER_ADMIN_TELEGRAM_ID. Alerting no-ops if neither is set or outside
+# production. DIGEST_HOUR_UTC is the hour the digest is sent.
 ALERT_TELEGRAM_CHAT_ID=
+DIGEST_HOUR_UTC=8
 
 # Worker only: max ms to wait for in-flight webhook batches on SIGTERM before
 # forcing close. Tune to ~2x observed p95 delivery time.

Dockerfile

@@ -35,6 +35,6 @@ WORKDIR /app
 ENV NODE_ENV=production
 COPY package.json package-lock.json ./
 RUN npm ci --omit=dev --ignore-scripts
-COPY --chown=node:node worker.js worker-log.js ./
+COPY --chown=node:node worker.js worker-log.js worker-alert.js worker-digest.js ./
 USER node
 CMD ["node", "worker.js"]

docker-compose.yml

@@ -188,6 +188,7 @@ services:
       ALERT_TELEGRAM_CHAT_ID: ${ALERT_TELEGRAM_CHAT_ID:-}
       BEARER_ADMIN_TELEGRAM_ID: ${BEARER_ADMIN_TELEGRAM_ID:-}
       GRACEFUL_SHUTDOWN_TIMEOUT_MS: ${GRACEFUL_SHUTDOWN_TIMEOUT_MS:-10000}
+      DIGEST_HOUR_UTC: ${DIGEST_HOUR_UTC:-8}
       NODE_ENV: production
     depends_on:
       redis:

docs/deployment.md

@@ -114,9 +114,33 @@ so `docker compose up -d --build app worker` is safe at any time.
 
 ## Operator alerts
 
-Set `ALERT_TELEGRAM_CHAT_ID` (falls back to `BEARER_ADMIN_TELEGRAM_ID`) to get a
-Telegram message when a webhook endpoint is auto-disabled or an enqueue fails.
-Alerts are rate-limited and no-op outside production or without a chat id.
+Set `ALERT_TELEGRAM_CHAT_ID` (a group the bot is a member of; falls back to
+`BEARER_ADMIN_TELEGRAM_ID`) to receive Telegram messages for:
+
+- a webhook endpoint auto-disabled after consecutive failures, or an enqueue
+  that failed (app);
+- Redis unreachable, with rate limiting on the in-process fallback (app);
+- a worker loop that errored: webhook batch, hygiene, activation expiry,
+  restriction, deletion, digest (worker);
+- a migration that failed at boot, which would otherwise be a silent crash
+  loop (app image, before the server starts).
+
+Every alert is deduplicated per key: five minutes for one-off events, thirty
+minutes for conditions that repeat every tick. Alerts no-op outside
+production or without a chat id, and never block the request or loop that
+raised them.
+
+The worker also sends a daily digest at `DIGEST_HOUR_UTC` (default 8): user
+and webhook counts, security events by type for the last 24 hours, worker
+uptime, and how many alerts went out that day. The digest arriving is the
+proof that the worker, the database, and the alert channel are alive; its
+absence is itself the alert. `docker compose exec -T worker node worker.js
+--digest` sends one immediately, bypassing the daily window, which is the
+runbook's "is the channel alive" check.
+
+These alerts are sent from the host. When the host, the tunnel, or the
+worker is down nothing can send them, so they do not replace an external
+uptime monitor.
 
 ## Backup and restore
 

lib/server/rateLimit.ts

@@ -1,4 +1,5 @@
 import redis from "./redis";
+import { alertRedisDegraded } from "./services/operatorAlerts";
 
 // Per-instance, volatile fallback windows used only when Redis is unreachable.
 // They reset on restart and are not shared across instances, so they are
@@ -60,11 +61,14 @@ export async function rateLimit(
     results = await multi.exec();
   } catch {
     // Redis unreachable: fall back to the in-process limiter rather than
-    // failing fully open.
+    // failing fully open, and tell the operator (deduped to one message per
+    // half hour) that limits are now per-instance and non-persistent.
+    alertRedisDegraded();
     return fallbackRateLimit(key, limit, windowMs);
   }
 
   if (!results) {
+    alertRedisDegraded();
     return fallbackRateLimit(key, limit, windowMs);
   }
 

lib/server/services/activation.ts

@@ -26,7 +26,7 @@ import type { ExternalApp, User } from "../types";
 import { toIso } from "../time";
 import { parseScopes } from "../validation";
 import { enqueueWebhookEvent } from "../webhooks";
-import { sendEnqueueFailedAlert } from "./webhookAlerts";
+import { sendOperatorAlert } from "./operatorAlerts";
 
 // Carries a machine-readable code and HTTP status so integrator-facing routes
 // can return {error, code} with the right status instead of a bare string.
@@ -108,10 +108,10 @@ async function fireActivationWebhook(input: {
       context: { ip: "", userAgent: "activation-service", country: "" },
       metadata: { activationId: input.payload.id, eventType: input.eventType },
     });
-    await sendEnqueueFailedAlert(
-      input.appId,
-      err instanceof Error ? err.message : "unknown",
-    ).catch(() => {});
+    await sendOperatorAlert(
+      `webhook_enqueue_failed:${input.appId}`,
+      `Webhook enqueue failed\napp #${input.appId}\n${err instanceof Error ? err.message : "unknown"}`,
+    );
   }
 }
 

lib/server/services/operatorAlerts.ts

@@ -0,0 +1,85 @@
+import { alertTelegramChatId, isProduction } from "../config";
+import redis, { getLastRedisError } from "../redis";
+import { sendTelegramMessage } from "../telegramSend";
+import { log } from "../log";
+
+// Operator alerts: one Telegram message per key per window, best effort,
+// never throws, no-op outside production or without an alert chat. Two
+// windows stack: Redis (`alert:<key>` NX) dedupes across the app and the
+// worker, and a per-process map in front of it bounds the rate when Redis
+// itself is the thing failing, which is exactly when redis_degraded fires
+// on every request.
+export const ALERT_WINDOW_SECONDS = 300;
+export const DEGRADED_WINDOW_SECONDS = 1800;
+const LOCAL_WINDOWS_MAX = 512;
+
+const localWindows = new Map<string, number>();
+
+function acquireLocalWindow(key: string, windowSeconds: number) {
+  const now = Date.now();
+  if ((localWindows.get(key) || 0) > now) return false;
+  if (localWindows.size >= LOCAL_WINDOWS_MAX) {
+    for (const [k, until] of localWindows) if (until <= now) localWindows.delete(k);
+  }
+  localWindows.set(key, now + windowSeconds * 1000);
+  return true;
+}
+
+async function acquireSharedWindow(key: string, windowSeconds: number) {
+  try {
+    return (await redis.set(`alert:${key}`, "1", "EX", windowSeconds, "NX")) === "OK";
+  } catch {
+    // Fail open: a possible duplicate beats a dropped alert, and the local
+    // window above still caps the rate.
+    return true;
+  }
+}
+
+// Counter the daily digest reports, so "alerts sent today: 0" and a dead
+// channel are distinguishable. Best effort.
+async function countSent() {
+  const day = new Date().toISOString().slice(0, 10);
+  try {
+    await redis.incr(`alerts:sent:${day}`);
+    await redis.expire(`alerts:sent:${day}`, 48 * 3600);
+  } catch {
+    // nothing to do
+  }
+}
+
+export async function sendOperatorAlert(
+  key: string,
+  text: string,
+  opts: { windowSeconds?: number } = {},
+): Promise<boolean> {
+  const chatId = alertTelegramChatId();
+  if (!chatId || !isProduction()) return false;
+  const windowSeconds = opts.windowSeconds ?? ALERT_WINDOW_SECONDS;
+  if (!acquireLocalWindow(key, windowSeconds)) return false;
+  if (!(await acquireSharedWindow(key, windowSeconds))) return false;
+  try {
+    await sendTelegramMessage({ chatId, text });
+    await countSent();
+    return true;
+  } catch (err) {
+    log.error("operator_alert_failed", { alert: key, error: err });
+    return false;
+  }
+}
+
+// Fired by the rate limiter on every request while Redis is unreachable; the
+// windows make that one message per half hour. The production gate comes
+// first so nothing below runs in tests that stub the redis module.
+export function alertRedisDegraded() {
+  if (!isProduction()) return;
+  const last = getLastRedisError();
+  void sendOperatorAlert(
+    "redis_degraded",
+    `Redis unreachable\nRate limiting is on the in-process fallback.\n${last ? `${last.at} ${last.message}` : "no error recorded"}`,
+    { windowSeconds: DEGRADED_WINDOW_SECONDS },
+  );
+}
+
+export function _resetForTests() {
+  localWindows.clear();
+}

scripts/migrate.mjs

@@ -33,54 +33,108 @@ function validateMigrationFiles(files) {
   }
 }
 
-const client = new pg.Client({ connectionString: databaseUrl });
-await client.connect();
+async function migrate() {
+  const client = new pg.Client({ connectionString: databaseUrl });
+  await client.connect();
 
-try {
-  await client.query(`
-    create table if not exists schema_migrations (
-      version text primary key,
-      applied_at timestamptz not null default now()
-    )
-  `);
+  try {
+    await client.query(`
+      create table if not exists schema_migrations (
+        version text primary key,
+        applied_at timestamptz not null default now()
+      )
+    `);
 
-  const files = (await readdir(migrationsDir))
-    .filter(file => file.endsWith(".sql"))
-    .sort();
+    const files = (await readdir(migrationsDir))
+      .filter(file => file.endsWith(".sql"))
+      .sort();
 
-  validateMigrationFiles(files);
+    validateMigrationFiles(files);
 
-  const applied = new Set(
-    (await client.query(`select version from schema_migrations`)).rows.map(r => r.version),
-  );
-  const pending = files.filter(file => !applied.has(file.replace(/\.sql$/, "")));
+    const applied = new Set(
+      (await client.query(`select version from schema_migrations`)).rows.map(r => r.version),
+    );
+    const pending = files.filter(file => !applied.has(file.replace(/\.sql$/, "")));
 
-  if (checkOnly) {
-    // Dry run for CI / pre-deploy: report pending migrations, change nothing.
-    if (pending.length === 0) {
-      console.log("migrations: up to date");
+    if (checkOnly) {
+      // Dry run for CI / pre-deploy: report pending migrations, change nothing.
+      if (pending.length === 0) {
+        console.log("migrations: up to date");
+      } else {
+        console.log(`migrations: ${pending.length} pending`);
+        for (const file of pending) console.log(`  ${file}`);
+      }
     } else {
-      console.log(`migrations: ${pending.length} pending`);
-      for (const file of pending) console.log(`  ${file}`);
-    }
-  } else {
-    for (const file of pending) {
-      const version = file.replace(/\.sql$/, "");
-      const sql = await readFile(path.join(migrationsDir, file), "utf8");
-      await client.query("begin");
-      try {
-        await client.query(sql);
-        await client.query(
-          `insert into schema_migrations (version) values ($1)`,
-          [version],
-        );
-        await client.query("commit");
-      } catch (err) {
-        await client.query("rollback");
-        throw err;
+      for (const file of pending) {
+        const version = file.replace(/\.sql$/, "");
+        const sql = await readFile(path.join(migrationsDir, file), "utf8");
+        await client.query("begin");
+        try {
+          await client.query(sql);
+          await client.query(
+            `insert into schema_migrations (version) values ($1)`,
+            [version],
+          );
+          await client.query("commit");
+        } catch (err) {
+          await client.query("rollback");
+          throw err;
+        }
       }
     }
+  } finally {
+    await client.end();
   }
-} finally {
-  await client.end();
+}
+
+// The app image runs this before server.js under restart: unless-stopped, so
+// a failing migration is a crash loop that would otherwise page once per
+// restart. Production only; deduped for 30 minutes through Redis when it
+// answers, sent anyway when it does not. Never throws.
+async function alertMigrationFailed(err) {
+  if (process.env.NODE_ENV !== "production") return;
+  const chatId = process.env.ALERT_TELEGRAM_CHAT_ID || process.env.BEARER_ADMIN_TELEGRAM_ID;
+  const token = process.env.TELEGRAM_BOT_TOKEN;
+  if (!chatId || !token) return;
+
+  let fresh = true;
+  try {
+    const { default: Redis } = await import("ioredis");
+    const redis = new Redis(process.env.REDIS_URL || "redis://localhost:6379", {
+      lazyConnect: true,
+      connectTimeout: 2000,
+      maxRetriesPerRequest: 1,
+    });
+    redis.on("error", () => {});
+    try {
+      await redis.connect();
+      fresh = (await redis.set("alert:migration_failed", "1", "EX", 1800, "NX")) === "OK";
+    } finally {
+      redis.disconnect();
+    }
+  } catch {
+    // No dedupe available; a duplicate beats silence here.
+  }
+  if (!fresh) return;
+
+  try {
+    await fetch(`https://api.telegram.org/bot${token}/sendMessage`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        chat_id: chatId,
+        text: `Migration failed at boot\n${err instanceof Error ? err.message : String(err)}`,
+      }),
+      signal: AbortSignal.timeout(5000),
+    });
+  } catch {
+    // Nothing left to try.
+  }
+}
+
+try {
+  await migrate();
+} catch (err) {
+  await alertMigrationFailed(err);
+  throw err;
 }