From e260364afdd870c6d9b6f3af325f9aa982b2144d Mon Sep 17 00:00:00 2001 From: Matthew Demidoff Date: Sun, 6 Sep 2026 08:03:53 +0200 Subject: [PATCH] feat(ops): generalize operator alerts and add the daily digest Operator alerts existed only for two webhook events, through a service the worker could not share, and the worker's own copy had no rate limit. Nothing alerted on Redis falling back, on a worker loop erroring every tick, or on a migration failing at boot (a crash loop that pages nobody), and there was no signal that the alert channel itself was alive. lib/server/services/operatorAlerts.ts replaces webhookAlerts.ts with one sendOperatorAlert(key, text, { windowSeconds }) behind two stacked windows: Redis NX for cross-process dedupe and a per-process map in front of it, so the redis_degraded alert that fires on every request during a Redis outage still yields one message per half hour. worker-alert.js is its CommonJS twin on a dedicated bounded connection, since the bullmq connection blocks for the whole of an outage. Wired: webhook enqueue failure and auto-disable, the rate limiter's fallback, every worker loop's catch, and scripts/migrate.mjs (best effort, deduped through a short-lived ioredis import, sent anyway when Redis is down). worker-digest.js sends a daily digest at DIGEST_HOUR_UTC from tables the service already writes; the digest arriving is the proof the worker, the database, and the channel are alive, and `node worker.js --digest` sends one on demand. --- .env.example | 9 +- Dockerfile | 2 +- docker-compose.yml | 1 + docs/deployment.md | 30 +++++- lib/server/rateLimit.ts | 6 +- lib/server/services/activation.ts | 10 +- lib/server/services/operatorAlerts.ts | 85 +++++++++++++++ lib/server/services/webhookAlerts.ts | 44 -------- scripts/migrate.mjs | 136 +++++++++++++++++------- tests/integration/worker-digest.test.ts | 43 ++++++++ tests/unit/operator-alerts.test.ts | 84 +++++++++++++++ tests/unit/rate-limit.test.ts | 2 +- tests/unit/worker-alert.test.ts | 82 ++++++++++++++ worker-alert.js | 69 ++++++++++++ worker-digest.js | 80 ++++++++++++++ worker.js | 100 +++++++++++++---- 16 files changed, 665 insertions(+), 118 deletions(-) create mode 100644 lib/server/services/operatorAlerts.ts delete mode 100644 lib/server/services/webhookAlerts.ts create mode 100644 tests/integration/worker-digest.test.ts create mode 100644 tests/unit/operator-alerts.test.ts create mode 100644 tests/unit/worker-alert.test.ts create mode 100644 worker-alert.js create mode 100644 worker-digest.js diff --git a/.env.example b/.env.example index 7082d2a..ad26f0d 100644 --- a/.env.example +++ b/.env.example @@ -59,10 +59,13 @@ EMAIL_FROM_ADDRESS=bottleneck 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. diff --git a/Dockerfile b/Dockerfile index b0898ff..9373171 100644 --- a/Dockerfile +++ b/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"] diff --git a/docker-compose.yml b/docker-compose.yml index a757656..8de69a4 100644 --- a/docker-compose.yml +++ b/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: diff --git a/docs/deployment.md b/docs/deployment.md index 4992c3a..43832e9 100644 --- a/docs/deployment.md +++ b/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 diff --git a/lib/server/rateLimit.ts b/lib/server/rateLimit.ts index a27c535..abce2f7 100644 --- a/lib/server/rateLimit.ts +++ b/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); } diff --git a/lib/server/services/activation.ts b/lib/server/services/activation.ts index 6e2e8e0..c5fa717 100644 --- a/lib/server/services/activation.ts +++ b/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"}`, + ); } } diff --git a/lib/server/services/operatorAlerts.ts b/lib/server/services/operatorAlerts.ts new file mode 100644 index 0000000..0b426ea --- /dev/null +++ b/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:` 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(); + +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 { + 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(); +} diff --git a/lib/server/services/webhookAlerts.ts b/lib/server/services/webhookAlerts.ts deleted file mode 100644 index c8b2fb1..0000000 --- a/lib/server/services/webhookAlerts.ts +++ /dev/null @@ -1,44 +0,0 @@ -import { alertTelegramChatId, isProduction } from "../config"; -import redis from "../redis"; -import { sendTelegramMessage } from "../telegramSend"; -import { log } from "../log"; - -// Operator alerts for webhook trouble. Best-effort and rate-limited: one -// message per key per window so a flapping endpoint cannot spam the operator -// chat. No-ops without an alert chat id or outside production. Never throws. -const ALERT_WINDOW_SECONDS = 300; - -async function withinRateLimit(key: string): Promise { - try { - const set = await redis.set(`alert:${key}`, "1", "EX", ALERT_WINDOW_SECONDS, "NX"); - return set === "OK"; - } catch { - // Fail open: a possible duplicate alert beats a silently dropped one. - return true; - } -} - -async function deliver(key: string, text: string) { - const chatId = alertTelegramChatId(); - if (!chatId || !isProduction()) return; - if (!(await withinRateLimit(key))) return; - try { - await sendTelegramMessage({ chatId, text }); - } catch (err) { - log.error("operator_alert_failed", { key, error: err }); - } -} - -export async function sendEnqueueFailedAlert(appId: number, reason: string) { - await deliver( - `webhook_enqueue_failed:${appId}`, - `Webhook enqueue failed\napp #${appId}\n${reason}`, - ); -} - -export async function sendWebhookDisabledAlert(endpointId: number, failures: number) { - await deliver( - `webhook_disabled:${endpointId}`, - `Webhook endpoint auto-disabled\nendpoint #${endpointId} after ${failures} consecutive failures`, - ); -} diff --git a/scripts/migrate.mjs b/scripts/migrate.mjs index f411fb7..bdb7dfc 100644 --- a/scripts/migrate.mjs +++ b/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; } diff --git a/tests/integration/worker-digest.test.ts b/tests/integration/worker-digest.test.ts new file mode 100644 index 0000000..a9e7aeb --- /dev/null +++ b/tests/integration/worker-digest.test.ts @@ -0,0 +1,43 @@ +import { createRequire } from 'node:module'; +import { afterAll, describe, expect, it } from 'vitest'; +import { Pool } from 'pg'; +import { query } from '@/lib/server/db'; + +const requireCjs = createRequire(import.meta.url); +const { buildDailyDigest, sendDailyDigest } = requireCjs('../../worker-digest.js') as { + buildDailyDigest: (pool: Pool, opts?: Record) => Promise; + sendDailyDigest: (opts: Record) => Promise; +}; + +const describeDb = process.env.DATABASE_URL ? describe : describe.skip; + +describeDb('daily digest', () => { + const pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 2 }); + afterAll(() => pool.end()); + + it('reports the last 24h from tables the service already writes', async () => { + await query( + `insert into security_events (event_type, result, metadata) + values ('digest_test_event', 'ok', '{}'::jsonb)`, + ); + + const text = await buildDailyDigest(pool, { startedAt: Date.now() - 90 * 60 * 1000 }); + + expect(text).toMatch(/^Daily digest auth\.bneck\.com \(\d{4}-\d{2}-\d{2} UTC\)/); + expect(text).toContain('worker up 1h 30m'); + expect(text).toMatch(/users: \d+ total, \d+ new, \d+ pending deletion/); + expect(text).toMatch(/webhooks: \d+ delivered, \d+ failed, \d+ cancelled, \d+ overdue; endpoints \d+ active, \d+ disabled today/); + expect(text).toContain('digest_test_event'); + }); + + it('sends only during the digest hour, keyed by day', async () => { + const sent: string[] = []; + const alerts = { send: async (key: string) => { sent.push(key); return true; } }; + const at = Date.UTC(2026, 8, 5, 8, 10); + const off = Date.UTC(2026, 8, 5, 9, 10); + + expect(await sendDailyDigest({ pool, alerts, hourUtc: 8, now: () => off })).toBe(false); + expect(await sendDailyDigest({ pool, alerts, hourUtc: 8, now: () => at })).toBe(true); + expect(sent).toEqual(['digest:2026-09-05']); + }); +}); diff --git a/tests/unit/operator-alerts.test.ts b/tests/unit/operator-alerts.test.ts new file mode 100644 index 0000000..6ce16e2 --- /dev/null +++ b/tests/unit/operator-alerts.test.ts @@ -0,0 +1,84 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +type AlertInput = { chatId: string | number; text: string }; +const redisSet = vi.fn<(...args: unknown[]) => Promise>(); +const redisIncr = vi.fn<(...args: unknown[]) => Promise>(async () => 1); +const redisExpire = vi.fn<(...args: unknown[]) => Promise>(async () => 1); +const sendTelegramMessage = vi.fn<(input: AlertInput) => Promise<{ ok: boolean }>>(async () => ({ ok: true })); +let production = true; +let chatId = '-100'; + +vi.mock('@/lib/server/redis', () => ({ + default: { + set: (...args: unknown[]) => redisSet(...args), + incr: (...args: unknown[]) => redisIncr(...args), + expire: (...args: unknown[]) => redisExpire(...args), + }, + getLastRedisError: () => ({ at: '2026-09-05T00:00:00.000Z', message: 'ECONNREFUSED' }), +})); +vi.mock('@/lib/server/config', () => ({ + alertTelegramChatId: () => chatId, + isProduction: () => production, +})); +vi.mock('@/lib/server/telegramSend', () => ({ + sendTelegramMessage: (input: AlertInput) => sendTelegramMessage(input), +})); +vi.mock('@/lib/server/log', () => ({ log: { error: vi.fn() } })); + +const { sendOperatorAlert, alertRedisDegraded, _resetForTests } = await import( + '@/lib/server/services/operatorAlerts' +); + +describe('sendOperatorAlert', () => { + beforeEach(() => { + _resetForTests(); + production = true; + chatId = '-100'; + redisSet.mockReset().mockResolvedValue('OK'); + sendTelegramMessage.mockClear(); + }); + + it('sends once per key inside the window', async () => { + expect(await sendOperatorAlert('k', 'first')).toBe(true); + expect(await sendOperatorAlert('k', 'second')).toBe(false); + expect(sendTelegramMessage).toHaveBeenCalledTimes(1); + expect(sendTelegramMessage.mock.calls[0][0]).toEqual({ chatId: '-100', text: 'first' }); + }); + + it('defers to another process that already holds the shared window', async () => { + redisSet.mockResolvedValue(null); + expect(await sendOperatorAlert('k', 'text')).toBe(false); + expect(sendTelegramMessage).not.toHaveBeenCalled(); + }); + + it('fails open when Redis is down, bounded by the local window', async () => { + redisSet.mockRejectedValue(new Error('ECONNREFUSED')); + expect(await sendOperatorAlert('k', 'text')).toBe(true); + expect(await sendOperatorAlert('k', 'text')).toBe(false); + expect(sendTelegramMessage).toHaveBeenCalledTimes(1); + }); + + it('is a no-op outside production or without a chat id', async () => { + production = false; + expect(await sendOperatorAlert('k', 'text')).toBe(false); + production = true; + chatId = ''; + expect(await sendOperatorAlert('k', 'text')).toBe(false); + expect(sendTelegramMessage).not.toHaveBeenCalled(); + }); + + it('reports a Telegram failure as not sent without throwing', async () => { + sendTelegramMessage.mockRejectedValueOnce(new Error('telegram sendMessage failed: 500')); + expect(await sendOperatorAlert('k', 'text')).toBe(false); + }); + + it('alertRedisDegraded carries the last Redis error and uses the long window', async () => { + alertRedisDegraded(); + await new Promise(r => setTimeout(r, 0)); + expect(sendTelegramMessage).toHaveBeenCalledTimes(1); + const { text } = sendTelegramMessage.mock.calls[0][0]; + expect(text).toContain('Redis unreachable'); + expect(text).toContain('ECONNREFUSED'); + expect(redisSet).toHaveBeenCalledWith('alert:redis_degraded', '1', 'EX', 1800, 'NX'); + }); +}); diff --git a/tests/unit/rate-limit.test.ts b/tests/unit/rate-limit.test.ts index 3b3f723..c19e226 100644 --- a/tests/unit/rate-limit.test.ts +++ b/tests/unit/rate-limit.test.ts @@ -60,7 +60,7 @@ const fakeRedis = { }, }; -vi.mock('@/lib/server/redis', () => ({ default: fakeRedis })); +vi.mock('@/lib/server/redis', () => ({ default: fakeRedis, getLastRedisError: () => null })); // Import after mock so the module gets our fake. const { rateLimit } = await import('@/lib/server/rateLimit'); diff --git a/tests/unit/worker-alert.test.ts b/tests/unit/worker-alert.test.ts new file mode 100644 index 0000000..5666cce --- /dev/null +++ b/tests/unit/worker-alert.test.ts @@ -0,0 +1,82 @@ +import { createRequire } from 'node:module'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const requireCjs = createRequire(import.meta.url); +const { createOperatorAlerter, noopAlerter } = requireCjs('../../worker-alert.js') as { + createOperatorAlerter: (opts: Record) => { + send: (key: string, text: string, opts?: { windowSeconds?: number }) => Promise; + }; + noopAlerter: { send: () => Promise }; +}; + +const redis = { + set: vi.fn<(...args: unknown[]) => Promise>(async () => 'OK'), + incr: vi.fn<(...args: unknown[]) => Promise>(async () => 1), + expire: vi.fn<(...args: unknown[]) => Promise>(async () => 1), +}; +const fetchImpl = vi.fn<(url: string, init: RequestInit) => Promise<{ ok: boolean; status?: number }>>( + async () => ({ ok: true }), +); +const log = { error: vi.fn() }; + +function alerter(overrides: Record = {}) { + return createOperatorAlerter({ + redis, + chatId: '-100', + token: 'tok', + enabled: true, + log, + fetchImpl, + ...overrides, + }); +} + +describe('worker operator alerter', () => { + beforeEach(() => { + redis.set.mockReset().mockResolvedValue('OK'); + fetchImpl.mockReset().mockResolvedValue({ ok: true }); + log.error.mockClear(); + }); + + it('sends once per key and posts to the Telegram API with a timeout', async () => { + const a = alerter(); + expect(await a.send('k', 'hello')).toBe(true); + expect(await a.send('k', 'again')).toBe(false); + expect(fetchImpl).toHaveBeenCalledTimes(1); + const [url, init] = fetchImpl.mock.calls[0]; + expect(url).toBe('https://api.telegram.org/bottok/sendMessage'); + expect(JSON.parse(String(init.body))).toEqual({ chat_id: '-100', text: 'hello' }); + expect(init.signal).toBeInstanceOf(AbortSignal); + expect(redis.set).toHaveBeenCalledWith('alert:k', '1', 'EX', 300, 'NX'); + }); + + it('honours a caller-supplied window', async () => { + await alerter().send('k', 'x', { windowSeconds: 1800 }); + expect(redis.set).toHaveBeenCalledWith('alert:k', '1', 'EX', 1800, 'NX'); + }); + + it('defers when the shared window is held elsewhere', async () => { + redis.set.mockResolvedValue(null); + expect(await alerter().send('k', 'x')).toBe(false); + expect(fetchImpl).not.toHaveBeenCalled(); + }); + + it('fails open on Redis errors but stays bounded by the local window', async () => { + redis.set.mockRejectedValue(new Error('down')); + const a = alerter(); + expect(await a.send('k', 'x')).toBe(true); + expect(await a.send('k', 'x')).toBe(false); + expect(fetchImpl).toHaveBeenCalledTimes(1); + }); + + it('is disabled outside production and logs a failed post', async () => { + expect(await alerter({ enabled: false }).send('k', 'x')).toBe(false); + fetchImpl.mockResolvedValue({ ok: false, status: 500 }); + expect(await alerter().send('k2', 'x')).toBe(false); + expect(log.error).toHaveBeenCalledWith('operator_alert_failed', expect.objectContaining({ alert: 'k2' })); + }); + + it('noopAlerter never sends', async () => { + expect(await noopAlerter.send()).toBe(false); + }); +}); diff --git a/worker-alert.js b/worker-alert.js new file mode 100644 index 0000000..9e4f91b --- /dev/null +++ b/worker-alert.js @@ -0,0 +1,69 @@ +// Operator alerts for the worker: plain-CommonJS twin of +// lib/server/services/operatorAlerts.ts. Same `alert:` NX window in +// Redis so the app and the worker dedupe against each other, and the same +// per-process window in front of it so a Redis outage cannot turn the 1s +// webhook loop's failures into a message per second. Best effort, never +// throws. Keep the two in sync. + +const ALERT_WINDOW_SECONDS = 300; +const LOCAL_WINDOWS_MAX = 512; +const SEND_TIMEOUT_MS = 5000; + +function createOperatorAlerter({ redis, chatId, token, enabled, log, fetchImpl = fetch, now = Date.now }) { + const localWindows = new Map(); + + function acquireLocalWindow(key, windowSeconds) { + const at = now(); + if ((localWindows.get(key) || 0) > at) return false; + if (localWindows.size >= LOCAL_WINDOWS_MAX) { + for (const [k, until] of localWindows) if (until <= at) localWindows.delete(k); + } + localWindows.set(key, at + windowSeconds * 1000); + return true; + } + + async function acquireSharedWindow(key, windowSeconds) { + try { + return (await redis.set(`alert:${key}`, "1", "EX", windowSeconds, "NX")) === "OK"; + } catch { + return true; + } + } + + async function countSent() { + const day = new Date(now()).toISOString().slice(0, 10); + try { + await redis.incr(`alerts:sent:${day}`); + await redis.expire(`alerts:sent:${day}`, 48 * 3600); + } catch { + // best effort + } + } + + async function send(key, text, opts = {}) { + if (!enabled || !chatId || !token) return false; + const windowSeconds = opts.windowSeconds ?? ALERT_WINDOW_SECONDS; + if (!acquireLocalWindow(key, windowSeconds)) return false; + if (!(await acquireSharedWindow(key, windowSeconds))) return false; + try { + const res = await fetchImpl(`https://api.telegram.org/bot${token}/sendMessage`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ chat_id: chatId, text }), + signal: AbortSignal.timeout(SEND_TIMEOUT_MS), + }); + if (!res.ok) throw new Error(`telegram sendMessage failed: ${res.status}`); + await countSent(); + return true; + } catch (err) { + log.error("operator_alert_failed", { alert: key, error: err }); + return false; + } + } + + return { send }; +} + +const noopAlerter = { send: async () => false }; + +module.exports = { createOperatorAlerter, noopAlerter, ALERT_WINDOW_SECONDS }; diff --git a/worker-digest.js b/worker-digest.js new file mode 100644 index 0000000..c58df2a --- /dev/null +++ b/worker-digest.js @@ -0,0 +1,80 @@ +// Daily operator digest. Its job is as much liveness as content: the digest +// arriving every day at DIGEST_HOUR_UTC is the signal that the worker, the +// database, and the alert channel are all alive, so its absence is itself +// the alert. Counts come from tables the service already writes; nothing is +// instrumented for this. + +const DIGEST_WINDOW_SECONDS = 36 * 3600; + +function formatUptime(ms) { + const minutes = Math.floor(ms / 60000); + return `${Math.floor(minutes / 60)}h ${minutes % 60}m`; +} + +async function buildDailyDigest(pool, { redis, startedAt, now = Date.now } = {}) { + const [events, deliveries, endpoints, users] = await Promise.all([ + pool.query( + `select event_type, count(*)::int as count + from security_events + where created_at > now() - interval '24 hours' + group by event_type + order by count desc, event_type + limit 12`, + ), + pool.query( + `select count(*) filter (where status = 'delivered' and delivered_at > now() - interval '24 hours')::int as delivered, + count(*) filter (where status = 'failed' and created_at > now() - interval '24 hours')::int as failed, + count(*) filter (where status = 'cancelled' and created_at > now() - interval '24 hours')::int as cancelled, + count(*) filter (where status = 'pending' and next_attempt_at < now() - interval '10 minutes')::int as overdue + from webhook_deliveries`, + ), + pool.query( + `select count(*) filter (where status = 'active')::int as active, + count(*) filter (where status = 'disabled' and disabled_at > now() - interval '24 hours')::int as disabled_today + from webhook_endpoints`, + ), + pool.query( + `select count(*)::int as total, + count(*) filter (where created_at > now() - interval '24 hours')::int as new_today, + count(*) filter (where deletion_requested_at is not null)::int as pending_deletion + from users`, + ), + ]); + + const day = new Date(now()).toISOString().slice(0, 10); + let alertsSent = "n/a"; + if (redis) { + try { + alertsSent = (await redis.get(`alerts:sent:${day}`)) || "0"; + } catch { + // Redis down is itself reported by the worker; the digest still goes out. + } + } + + const d = deliveries.rows[0]; + const e = endpoints.rows[0]; + const u = users.rows[0]; + const lines = [ + `Daily digest auth.bneck.com (${day} UTC)`, + `worker up ${startedAt ? formatUptime(now() - startedAt) : "n/a"}, alerts sent today: ${alertsSent}`, + `users: ${u.total} total, ${u.new_today} new, ${u.pending_deletion} pending deletion`, + `webhooks: ${d.delivered} delivered, ${d.failed} failed, ${d.cancelled} cancelled, ${d.overdue} overdue; endpoints ${e.active} active, ${e.disabled_today} disabled today`, + "events (24h):", + ...(events.rows.length + ? events.rows.map(row => ` ${row.event_type} ${row.count}`) + : [" none"]), + ]; + return lines.join("\n"); +} + +// Runs on an hourly tick; the hour gate plus the 36h NX window on +// `digest:` yields exactly one send per UTC day regardless of restarts. +async function sendDailyDigest({ pool, alerts, redis, hourUtc, startedAt, now = Date.now }) { + const current = new Date(now()); + if (current.getUTCHours() !== hourUtc) return false; + const day = current.toISOString().slice(0, 10); + const text = await buildDailyDigest(pool, { redis, startedAt, now }); + return alerts.send(`digest:${day}`, text, { windowSeconds: DIGEST_WINDOW_SECONDS }); +} + +module.exports = { buildDailyDigest, sendDailyDigest, DIGEST_WINDOW_SECONDS }; diff --git a/worker.js b/worker.js index 0adde12..f23cc57 100644 --- a/worker.js +++ b/worker.js @@ -6,22 +6,22 @@ const { lookup } = require("dns/promises"); const { isIP } = require("net"); const logger = require("./worker-log.js"); -// Operator alert (plain JS; the worker cannot import the TS webhookAlerts -// service). Auto-disable transitions fire exactly once per endpoint, so no -// rate limit is needed here. Best-effort, never throws. -async function sendOperatorAlert(text) { - const chatId = process.env.ALERT_TELEGRAM_CHAT_ID || process.env.BEARER_ADMIN_TELEGRAM_ID; - const token = process.env.TELEGRAM_BOT_TOKEN; - if (!chatId || !token || process.env.NODE_ENV !== "production") 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 }), - }); - } catch (err) { - logger.error("operator_alert_failed", { error: err }); - } +const { createOperatorAlerter, noopAlerter } = require("./worker-alert.js"); +const { buildDailyDigest, sendDailyDigest } = require("./worker-digest.js"); + +// Operator alerts are wired up in startWorker(); until then (and in tests +// that require this module) they are a no-op. +let alerts = noopAlerter; +let opsRedis = null; +const startedAt = Date.now(); +const DIGEST_HOUR_UTC = Number(process.env.DIGEST_HOUR_UTC || 8); +// Loop failures repeat every tick while the cause persists; one message per +// half hour per loop is enough to act on. +const LOOP_ALERT_WINDOW_SECONDS = 1800; + +function alertLoopError(event, err) { + const message = err instanceof Error ? err.message : String(err); + return alerts.send(event, `${event}\n${message}`, { windowSeconds: LOOP_ALERT_WINDOW_SECONDS }); } // Graceful-shutdown state. On SIGTERM/SIGINT we stop scheduling new work, @@ -310,7 +310,8 @@ async function disableEndpointIfFailing(row) { endpointId: row.webhook_endpoint_id, consecutiveFailures: endpoint.consecutive_failures, }); - await sendOperatorAlert( + await alerts.send( + `webhook_disabled:${row.webhook_endpoint_id}`, `Webhook endpoint auto-disabled\nendpoint #${row.webhook_endpoint_id} after ${endpoint.consecutive_failures} consecutive failures`, ); } @@ -367,6 +368,7 @@ async function processWebhookBatch() { } } catch (err) { logger.error("webhook_batch_error", { error: err }); + await alertLoopError("webhook_batch_error", err); } } @@ -485,6 +487,7 @@ async function sweepHygiene() { ); } catch (err) { logger.error("hygiene_sweep_error", { error: err }); + await alertLoopError("hygiene_sweep_error", err); } } @@ -599,6 +602,7 @@ async function sweepRestrictedInactive() { } } catch (err) { logger.error("restriction_sweep_error", { error: err }); + await alertLoopError("restriction_sweep_error", err); } } @@ -621,6 +625,7 @@ async function sweepPendingDeletions() { ); } catch (err) { logger.error("deletion_sweep_error", { error: err }); + await alertLoopError("deletion_sweep_error", err); return; } @@ -704,7 +709,10 @@ function runBatch(fn, errorEvent) { if (isShuttingDown) return; inFlightBatches += 1; fn() - .catch(err => logger.error(errorEvent, { error: err })) + .catch(err => { + logger.error(errorEvent, { error: err }); + return alertLoopError(errorEvent, err); + }) .finally(() => { inFlightBatches -= 1; }); @@ -734,6 +742,7 @@ async function shutdownGracefully(signal) { } try { if (bullConnection) await bullConnection.quit(); + if (opsRedis) await opsRedis.quit(); } catch (err) { logger.error("redis_quit_failed", { error: err }); } @@ -751,6 +760,19 @@ function startWorker() { // listener node treats the first one as fatal. bullConnection.on("error", err => logger.error("redis_error", { error: err })); + // Alerts and the digest use their own bounded connection: bullConnection + // runs with maxRetriesPerRequest: null, so a command on it would block for + // the whole of a Redis outage, which is the moment alerts matter most. + opsRedis = new Redis(redisUrl, { maxRetriesPerRequest: 1, enableOfflineQueue: false }); + opsRedis.on("error", err => logger.error("redis_error", { error: err, connection: "ops" })); + alerts = createOperatorAlerter({ + redis: opsRedis, + chatId: process.env.ALERT_TELEGRAM_CHAT_ID || process.env.BEARER_ADMIN_TELEGRAM_ID, + token: botToken, + enabled: process.env.NODE_ENV === "production", + log: logger, + }); + bullWorker = new Worker("telegram-notifications", async (job) => { if (job.name === "send") { logger.debug("telegram_job_send", { jobId: job.id }); @@ -801,6 +823,14 @@ function startWorker() { runBatch(sweepPendingDeletions, "initial_deletion_sweep_error"); logger.info("deletion_sweep_started"); + // Hourly tick; the hour gate and the 36h NX window inside make it one + // send per UTC day. Running once at start covers a restart during the + // digest hour. + const digest = () => sendDailyDigest({ pool, alerts, redis: opsRedis, hourUtc: DIGEST_HOUR_UTC, startedAt }); + intervalIds.push(setInterval(() => runBatch(digest, "digest_loop_error"), 60 * 60 * 1000)); + runBatch(digest, "initial_digest_error"); + logger.info("daily_digest_started", { hourUtc: DIGEST_HOUR_UTC }); + process.on("SIGTERM", () => shutdownGracefully("SIGTERM")); process.on("SIGINT", () => shutdownGracefully("SIGINT")); @@ -819,8 +849,40 @@ function startWorker() { }); } +// `node worker.js --digest` builds and sends the digest once, bypassing the +// hour gate and the daily window, then exits: the runbook's "is the alert +// channel alive" check. +async function runDigestOnce() { + const redisUrl = process.env.REDIS_URL || "redis://localhost:6379"; + const redis = new Redis(redisUrl, { maxRetriesPerRequest: 1, enableOfflineQueue: false }); + redis.on("error", err => logger.error("redis_error", { error: err, connection: "ops" })); + const sender = createOperatorAlerter({ + redis, + chatId: process.env.ALERT_TELEGRAM_CHAT_ID || process.env.BEARER_ADMIN_TELEGRAM_ID, + token: process.env.TELEGRAM_BOT_TOKEN, + enabled: true, + log: logger, + }); + try { + const text = await buildDailyDigest(pool, { redis, startedAt }); + const sent = await sender.send(`digest:manual:${Date.now()}`, text, { windowSeconds: 1 }); + logger.info("digest_sent_once", { sent }); + process.stdout.write(text + "\n"); + } finally { + await pool.end(); + redis.disconnect(); + } +} + if (require.main === module) { - startWorker(); + if (process.argv.includes("--digest")) { + runDigestOnce().catch(err => { + logger.error("digest_once_failed", { error: err }); + process.exit(1); + }); + } else { + startWorker(); + } } module.exports = {