From 468f4dc0a6cc09eca58049398224d327578ceb00 Mon Sep 17 00:00:00 2001 From: Pontus Abrahamsson Date: Sat, 30 May 2026 12:54:56 +0200 Subject: [PATCH 1/3] feat(core): add self-hosted alerting with Slack and webhook destinations Enable dashboard-managed alert rules on BullMQ QueueEvents by default, with Redis-backed config, Block Kit Slack delivery, and a destinations/rules/activity UI. Co-authored-by: Cursor --- CHANGELOG.md | 7 + README.md | 5 + examples/with-elysia/src/index.ts | 3 + packages/core/README.md | 64 + packages/core/src/api/handlers.ts | 205 ++++ .../core/src/core/alert-destinations.test.ts | 112 ++ packages/core/src/core/alert-destinations.ts | 280 +++++ packages/core/src/core/alert-manager.ts | 482 ++++++++ packages/core/src/core/alert-store.ts | 195 ++++ packages/core/src/core/queue-manager.ts | 7 + packages/core/src/core/redis-alert-store.ts | 190 +++ packages/core/src/core/types.ts | 154 +++ packages/core/src/core/workbench.ts | 30 + packages/core/src/index.ts | 20 +- .../core/src/ui/components/app-sidebar.tsx | 3 + .../ui/components/layout/command-palette.tsx | 8 + packages/core/src/ui/lib/api.ts | 92 ++ packages/core/src/ui/lib/hooks.ts | 121 ++ packages/core/src/ui/pages/alerts.tsx | 1029 +++++++++++++++++ packages/core/src/ui/router.tsx | 22 + packages/core/tsconfig.json | 2 +- 21 files changed, 3029 insertions(+), 2 deletions(-) create mode 100644 packages/core/src/core/alert-destinations.test.ts create mode 100644 packages/core/src/core/alert-destinations.ts create mode 100644 packages/core/src/core/alert-manager.ts create mode 100644 packages/core/src/core/alert-store.ts create mode 100644 packages/core/src/core/redis-alert-store.ts create mode 100644 packages/core/src/ui/pages/alerts.tsx diff --git a/CHANGELOG.md b/CHANGELOG.md index f6ab50d..83a70ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,13 @@ All notable changes to Workbench will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- **`@getworkbench/core`** Self-hosted alerting: BullMQ `QueueEvents` listeners, Slack/webhook presets, dashboard **Alerts** page, and Redis-backed persistence for contact points and rules (keys under `{prefix}:workbench:alerts:*`). Alerts are enabled by default; configure contact points and rules in the dashboard to start receiving notifications. +- **`@getworkbench/core`** Slack incoming webhook validation, Block Kit payload fixes (HTTPS dashboard buttons, HTTP dev links), and in-dialog Slack setup instructions. + ## [0.7.1] - 2026-05-28 ### Changed diff --git a/README.md b/README.md index 8dac28c..0b7fdb3 100644 --- a/README.md +++ b/README.md @@ -395,6 +395,11 @@ Visit `http://localhost:PORT/jobs`. | `basePath` | `string` | Override base path detection. Required for `@getworkbench/elysia`, `@getworkbench/koa`, `@getworkbench/next`, `@getworkbench/tanstack-start`, `@getworkbench/astro`, `@getworkbench/nuxt`, and `@getworkbench/h3`. | | `readonly` | `boolean` | Disable actions (retry, remove, promote). | | `tags` | `string[]` | Fields from `job.data` to extract as filterable tags. | +| `alerts` | `AlertsOptions` | Self-hosted Slack/webhook alerting via BullMQ QueueEvents. | + +### Alerts + +Alerts are on by default. Configure Slack or webhook **contact points** and **rules** in the dashboard **Alerts** page — notifications are only sent after you set that up. The model follows Grafana-style contact points + rules (unlike Bull Board, which has no built-in alerting). See [`@getworkbench/core` README](./packages/core/README.md) for Slack setup steps. ## Packages diff --git a/examples/with-elysia/src/index.ts b/examples/with-elysia/src/index.ts index 14b77ee..15fb14a 100644 --- a/examples/with-elysia/src/index.ts +++ b/examples/with-elysia/src/index.ts @@ -26,6 +26,9 @@ new Elysia() title: "Elysia · Workbench", basePath: "/jobs", auth, + alerts: { + dashboardUrl: `http://localhost:${port}/jobs`, + }, }), ) .listen(port, ({ port }) => { diff --git a/packages/core/README.md b/packages/core/README.md index f90ec65..e99d56a 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -118,6 +118,70 @@ Shared by every adapter and by `createFetchHandler` / `WorkbenchCore`: | `basePath` | `string` | Override base path detection. Required for some adapters (Elysia, Koa, Next.js, TanStack Start, Astro, Nuxt, h3, Adonis). | | `readonly` | `boolean` | Disable actions (retry, remove, promote). | | `tags` | `string[]` | Fields from `job.data` to extract as filterable tags. | +| `alerts` | `AlertsOptions` | Self-hosted alerting via BullMQ `QueueEvents`, Slack/webhook contact points, and dashboard-managed rules. | + +## Alerts + +Alerts are enabled by default. Configure contact points and rules from the dashboard **Alerts** page — nothing is sent until you add destinations and matching rules. + +Optional seed data and settings in code: + +```ts +workbench({ + queues: [emailQueue], + basePath: "/jobs", + alerts: { + dashboardUrl: "https://jobs.example.com/jobs", + contactPoints: [ + { + id: "slack-prod", + name: "Slack #ops", + preset: "slack", + url: process.env.SLACK_WEBHOOK_URL!, + enabled: true, + createdAt: Date.now(), + updatedAt: Date.now(), + }, + ], + rules: [ + { + id: "failed-jobs", + name: "Job failed", + enabled: true, + trigger: "job_failed", + severity: "warning", + contactPointIds: ["slack-prod"], + createdAt: Date.now(), + updatedAt: Date.now(), + }, + ], + }, +}); +``` + +Supported triggers: `job_failed`, `job_stalled`, `retries_exhausted`, `failed_backlog`, `no_workers_with_backlog`. + +Dashboard-managed config is **persisted in Redis** at `{prefix}:workbench:alerts:*` using the same connection as your queues. Code-defined `contactPoints` / `rules` seed Redis once when the store is empty. + +Set `alerts.persistence: "memory"` to keep config in-process only (resets on restart). Set `alerts.enabled: false` to disable alerting entirely. + +### How this compares + +| Tool | Alert setup | +| --- | --- | +| **Bull Board** | No built-in alerting — you wire Slack/webhooks in worker code | +| **Sidekiq** | Code hooks (`death_handlers`) in your initializer | +| **Grafana** | UI contact points + alert rules + optional routing tree | +| **Workbench** | Grafana-style **contact points** and **rules** in the dashboard; event-driven via BullMQ `QueueEvents` | + +### Slack setup + +1. Create a Slack app at [api.slack.com/apps](https://api.slack.com/apps) +2. Enable **Incoming Webhooks** → **Add New Webhook to Workspace** → pick a channel +3. Copy the webhook URL (`https://hooks.slack.com/services/...`) +4. In Workbench **Alerts** → **Add contact point** → preset **Slack** → paste URL → **Test** + +Slack preset sends Block Kit messages with queue, job, and severity details. Set `alerts.dashboardUrl` so notifications include an **Open in Workbench** link (HTTPS links render as buttons; HTTP dev URLs use a text link). Generic webhook preset sends a stable JSON payload for automation. ## Requirements diff --git a/packages/core/src/api/handlers.ts b/packages/core/src/api/handlers.ts index 346d5c7..f28e2c8 100644 --- a/packages/core/src/api/handlers.ts +++ b/packages/core/src/api/handlers.ts @@ -1,4 +1,8 @@ +import { validateContactPointUrl } from "../core/alert-destinations"; +import { toPublicContactPoint } from "../core/alert-store"; import type { + AlertContactPoint, + AlertRule, CreateFlowRequest, JobStatus, SortOptions, @@ -67,6 +71,205 @@ const readonlyError = { body: { error: "Dashboard is in readonly mode" }, }; +function alertsRoutes(core: WorkbenchCore): RouteDef[] { + const am = core.alertManager; + if (!am) return []; + + const store = am.getStore(); + const isReadonly = () => !!core.options.readonly; + + return [ + { + method: "get", + path: "/alerts/status", + handler: async () => ({ + status: 200, + body: await am.getStatus(), + }), + }, + { + method: "get", + path: "/alerts/contact-points", + handler: async () => { + const points = await store.getContactPoints(); + return { + status: 200, + body: points.map(toPublicContactPoint), + }; + }, + }, + { + method: "post", + path: "/alerts/contact-points", + handler: async ({ body }) => { + if (isReadonly()) return readonlyError; + const input = body as Partial | undefined; + if (!input?.name || !input.url || !input.preset) { + return { + status: 400, + body: { error: "name, url, and preset are required" }, + }; + } + const urlError = validateContactPointUrl(input.preset, input.url); + if (urlError) { + return { status: 400, body: { error: urlError } }; + } + const created = await store.createContactPoint({ + name: input.name, + preset: input.preset, + url: input.url, + enabled: input.enabled ?? true, + displayName: input.displayName, + iconUrl: input.iconUrl, + headers: input.headers, + }); + return { status: 201, body: toPublicContactPoint(created) }; + }, + }, + { + method: "put", + path: "/alerts/contact-points/:id", + handler: async ({ params, body }) => { + if (isReadonly()) return readonlyError; + const input = body as Partial | undefined; + if (input?.url && input.preset) { + const urlError = validateContactPointUrl(input.preset, input.url); + if (urlError) { + return { status: 400, body: { error: urlError } }; + } + } else if (input?.url) { + const existing = await store.getContactPoint(params.id!); + if (existing) { + const urlError = validateContactPointUrl( + existing.preset, + input.url, + ); + if (urlError) { + return { status: 400, body: { error: urlError } }; + } + } + } + const updated = await store.updateContactPoint(params.id!, { + name: input?.name, + preset: input?.preset, + url: input?.url, + enabled: input?.enabled, + displayName: input?.displayName, + iconUrl: input?.iconUrl, + headers: input?.headers, + }); + if (!updated) { + return { status: 404, body: { error: "Contact point not found" } }; + } + return { status: 200, body: toPublicContactPoint(updated) }; + }, + }, + { + method: "delete", + path: "/alerts/contact-points/:id", + handler: async ({ params }) => { + if (isReadonly()) return readonlyError; + const ok = await store.deleteContactPoint(params.id!); + if (!ok) { + return { status: 404, body: { error: "Contact point not found" } }; + } + return { status: 200, body: { success: true } }; + }, + }, + { + method: "post", + path: "/alerts/contact-points/:id/test", + handler: async ({ params }) => { + if (isReadonly()) return readonlyError; + const record = await am.sendTest(params.id!); + return { status: 200, body: record }; + }, + }, + { + method: "get", + path: "/alerts/rules", + handler: async () => ({ + status: 200, + body: await store.getRules(), + }), + }, + { + method: "post", + path: "/alerts/rules", + handler: async ({ body }) => { + if (isReadonly()) return readonlyError; + const input = body as Partial | undefined; + if (!input?.name || !input.trigger || !input.contactPointIds?.length) { + return { + status: 400, + body: { + error: "name, trigger, and contactPointIds are required", + }, + }; + } + const created = await store.createRule({ + name: input.name, + enabled: input.enabled ?? true, + trigger: input.trigger, + severity: input.severity ?? "warning", + queues: input.queues, + jobNames: input.jobNames, + threshold: input.threshold, + contactPointIds: input.contactPointIds, + cooldownMs: input.cooldownMs, + }); + return { status: 201, body: created }; + }, + }, + { + method: "put", + path: "/alerts/rules/:id", + handler: async ({ params, body }) => { + if (isReadonly()) return readonlyError; + const input = body as Partial | undefined; + const updated = await store.updateRule(params.id!, { + name: input?.name, + enabled: input?.enabled, + trigger: input?.trigger, + severity: input?.severity, + queues: input?.queues, + jobNames: input?.jobNames, + threshold: input?.threshold, + contactPointIds: input?.contactPointIds, + cooldownMs: input?.cooldownMs, + }); + if (!updated) { + return { status: 404, body: { error: "Rule not found" } }; + } + return { status: 200, body: updated }; + }, + }, + { + method: "delete", + path: "/alerts/rules/:id", + handler: async ({ params }) => { + if (isReadonly()) return readonlyError; + const ok = await store.deleteRule(params.id!); + if (!ok) { + return { status: 404, body: { error: "Rule not found" } }; + } + return { status: 200, body: { success: true } }; + }, + }, + { + method: "post", + path: "/alerts/rules/:id/preview", + handler: async ({ params }) => { + const rule = await store.getRule(params.id!); + if (!rule) { + return { status: 404, body: { error: "Rule not found" } }; + } + return { status: 200, body: am.previewRule(rule) }; + }, + }, + ]; +} + /** * Build the framework-agnostic route table for the Workbench API. * @@ -537,5 +740,7 @@ export function buildRouteTable(core: WorkbenchCore): RouteDef[] { } }, }, + + ...alertsRoutes(core), ]; } diff --git a/packages/core/src/core/alert-destinations.test.ts b/packages/core/src/core/alert-destinations.test.ts new file mode 100644 index 0000000..3d1244f --- /dev/null +++ b/packages/core/src/core/alert-destinations.test.ts @@ -0,0 +1,112 @@ +import { describe, expect, test } from "bun:test"; +import { + formatSlackPayload, + isSlackIncomingWebhookUrl, + validateContactPointUrl, +} from "./alert-destinations"; +import type { AlertContactPoint, AlertEvent } from "./types"; + +const slackCp: AlertContactPoint = { + id: "cp-1", + name: "Slack #ops", + preset: "slack", + url: "https://hooks.slack.com/services/T000/B000/XXXXXXXX", + enabled: true, + createdAt: 0, + updatedAt: 0, +}; + +const sampleEvent: AlertEvent = { + id: "evt-1", + ruleId: "rule-1", + ruleName: "Job failed", + trigger: "job_failed", + severity: "warning", + status: "firing", + fingerprint: "fp-1", + queue: "email", + jobId: "42", + jobName: "sendReceipt", + message: "Job sendReceipt failed in email", + failedReason: "SMTP timeout", + firedAt: Date.now(), +}; + +describe("isSlackIncomingWebhookUrl", () => { + test("accepts standard Slack webhook URLs", () => { + expect( + isSlackIncomingWebhookUrl( + "https://hooks.slack.com/services/T000/B000/XXXXXXXX", + ), + ).toBe(true); + }); + + test("accepts GovSlack webhook URLs", () => { + expect( + isSlackIncomingWebhookUrl( + "https://hooks.slack-gov.com/services/T000/B000/XXXXXXXX", + ), + ).toBe(true); + }); + + test("rejects non-Slack URLs", () => { + expect(isSlackIncomingWebhookUrl("https://example.com/webhook")).toBe( + false, + ); + }); +}); + +describe("validateContactPointUrl", () => { + test("requires Slack webhook host for slack preset", () => { + expect( + validateContactPointUrl("slack", "https://example.com/hook"), + ).toContain("incoming webhook"); + }); + + test("allows generic https URLs for webhook preset", () => { + expect( + validateContactPointUrl("webhook", "https://example.com/hook"), + ).toBeUndefined(); + }); +}); + +describe("formatSlackPayload", () => { + test("includes required text fallback and Block Kit blocks", () => { + const { url, headers, body } = formatSlackPayload(slackCp, sampleEvent); + + expect(url).toBe(slackCp.url); + expect(headers["Content-Type"]).toBe("application/json"); + + const payload = body as { + text: string; + blocks: Array<{ type: string }>; + }; + expect(payload.text).toContain("Job failed"); + expect(payload.blocks.some((b) => b.type === "header")).toBe(true); + expect(payload.blocks.some((b) => b.type === "section")).toBe(true); + }); + + test("uses a button for https dashboard links", () => { + const { body } = formatSlackPayload( + slackCp, + sampleEvent, + "https://jobs.example.com/jobs", + ); + const payload = body as { blocks: Array<{ type: string }> }; + expect(payload.blocks.some((b) => b.type === "actions")).toBe(true); + }); + + test("uses a text link for http dev dashboard URLs", () => { + const { body } = formatSlackPayload( + slackCp, + sampleEvent, + "http://localhost:3010/jobs", + ); + const payload = body as { + blocks: Array<{ type: string; elements?: unknown[] }>; + }; + expect(payload.blocks.some((b) => b.type === "actions")).toBe(false); + const context = payload.blocks.find((b) => b.type === "context"); + expect(context?.elements).toBeDefined(); + }); +}); diff --git a/packages/core/src/core/alert-destinations.ts b/packages/core/src/core/alert-destinations.ts new file mode 100644 index 0000000..11d64a1 --- /dev/null +++ b/packages/core/src/core/alert-destinations.ts @@ -0,0 +1,280 @@ +import type { AlertContactPoint, AlertEvent } from "./types"; + +/** Default Workbench notification identity */ +export const DEFAULT_ALERT_SENDER_NAME = "Workbench Alerts"; + +/** Inline SVG data URI for Workbench app icon (used as Slack avatar when supported) */ +export const DEFAULT_WORKBENCH_ICON_URL = + "https://raw.githubusercontent.com/pontusab/workbench/main/packages/core/src/ui/public/app-icon.svg"; + +/** Slack incoming webhook URLs (modern app or legacy custom integration). */ +export function isSlackIncomingWebhookUrl(url: string): boolean { + try { + const parsed = new URL(url); + return ( + parsed.protocol === "https:" && + (parsed.hostname === "hooks.slack.com" || + parsed.hostname === "hooks.slack-gov.com") && + parsed.pathname.startsWith("/services/") + ); + } catch { + return false; + } +} + +export function validateContactPointUrl( + preset: AlertContactPoint["preset"], + url: string, +): string | undefined { + try { + const parsed = new URL(url); + if (parsed.protocol !== "https:" && parsed.protocol !== "http:") { + return "URL must use http or https"; + } + } catch { + return "Invalid URL"; + } + + if (preset === "slack" && !isSlackIncomingWebhookUrl(url)) { + return "Slack contact points require an incoming webhook URL (https://hooks.slack.com/services/...)"; + } + + return undefined; +} + +const SLACK_HEADER_MAX = 150; + +const SEVERITY_EMOJI: Record = { + critical: "🔴", + warning: "🟡", + info: "🔵", +}; + +function severityLabel(severity: string): string { + return severity.charAt(0).toUpperCase() + severity.slice(1); +} + +function buildDashboardLink( + dashboardUrl: string | undefined, + event: AlertEvent, +): string | undefined { + if (!dashboardUrl) return undefined; + const base = dashboardUrl.replace(/\/$/, ""); + if (event.queue && event.jobId) { + return `${base}/queues/${encodeURIComponent(event.queue)}/jobs/${encodeURIComponent(event.jobId)}`; + } + if (event.queue) { + return `${base}/queues/${encodeURIComponent(event.queue)}?status=failed`; + } + return `${base}/`; +} + +export interface FormattedDestinationPayload { + url: string; + headers: Record; + body: unknown; +} + +export function formatSlackPayload( + contactPoint: AlertContactPoint, + event: AlertEvent, + dashboardUrl?: string, +): FormattedDestinationPayload { + const emoji = SEVERITY_EMOJI[event.severity] ?? "⚪"; + const link = buildDashboardLink(dashboardUrl, event); + const senderName = contactPoint.displayName ?? DEFAULT_ALERT_SENDER_NAME; + const iconUrl = contactPoint.iconUrl ?? DEFAULT_WORKBENCH_ICON_URL; + + const fields: Array<{ type: string; text: string }> = []; + if (event.queue) { + fields.push({ type: "mrkdwn", text: `*Queue:*\n${event.queue}` }); + } + if (event.jobName || event.jobId) { + fields.push({ + type: "mrkdwn", + text: `*Job:*\n${event.jobName ?? "—"} (\`${event.jobId ?? "—"}\`)`, + }); + } + if (event.failedReason) { + fields.push({ + type: "mrkdwn", + text: `*Reason:*\n${event.failedReason.slice(0, 500)}`, + }); + } + if (event.counts) { + const parts: string[] = []; + if (event.counts.failed !== undefined) { + parts.push(`Failed: ${event.counts.failed}`); + } + if (event.counts.backlog !== undefined) { + parts.push(`Backlog: ${event.counts.backlog}`); + } + if (event.counts.workers !== undefined && event.counts.workers !== null) { + parts.push(`Workers: ${event.counts.workers}`); + } + if (parts.length > 0) { + fields.push({ type: "mrkdwn", text: `*Counts:*\n${parts.join(" · ")}` }); + } + } + + const blocks: unknown[] = [ + { + type: "header", + text: { + type: "plain_text", + text: `${emoji} ${event.ruleName}`.slice(0, SLACK_HEADER_MAX), + emoji: true, + }, + }, + { + type: "section", + text: { + type: "mrkdwn", + text: `*${severityLabel(event.severity)}* · \`${event.trigger}\`\n${event.message}`, + }, + }, + ]; + + if (fields.length > 0) { + blocks.push({ + type: "section", + fields: fields.slice(0, 10), + }); + } + + if (link?.startsWith("https://")) { + blocks.push({ + type: "actions", + elements: [ + { + type: "button", + text: { type: "plain_text", text: "Open in Workbench", emoji: true }, + url: link, + }, + ], + }); + } else if (link) { + blocks.push({ + type: "context", + elements: [{ type: "mrkdwn", text: `<${link}|Open in Workbench>` }], + }); + } + + blocks.push({ + type: "context", + elements: [ + { + type: "mrkdwn", + text: event.status === "resolved" ? "✅ Resolved" : "🔔 Firing", + }, + ], + }); + + const body: Record = { + username: senderName, + icon_url: iconUrl, + blocks, + text: `${emoji} ${event.ruleName}: ${event.message}`, + }; + + return { + url: contactPoint.url, + headers: { "Content-Type": "application/json" }, + body, + }; +} + +export function formatWebhookPayload( + contactPoint: AlertContactPoint, + event: AlertEvent, + dashboardUrl?: string, +): FormattedDestinationPayload { + const link = buildDashboardLink(dashboardUrl, event); + const body = { + source: "workbench", + version: 1, + contactPoint: { id: contactPoint.id, name: contactPoint.name }, + event: { + id: event.id, + ruleId: event.ruleId, + ruleName: event.ruleName, + trigger: event.trigger, + severity: event.severity, + status: event.status, + fingerprint: event.fingerprint, + message: event.message, + queue: event.queue, + jobId: event.jobId, + jobName: event.jobName, + failedReason: event.failedReason, + attemptsMade: event.attemptsMade, + counts: event.counts, + firedAt: event.firedAt, + resolvedAt: event.resolvedAt, + dashboardUrl: link, + }, + }; + + return { + url: contactPoint.url, + headers: { + "Content-Type": "application/json", + "User-Agent": "Workbench-Alerts/1.0", + ...contactPoint.headers, + }, + body, + }; +} + +export function formatDestinationPayload( + contactPoint: AlertContactPoint, + event: AlertEvent, + dashboardUrl?: string, +): FormattedDestinationPayload { + if (contactPoint.preset === "slack") { + return formatSlackPayload(contactPoint, event, dashboardUrl); + } + return formatWebhookPayload(contactPoint, event, dashboardUrl); +} + +export async function deliverToContactPoint( + contactPoint: AlertContactPoint, + event: AlertEvent, + dashboardUrl?: string, +): Promise<{ success: boolean; error?: string }> { + if (!contactPoint.enabled) { + return { success: false, error: "Contact point disabled" }; + } + + const { url, headers, body } = formatDestinationPayload( + contactPoint, + event, + dashboardUrl, + ); + + try { + const res = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify(body), + signal: AbortSignal.timeout(15_000), + }); + if (!res.ok) { + const text = await res.text().catch(() => ""); + let error = `HTTP ${res.status}`; + try { + const json = JSON.parse(text) as { error?: string; ok?: boolean }; + if (json.error) error += `: ${json.error}`; + } catch { + if (text) error += `: ${text.slice(0, 200)}`; + } + return { success: false, error }; + } + return { success: true }; + } catch (err) { + return { + success: false, + error: err instanceof Error ? err.message : "Delivery failed", + }; + } +} diff --git a/packages/core/src/core/alert-manager.ts b/packages/core/src/core/alert-manager.ts new file mode 100644 index 0000000..9c826b6 --- /dev/null +++ b/packages/core/src/core/alert-manager.ts @@ -0,0 +1,482 @@ +import { type Queue, QueueEvents } from "bullmq"; +import { deliverToContactPoint } from "./alert-destinations"; +import { type AlertStore, createAlertStore } from "./alert-store"; +import type { QueueManager } from "./queue-manager"; +import type { + AlertDeliveryRecord, + AlertEvent, + AlertRule, + AlertRuntimeStatus, + AlertsOptions, + AlertTrigger, +} from "./types"; + +const HEALTH_CHECK_INTERVAL_MS = 30_000; +const MAX_RECENT_EVENTS = 100; +const MAX_DELIVERY_RECORDS = 50; + +interface CooldownEntry { + lastFiredAt: number; + event?: AlertEvent; +} + +export class AlertManager { + private readonly store: AlertStore; + private readonly persistence: "redis" | "memory" | "custom"; + private readonly options: AlertsOptions; + private readonly queueManager: QueueManager; + private readonly getQueues: () => Map; + + private queueEvents: QueueEvents[] = []; + private listenerStatus = new Map(); + private healthTimer: ReturnType | null = null; + private lastHealthCheckAt?: number; + private recentEvents: AlertEvent[] = []; + private lastDeliveries: AlertDeliveryRecord[] = []; + private cooldowns = new Map(); + private activeHealthAlerts = new Map(); + private started = false; + private closed = false; + + constructor( + queueManager: QueueManager, + getQueues: () => Map, + options: AlertsOptions, + store?: AlertStore, + persistence: "redis" | "memory" | "custom" = "memory", + ) { + this.queueManager = queueManager; + this.getQueues = getQueues; + this.options = options; + this.store = store ?? createAlertStore(options, {}).store; + this.persistence = persistence; + } + + get enabled(): boolean { + return this.options.enabled !== false; + } + + getStore(): AlertStore { + return this.store; + } + + /** Start QueueEvents listeners and health-check loop */ + async start(): Promise { + if (this.options.enabled === false || this.started || this.closed) return; + this.started = true; + + const queues = this.getQueues(); + for (const [name, queue] of queues) { + const connection = queue.opts?.connection; + if (!connection) { + this.listenerStatus.set(name, false); + continue; + } + + try { + const qe = new QueueEvents(name, { connection }); + this.queueEvents.push(qe); + this.listenerStatus.set(name, true); + + qe.on("failed", ({ jobId, failedReason }) => { + void this.handleQueueEvent(name, "job_failed", { + jobId, + failedReason, + }); + }); + + qe.on("stalled", ({ jobId }) => { + void this.handleQueueEvent(name, "job_stalled", { jobId }); + }); + + qe.on("retries-exhausted", ({ jobId, attemptsMade }) => { + void this.handleQueueEvent(name, "retries_exhausted", { + jobId, + attemptsMade: Number(attemptsMade), + }); + }); + + qe.on("error", () => { + this.listenerStatus.set(name, false); + }); + } catch { + this.listenerStatus.set(name, false); + } + } + + this.healthTimer = setInterval(() => { + void this.runHealthChecks(); + }, HEALTH_CHECK_INTERVAL_MS); + void this.runHealthChecks(); + } + + async close(): Promise { + await this.store.close?.(); + await this.closeListeners(); + } + + private async closeListeners(): Promise { + this.closed = true; + if (this.healthTimer) { + clearInterval(this.healthTimer); + this.healthTimer = null; + } + await Promise.all( + this.queueEvents.map(async (qe) => { + try { + await qe.close(); + } catch { + // ignore close errors + } + }), + ); + this.queueEvents = []; + } + + async getStatus(): Promise { + const listeners = Array.from(this.listenerStatus.entries()).map( + ([queue, connected]) => ({ queue, connected }), + ); + return { + enabled: this.options.enabled !== false, + persistence: this.persistence, + listenerCount: listeners.length, + listeners, + healthCheckIntervalMs: HEALTH_CHECK_INTERVAL_MS, + lastHealthCheckAt: this.lastHealthCheckAt, + recentEvents: [...this.recentEvents], + lastDeliveries: [...this.lastDeliveries], + defaults: { + cooldownMs: this.options.defaults?.cooldownMs ?? 5 * 60 * 1000, + sendResolved: this.options.defaults?.sendResolved ?? true, + }, + }; + } + + /** Send a test notification through a contact point */ + async sendTest(contactPointId: string): Promise { + const cp = await this.store.getContactPoint(contactPointId); + if (!cp) { + return { + contactPointId, + contactPointName: "Unknown", + success: false, + error: "Contact point not found", + at: Date.now(), + }; + } + + const testEvent: AlertEvent = { + id: crypto.randomUUID(), + ruleId: "test", + ruleName: "Test notification", + trigger: "job_failed", + severity: "info", + status: "firing", + fingerprint: "test", + queue: "example", + jobId: "test-job", + jobName: "exampleJob", + message: "This is a test alert from Workbench.", + firedAt: Date.now(), + }; + + const result = await deliverToContactPoint( + cp, + testEvent, + this.options.dashboardUrl, + ); + const record: AlertDeliveryRecord = { + contactPointId: cp.id, + contactPointName: cp.name, + success: result.success, + error: result.error, + at: Date.now(), + }; + this.pushDelivery(record); + return record; + } + + /** Preview what a rule would look like without sending */ + previewRule(rule: AlertRule): AlertEvent { + return { + id: crypto.randomUUID(), + ruleId: rule.id, + ruleName: rule.name, + trigger: rule.trigger, + severity: rule.severity, + status: "firing", + fingerprint: `preview:${rule.id}`, + queue: rule.queues?.[0] ?? "my-queue", + jobId: "12345", + jobName: "exampleJob", + message: this.buildMessage(rule, { + queue: rule.queues?.[0] ?? "my-queue", + jobName: "exampleJob", + failedReason: "Example failure reason", + }), + failedReason: "Example failure reason", + firedAt: Date.now(), + }; + } + + private async handleQueueEvent( + queueName: string, + trigger: AlertTrigger, + ctx: { jobId: string; failedReason?: string; attemptsMade?: number }, + ): Promise { + const rules = await this.store.getRules(); + const matching = rules.filter( + (r) => + r.enabled && r.trigger === trigger && this.matchesQueue(r, queueName), + ); + if (matching.length === 0) return; + + let jobName: string | undefined; + if (ctx.jobId) { + const job = await this.queueManager.getJob(queueName, ctx.jobId); + jobName = job?.name; + if (jobName && matching.every((r) => !this.matchesJobName(r, jobName!))) { + return; + } + } + + for (const rule of matching) { + if (jobName && !this.matchesJobName(rule, jobName)) continue; + await this.fireRule(rule, { + queue: queueName, + jobId: ctx.jobId, + jobName, + failedReason: ctx.failedReason, + attemptsMade: ctx.attemptsMade, + }); + } + } + + private async runHealthChecks(): Promise { + if (this.options.enabled === false || this.closed) return; + this.lastHealthCheckAt = Date.now(); + + const rules = await this.store.getRules(); + const healthRules = rules.filter( + (r) => + r.enabled && + (r.trigger === "failed_backlog" || + r.trigger === "no_workers_with_backlog"), + ); + if (healthRules.length === 0) return; + + let queuesInfo: Awaited>; + try { + queuesInfo = await this.queueManager.getQueues(); + } catch { + return; + } + + for (const rule of healthRules) { + for (const queue of queuesInfo) { + if (!this.matchesQueue(rule, queue.name)) continue; + + const backlog = + queue.counts.waiting + + queue.counts.prioritized + + queue.counts["waiting-children"]; + + let firing = false; + let message = ""; + const counts: AlertEvent["counts"] = { + failed: queue.counts.failed, + backlog, + workers: queue.workerCount, + }; + + if (rule.trigger === "failed_backlog") { + const threshold = rule.threshold ?? 1; + if (queue.counts.failed >= threshold) { + firing = true; + message = `${queue.counts.failed} failed jobs in ${queue.name} (threshold: ${threshold})`; + } + } else if (rule.trigger === "no_workers_with_backlog") { + if ( + queue.workerCount === 0 && + backlog > 0 && + queue.workerCount !== null && + queue.workerCount !== undefined + ) { + firing = true; + message = `${backlog} jobs waiting in ${queue.name} with no workers connected`; + } + } + + const fp = `${rule.id}:${queue.name}:${rule.trigger}`; + const existing = this.activeHealthAlerts.get(fp); + + if (firing) { + if (!existing) { + await this.fireRule(rule, { queue: queue.name, counts, message }); + // fireRule stores in recentEvents; track active for resolve + const last = this.recentEvents[0]; + if (last?.fingerprint === fp) { + this.activeHealthAlerts.set(fp, last); + } + } + } else if (existing && this.options.defaults?.sendResolved !== false) { + await this.resolveHealthAlert(existing, rule); + this.activeHealthAlerts.delete(fp); + } else if (existing) { + this.activeHealthAlerts.delete(fp); + } + } + } + } + + private async resolveHealthAlert( + event: AlertEvent, + rule: AlertRule, + ): Promise { + const resolved: AlertEvent = { + ...event, + id: crypto.randomUUID(), + status: "resolved", + message: `Resolved: ${event.message}`, + resolvedAt: Date.now(), + }; + this.pushEvent(resolved); + await this.deliver(resolved, rule); + } + + private async fireRule( + rule: AlertRule, + ctx: { + queue?: string; + jobId?: string; + jobName?: string; + failedReason?: string; + attemptsMade?: number; + counts?: AlertEvent["counts"]; + message?: string; + }, + ): Promise { + const fingerprint = this.buildFingerprint(rule, ctx); + const cooldownMs = + rule.cooldownMs ?? this.options.defaults?.cooldownMs ?? 5 * 60 * 1000; + const prev = this.cooldowns.get(fingerprint); + if (prev && Date.now() - prev.lastFiredAt < cooldownMs) { + return; + } + + const event: AlertEvent = { + id: crypto.randomUUID(), + ruleId: rule.id, + ruleName: rule.name, + trigger: rule.trigger, + severity: rule.severity, + status: "firing", + fingerprint, + queue: ctx.queue, + jobId: ctx.jobId, + jobName: ctx.jobName, + message: ctx.message ?? this.buildMessage(rule, ctx), + failedReason: ctx.failedReason, + attemptsMade: ctx.attemptsMade, + counts: ctx.counts, + firedAt: Date.now(), + }; + + this.cooldowns.set(fingerprint, { lastFiredAt: Date.now(), event }); + this.pushEvent(event); + await this.deliver(event, rule); + } + + private async deliver(event: AlertEvent, rule: AlertRule): Promise { + const allCps = await this.store.getContactPoints(); + const cps = allCps.filter( + (cp) => cp.enabled && rule.contactPointIds.includes(cp.id), + ); + + await Promise.all( + cps.map(async (cp) => { + const result = await deliverToContactPoint( + cp, + event, + this.options.dashboardUrl, + ); + this.pushDelivery({ + contactPointId: cp.id, + contactPointName: cp.name, + success: result.success, + error: result.error, + at: Date.now(), + }); + }), + ); + } + + private buildFingerprint( + rule: AlertRule, + ctx: { + queue?: string; + jobId?: string; + }, + ): string { + const parts = [rule.id, rule.trigger, ctx.queue ?? "all"]; + if ( + rule.trigger === "job_failed" || + rule.trigger === "job_stalled" || + rule.trigger === "retries_exhausted" + ) { + parts.push(ctx.jobId ?? "unknown"); + } + return parts.join(":"); + } + + private buildMessage( + rule: AlertRule, + ctx: { + queue?: string; + jobId?: string; + jobName?: string; + failedReason?: string; + attemptsMade?: number; + }, + ): string { + switch (rule.trigger) { + case "job_failed": + return `Job ${ctx.jobName ?? "unknown"} failed in ${ctx.queue ?? "queue"}${ctx.failedReason ? `: ${ctx.failedReason.slice(0, 200)}` : ""}`; + case "job_stalled": + return `Job ${ctx.jobName ?? ctx.jobId ?? "unknown"} stalled in ${ctx.queue ?? "queue"}`; + case "retries_exhausted": + return `Job ${ctx.jobName ?? ctx.jobId ?? "unknown"} exhausted retries in ${ctx.queue ?? "queue"}${ctx.attemptsMade ? ` (${ctx.attemptsMade} attempts)` : ""}`; + case "failed_backlog": + return `Failed job backlog in ${ctx.queue ?? "queue"}`; + case "no_workers_with_backlog": + return `No workers processing ${ctx.queue ?? "queue"}`; + default: + return rule.name; + } + } + + private matchesQueue(rule: AlertRule, queueName: string): boolean { + if (!rule.queues || rule.queues.length === 0) return true; + return rule.queues.includes(queueName); + } + + private matchesJobName(rule: AlertRule, jobName: string): boolean { + if (!rule.jobNames || rule.jobNames.length === 0) return true; + return rule.jobNames.includes(jobName); + } + + private pushEvent(event: AlertEvent): void { + this.recentEvents.unshift(event); + if (this.recentEvents.length > MAX_RECENT_EVENTS) { + this.recentEvents.length = MAX_RECENT_EVENTS; + } + } + + private pushDelivery(record: AlertDeliveryRecord): void { + this.lastDeliveries.unshift(record); + if (this.lastDeliveries.length > MAX_DELIVERY_RECORDS) { + this.lastDeliveries.length = MAX_DELIVERY_RECORDS; + } + } +} diff --git a/packages/core/src/core/alert-store.ts b/packages/core/src/core/alert-store.ts new file mode 100644 index 0000000..af378a1 --- /dev/null +++ b/packages/core/src/core/alert-store.ts @@ -0,0 +1,195 @@ +import { RedisAlertStore } from "./redis-alert-store"; +import type { + AlertContactPoint, + AlertContactPointPublic, + AlertPersistence, + AlertRule, + AlertStore, + AlertsOptions, +} from "./types"; + +function maskUrl(url: string): string { + try { + const parsed = new URL(url); + const segments = parsed.pathname.split("/").filter(Boolean); + const secret = segments[segments.length - 1] ?? ""; + const tail = secret.length >= 4 ? secret.slice(-4) : "****"; + return `${parsed.hostname}/…${tail}`; + } catch { + return "••••••••"; + } +} + +export function toPublicContactPoint( + cp: AlertContactPoint, +): AlertContactPointPublic { + const { url, ...rest } = cp; + return { ...rest, urlMasked: maskUrl(url) }; +} + +/** + * Use {@link createAlertStore} — defaults to Redis when a connection is available. + */ +export type { AlertPersistence, AlertStore } from "./types"; + +export interface CreateAlertStoreContext { + /** Workbench-level Redis connection (auto-discovery) */ + redis?: string | import("bullmq").RedisOptions; + /** Connection from the first mounted BullMQ queue (BullMQ `ConnectionOptions`) */ + queueConnection?: unknown; + /** Key prefix for Redis storage; defaults to Workbench `prefix` or `"bull"` */ + prefix?: string; +} + +/** + * Resolve the alert config store. Redis is the default when a connection exists. + * Code-defined `contactPoints` / `rules` seed Redis on first run only. + */ +export function createAlertStore( + alerts: AlertsOptions, + ctx: CreateAlertStoreContext, +): { store: AlertStore; persistence: AlertPersistence } { + if (alerts.store) { + return { store: alerts.store, persistence: "custom" }; + } + + if (alerts.persistence === "memory") { + return { + store: new MemoryAlertStore({ + contactPoints: alerts.contactPoints, + rules: alerts.rules, + }), + persistence: "memory", + }; + } + + const connection = (ctx.queueConnection ?? ctx.redis) as + | string + | import("bullmq").RedisOptions + | undefined; + if (connection) { + return { + store: new RedisAlertStore({ + connection, + prefix: alerts.storagePrefix ?? ctx.prefix ?? "bull", + seed: { + contactPoints: alerts.contactPoints, + rules: alerts.rules, + }, + }), + persistence: "redis", + }; + } + + return { + store: new MemoryAlertStore({ + contactPoints: alerts.contactPoints, + rules: alerts.rules, + }), + persistence: "memory", + }; +} + +export class MemoryAlertStore implements AlertStore { + private contactPoints = new Map(); + private rules = new Map(); + + constructor(seed?: Pick) { + for (const cp of seed?.contactPoints ?? []) { + this.contactPoints.set(cp.id, { ...cp }); + } + for (const rule of seed?.rules ?? []) { + this.rules.set(rule.id, { ...rule }); + } + } + + async getContactPoints(): Promise { + return Array.from(this.contactPoints.values()).sort((a, b) => + a.name.localeCompare(b.name), + ); + } + + async getContactPoint(id: string): Promise { + return this.contactPoints.get(id); + } + + async createContactPoint( + input: Omit, + ): Promise { + const now = Date.now(); + const cp: AlertContactPoint = { + ...input, + id: crypto.randomUUID(), + createdAt: now, + updatedAt: now, + }; + this.contactPoints.set(cp.id, cp); + return cp; + } + + async updateContactPoint( + id: string, + input: Partial>, + ): Promise { + const existing = this.contactPoints.get(id); + if (!existing) return undefined; + const updated: AlertContactPoint = { + ...existing, + ...input, + id, + createdAt: existing.createdAt, + updatedAt: Date.now(), + }; + this.contactPoints.set(id, updated); + return updated; + } + + async deleteContactPoint(id: string): Promise { + return this.contactPoints.delete(id); + } + + async getRules(): Promise { + return Array.from(this.rules.values()).sort((a, b) => + a.name.localeCompare(b.name), + ); + } + + async getRule(id: string): Promise { + return this.rules.get(id); + } + + async createRule( + input: Omit, + ): Promise { + const now = Date.now(); + const rule: AlertRule = { + ...input, + id: crypto.randomUUID(), + createdAt: now, + updatedAt: now, + }; + this.rules.set(rule.id, rule); + return rule; + } + + async updateRule( + id: string, + input: Partial>, + ): Promise { + const existing = this.rules.get(id); + if (!existing) return undefined; + const updated: AlertRule = { + ...existing, + ...input, + id, + createdAt: existing.createdAt, + updatedAt: Date.now(), + }; + this.rules.set(id, updated); + return updated; + } + + async deleteRule(id: string): Promise { + return this.rules.delete(id); + } +} diff --git a/packages/core/src/core/queue-manager.ts b/packages/core/src/core/queue-manager.ts index e76834c..5b5caa4 100644 --- a/packages/core/src/core/queue-manager.ts +++ b/packages/core/src/core/queue-manager.ts @@ -318,6 +318,13 @@ export class QueueManager { return this.queues.get(name); } + /** + * Internal map of queue instances (for alerting and advanced integrations) + */ + getQueueMap(): Map { + return this.queues; + } + /** * Get information for all queues (cached) */ diff --git a/packages/core/src/core/redis-alert-store.ts b/packages/core/src/core/redis-alert-store.ts new file mode 100644 index 0000000..7dd4381 --- /dev/null +++ b/packages/core/src/core/redis-alert-store.ts @@ -0,0 +1,190 @@ +import type { RedisOptions } from "bullmq"; +import { Redis } from "ioredis"; +import type { + AlertContactPoint, + AlertRule, + AlertStore, + AlertsOptions, +} from "./types"; + +export interface RedisAlertStoreOptions { + connection: string | RedisOptions; + /** BullMQ-style prefix; keys are `${prefix}:workbench:alerts:*` */ + prefix?: string; + /** Imported once when Redis has no stored config yet */ + seed?: Pick; +} + +function createRedisClient(connection: string | RedisOptions): Redis { + if (typeof connection === "string") { + return new Redis(connection, { maxRetriesPerRequest: null }); + } + const { url, ...rest } = connection as RedisOptions & { url?: string }; + if (url) { + return new Redis(url, { ...rest, maxRetriesPerRequest: null }); + } + return new Redis({ ...rest, maxRetriesPerRequest: null }); +} + +/** + * Persists alert contact points and rules in the user's Redis. + * Webhook URLs and rules created in the dashboard survive process restarts. + */ +export class RedisAlertStore implements AlertStore { + private readonly client: Redis; + private readonly contactPointsKey: string; + private readonly rulesKey: string; + private readonly seed?: Pick; + private seeded = false; + + constructor(options: RedisAlertStoreOptions) { + this.client = createRedisClient(options.connection); + const prefix = options.prefix ?? "bull"; + this.contactPointsKey = `${prefix}:workbench:alerts:contact-points`; + this.rulesKey = `${prefix}:workbench:alerts:rules`; + this.seed = options.seed; + } + + async close(): Promise { + await this.client.quit().catch(() => this.client.disconnect()); + } + + private async ensureSeeded(): Promise { + if (this.seeded) return; + this.seeded = true; + + const [cpCount, ruleCount] = await Promise.all([ + this.client.hlen(this.contactPointsKey), + this.client.hlen(this.rulesKey), + ]); + + if (cpCount > 0 || ruleCount > 0) return; + if (!this.seed?.contactPoints?.length && !this.seed?.rules?.length) return; + + const pipeline = this.client.pipeline(); + for (const cp of this.seed.contactPoints ?? []) { + pipeline.hset(this.contactPointsKey, cp.id, JSON.stringify(cp)); + } + for (const rule of this.seed.rules ?? []) { + pipeline.hset(this.rulesKey, rule.id, JSON.stringify(rule)); + } + await pipeline.exec(); + } + + private async readHash(key: string): Promise { + const raw = await this.client.hgetall(key); + const items: T[] = []; + for (const value of Object.values(raw)) { + try { + items.push(JSON.parse(value) as T); + } catch { + // skip corrupt entries + } + } + return items; + } + + async getContactPoints(): Promise { + await this.ensureSeeded(); + const items = await this.readHash(this.contactPointsKey); + return items.sort((a, b) => a.name.localeCompare(b.name)); + } + + async getContactPoint(id: string): Promise { + await this.ensureSeeded(); + const raw = await this.client.hget(this.contactPointsKey, id); + if (!raw) return undefined; + return JSON.parse(raw) as AlertContactPoint; + } + + async createContactPoint( + input: Omit, + ): Promise { + await this.ensureSeeded(); + const now = Date.now(); + const cp: AlertContactPoint = { + ...input, + id: crypto.randomUUID(), + createdAt: now, + updatedAt: now, + }; + await this.client.hset(this.contactPointsKey, cp.id, JSON.stringify(cp)); + return cp; + } + + async updateContactPoint( + id: string, + input: Partial>, + ): Promise { + await this.ensureSeeded(); + const existing = await this.getContactPoint(id); + if (!existing) return undefined; + const updated: AlertContactPoint = { + ...existing, + ...input, + id, + createdAt: existing.createdAt, + updatedAt: Date.now(), + }; + await this.client.hset(this.contactPointsKey, id, JSON.stringify(updated)); + return updated; + } + + async deleteContactPoint(id: string): Promise { + await this.ensureSeeded(); + const removed = await this.client.hdel(this.contactPointsKey, id); + return removed > 0; + } + + async getRules(): Promise { + await this.ensureSeeded(); + const items = await this.readHash(this.rulesKey); + return items.sort((a, b) => a.name.localeCompare(b.name)); + } + + async getRule(id: string): Promise { + await this.ensureSeeded(); + const raw = await this.client.hget(this.rulesKey, id); + if (!raw) return undefined; + return JSON.parse(raw) as AlertRule; + } + + async createRule( + input: Omit, + ): Promise { + await this.ensureSeeded(); + const now = Date.now(); + const rule: AlertRule = { + ...input, + id: crypto.randomUUID(), + createdAt: now, + updatedAt: now, + }; + await this.client.hset(this.rulesKey, rule.id, JSON.stringify(rule)); + return rule; + } + + async updateRule( + id: string, + input: Partial>, + ): Promise { + await this.ensureSeeded(); + const existing = await this.getRule(id); + if (!existing) return undefined; + const updated: AlertRule = { + ...existing, + ...input, + id, + createdAt: existing.createdAt, + updatedAt: Date.now(), + }; + await this.client.hset(this.rulesKey, id, JSON.stringify(updated)); + return updated; + } + + async deleteRule(id: string): Promise { + await this.ensureSeeded(); + const removed = await this.client.hdel(this.rulesKey, id); + return removed > 0; + } +} diff --git a/packages/core/src/core/types.ts b/packages/core/src/core/types.ts index 8b7479c..a9660d9 100644 --- a/packages/core/src/core/types.ts +++ b/packages/core/src/core/types.ts @@ -48,6 +48,160 @@ export interface WorkbenchOptions { * 100. Ignored when `queues` is set explicitly. */ maxQueues?: number; + /** Self-hosted alerting configuration */ + alerts?: AlertsOptions; +} + +// ───────────────────────────────────────────────────────────────────────────── +// Alert Types +// ───────────────────────────────────────────────────────────────────────────── + +/** Supported alert trigger types */ +export type AlertTrigger = + | "job_failed" + | "job_stalled" + | "retries_exhausted" + | "failed_backlog" + | "no_workers_with_backlog"; + +export type AlertSeverity = "critical" | "warning" | "info"; + +export type AlertContactPointPreset = "slack" | "webhook"; + +/** Where notifications are sent (Slack incoming webhook or generic webhook) */ +export interface AlertContactPoint { + id: string; + name: string; + preset: AlertContactPointPreset; + /** Webhook URL — stored server-side; API responses mask this value */ + url: string; + enabled: boolean; + displayName?: string; + iconUrl?: string; + /** Extra headers for generic webhook preset */ + headers?: Record; + createdAt: number; + updatedAt: number; +} + +/** Rule that maps a trigger to one or more contact points */ +export interface AlertRule { + id: string; + name: string; + enabled: boolean; + trigger: AlertTrigger; + severity: AlertSeverity; + /** Empty or omitted = all queues */ + queues?: string[]; + /** Optional job name filter for event triggers */ + jobNames?: string[]; + /** Threshold for backlog-style triggers (failed count or waiting count) */ + threshold?: number; + contactPointIds: string[]; + cooldownMs?: number; + createdAt: number; + updatedAt: number; +} + +/** Pluggable persistence for alert contact points and rules */ +export interface AlertStore { + getContactPoints(): Promise; + getContactPoint(id: string): Promise; + createContactPoint( + input: Omit, + ): Promise; + updateContactPoint( + id: string, + input: Partial>, + ): Promise; + deleteContactPoint(id: string): Promise; + getRules(): Promise; + getRule(id: string): Promise; + createRule( + input: Omit, + ): Promise; + updateRule( + id: string, + input: Partial>, + ): Promise; + deleteRule(id: string): Promise; + close?(): Promise; +} + +export type AlertPersistence = "redis" | "memory" | "custom"; + +export interface AlertsOptions { + /** Set to `false` to disable alerting entirely. Default: on. */ + enabled?: boolean; + /** Seed data imported into Redis on first run when the store is empty */ + contactPoints?: AlertContactPoint[]; + /** Seed data imported into Redis on first run when the store is empty */ + rules?: AlertRule[]; + /** Override the config store entirely */ + store?: AlertStore; + /** Where to persist dashboard-managed config. Default: `"redis"` when a connection exists */ + persistence?: "redis" | "memory"; + /** Redis key prefix for stored config. Default: Workbench `prefix` or `"bull"` */ + storagePrefix?: string; + defaults?: { + cooldownMs?: number; + sendResolved?: boolean; + }; + /** Public dashboard URL included in notification links */ + dashboardUrl?: string; +} + +/** Normalized alert event for delivery and activity log */ +export interface AlertEvent { + id: string; + ruleId: string; + ruleName: string; + trigger: AlertTrigger; + severity: AlertSeverity; + status: "firing" | "resolved"; + fingerprint: string; + queue?: string; + jobId?: string; + jobName?: string; + message: string; + failedReason?: string; + attemptsMade?: number; + counts?: { + failed?: number; + backlog?: number; + workers?: number | null; + }; + firedAt: number; + resolvedAt?: number; +} + +/** Contact point as returned by the API (URL masked) */ +export type AlertContactPointPublic = Omit & { + urlMasked: string; +}; + +export interface AlertDeliveryRecord { + contactPointId: string; + contactPointName: string; + success: boolean; + error?: string; + at: number; +} + +/** Runtime status exposed to the dashboard */ +export interface AlertRuntimeStatus { + enabled: boolean; + persistence: AlertPersistence; + listenerCount: number; + listeners: Array<{ queue: string; connected: boolean }>; + healthCheckIntervalMs: number; + lastHealthCheckAt?: number; + recentEvents: AlertEvent[]; + lastDeliveries: AlertDeliveryRecord[]; + defaults: { + cooldownMs: number; + sendResolved: boolean; + }; } /** diff --git a/packages/core/src/core/workbench.ts b/packages/core/src/core/workbench.ts index 627a1ba..d24a01c 100644 --- a/packages/core/src/core/workbench.ts +++ b/packages/core/src/core/workbench.ts @@ -1,4 +1,6 @@ import type { Queue, RedisOptions } from "bullmq"; +import { AlertManager } from "./alert-manager"; +import { createAlertStore } from "./alert-store"; import { discoverQueues } from "./discover"; import { QueueManager } from "./queue-manager"; import type { WorkbenchOptions } from "./types"; @@ -24,6 +26,8 @@ export class WorkbenchCore { WorkbenchOptions; readonly queueManager: QueueManager; readonly discovery: DiscoveryMeta | null; + readonly alertManager: AlertManager | null; + readonly alertsPersistence: "redis" | "memory" | "custom" | null; constructor( options: WorkbenchOptions | Queue[], @@ -49,6 +53,30 @@ export class WorkbenchCore { } this.queueManager = new QueueManager(explicit, this.options.tags || []); + + if (this.options.alerts?.enabled !== false) { + const alertsOpts = this.options.alerts ?? {}; + const queueConnection = explicit[0]?.opts?.connection; + const { store, persistence } = createAlertStore(alertsOpts, { + queueConnection, + redis: this.options.redis, + prefix: this.options.prefix ?? "bull", + }); + this.alertsPersistence = persistence; + this.alertManager = new AlertManager( + this.queueManager, + () => this.queueManager.getQueueMap(), + alertsOpts, + store, + persistence, + ); + void this.alertManager.start().catch((err) => { + console.error("[workbench] Failed to start alert manager:", err); + }); + } else { + this.alertsPersistence = null; + this.alertManager = null; + } } /** @@ -121,6 +149,8 @@ export class WorkbenchCore { queues: this.queueManager.getQueueNames(), tags: this.queueManager.getTagFields(), discovery: this.discovery, + alertsEnabled: this.alertManager !== null, + alertsPersistence: this.alertsPersistence, }; } } diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index a855cd8..2188705 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -10,12 +10,30 @@ export { type HttpMethod, type RouteDef, } from "./api/handlers"; - +export { AlertManager } from "./core/alert-manager"; +export { + createAlertStore, + MemoryAlertStore, + toPublicContactPoint, +} from "./core/alert-store"; export { discoverQueues } from "./core/discover"; export { QueueManager } from "./core/queue-manager"; +export { RedisAlertStore } from "./core/redis-alert-store"; export type { ActivityBucket, ActivityStatsResponse, + AlertContactPoint, + AlertContactPointPreset, + AlertContactPointPublic, + AlertDeliveryRecord, + AlertEvent, + AlertPersistence, + AlertRule, + AlertRuntimeStatus, + AlertSeverity, + AlertStore, + AlertsOptions, + AlertTrigger, CreateFlowChildRequest, CreateFlowRequest, DelayedJobInfo, diff --git a/packages/core/src/ui/components/app-sidebar.tsx b/packages/core/src/ui/components/app-sidebar.tsx index 43d1df0..2832076 100644 --- a/packages/core/src/ui/components/app-sidebar.tsx +++ b/packages/core/src/ui/components/app-sidebar.tsx @@ -3,6 +3,7 @@ import type { LucideIcon } from "lucide-react"; import { BarChart3, + Bell, Clock, FlaskConical, Hourglass, @@ -59,6 +60,7 @@ export type NavItem = | "schedulers" | "flows" | "queues" + | "alerts" | "test"; interface AppSidebarProps { @@ -80,6 +82,7 @@ const mainNavItems = [ { id: "metrics" as const, label: "Metrics", icon: BarChart3 }, { id: "schedulers" as const, label: "Schedulers", icon: Clock }, { id: "flows" as const, label: "Flows", icon: Network }, + { id: "alerts" as const, label: "Alerts", icon: Bell }, { id: "test" as const, label: "Test", icon: FlaskConical }, ]; diff --git a/packages/core/src/ui/components/layout/command-palette.tsx b/packages/core/src/ui/components/layout/command-palette.tsx index 2a9da74..1a59e63 100644 --- a/packages/core/src/ui/components/layout/command-palette.tsx +++ b/packages/core/src/ui/components/layout/command-palette.tsx @@ -1,6 +1,7 @@ import { ArrowRight, BarChart3, + Bell, Calendar, FileText, FlaskConical, @@ -83,6 +84,13 @@ const navigationItems: NavigationItem[] = [ icon: , keywords: ["flows", "workflows", "graph"], }, + { + id: "alerts", + label: "Alerts", + path: "/alerts", + icon: , + keywords: ["alerts", "notifications", "slack", "webhook"], + }, { id: "test", label: "Test", diff --git a/packages/core/src/ui/lib/api.ts b/packages/core/src/ui/lib/api.ts index 434c628..e9cf4ed 100644 --- a/packages/core/src/ui/lib/api.ts +++ b/packages/core/src/ui/lib/api.ts @@ -1,5 +1,11 @@ import type { ActivityStatsResponse, + AlertContactPoint, + AlertContactPointPublic, + AlertDeliveryRecord, + AlertEvent, + AlertRule, + AlertRuntimeStatus, CreateFlowRequest, DelayedJobInfo, FlowNode, @@ -299,6 +305,8 @@ export const api = { capped: boolean; cap: number; } | null; + alertsEnabled: boolean; + alertsPersistence: "redis" | "memory" | "custom" | null; }> { return fetchJson(getConfigUrl()); }, @@ -533,4 +541,88 @@ export const api = { body: JSON.stringify(request), }); }, + + // ───────────────────────────────────────────────────────────────────────────── + // Alerts + // ───────────────────────────────────────────────────────────────────────────── + + async getAlertStatus(): Promise { + return fetchJson(`${API_BASE}/alerts/status`); + }, + + async getAlertContactPoints(): Promise { + return fetchJson(`${API_BASE}/alerts/contact-points`); + }, + + async createAlertContactPoint( + input: Omit, + ): Promise { + return fetchJson(`${API_BASE}/alerts/contact-points`, { + method: "POST", + body: JSON.stringify(input), + }); + }, + + async updateAlertContactPoint( + id: string, + input: Partial>, + ): Promise { + return fetchJson( + `${API_BASE}/alerts/contact-points/${encodeURIComponent(id)}`, + { + method: "PUT", + body: JSON.stringify(input), + }, + ); + }, + + async deleteAlertContactPoint(id: string): Promise<{ success: boolean }> { + return fetchJson( + `${API_BASE}/alerts/contact-points/${encodeURIComponent(id)}`, + { method: "DELETE" }, + ); + }, + + async testAlertContactPoint(id: string): Promise { + return fetchJson( + `${API_BASE}/alerts/contact-points/${encodeURIComponent(id)}/test`, + { method: "POST" }, + ); + }, + + async getAlertRules(): Promise { + return fetchJson(`${API_BASE}/alerts/rules`); + }, + + async createAlertRule( + input: Omit, + ): Promise { + return fetchJson(`${API_BASE}/alerts/rules`, { + method: "POST", + body: JSON.stringify(input), + }); + }, + + async updateAlertRule( + id: string, + input: Partial>, + ): Promise { + return fetchJson(`${API_BASE}/alerts/rules/${encodeURIComponent(id)}`, { + method: "PUT", + body: JSON.stringify(input), + }); + }, + + async deleteAlertRule(id: string): Promise<{ success: boolean }> { + return fetchJson(`${API_BASE}/alerts/rules/${encodeURIComponent(id)}`, { + method: "DELETE", + }); + }, + + async previewAlertRule(id: string): Promise { + return fetchJson( + `${API_BASE}/alerts/rules/${encodeURIComponent(id)}/preview`, + { method: "POST" }, + ); + }, }; diff --git a/packages/core/src/ui/lib/hooks.ts b/packages/core/src/ui/lib/hooks.ts index 991d1bc..c647720 100644 --- a/packages/core/src/ui/lib/hooks.ts +++ b/packages/core/src/ui/lib/hooks.ts @@ -46,6 +46,11 @@ export const queryKeys = { flows: ["flows"] as const, flow: (queueName: string, jobId: string) => ["flow", queueName, jobId] as const, + alerts: { + status: ["alerts", "status"] as const, + contactPoints: ["alerts", "contact-points"] as const, + rules: ["alerts", "rules"] as const, + }, }; /** @@ -520,3 +525,119 @@ export function useCreateFlow() { }, }); } + +// ───────────────────────────────────────────────────────────────────────────── +// Alerts +// ───────────────────────────────────────────────────────────────────────────── + +export function useAlertStatus() { + return useQuery({ + queryKey: queryKeys.alerts.status, + queryFn: () => api.getAlertStatus(), + refetchInterval: 10_000, + retry: false, + }); +} + +export function useAlertContactPoints() { + return useQuery({ + queryKey: queryKeys.alerts.contactPoints, + queryFn: () => api.getAlertContactPoints(), + retry: false, + }); +} + +export function useAlertRules() { + return useQuery({ + queryKey: queryKeys.alerts.rules, + queryFn: () => api.getAlertRules(), + retry: false, + }); +} + +export function useCreateAlertContactPoint() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (input: Parameters[0]) => + api.createAlertContactPoint(input), + onSuccess: () => { + queryClient.invalidateQueries({ + queryKey: queryKeys.alerts.contactPoints, + }); + }, + }); +} + +export function useUpdateAlertContactPoint() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: ({ + id, + ...input + }: { id: string } & Parameters[1]) => + api.updateAlertContactPoint(id, input), + onSuccess: () => { + queryClient.invalidateQueries({ + queryKey: queryKeys.alerts.contactPoints, + }); + }, + }); +} + +export function useDeleteAlertContactPoint() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (id: string) => api.deleteAlertContactPoint(id), + onSuccess: () => { + queryClient.invalidateQueries({ + queryKey: queryKeys.alerts.contactPoints, + }); + queryClient.invalidateQueries({ queryKey: queryKeys.alerts.rules }); + }, + }); +} + +export function useTestAlertContactPoint() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (id: string) => api.testAlertContactPoint(id), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: queryKeys.alerts.status }); + }, + }); +} + +export function useCreateAlertRule() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (input: Parameters[0]) => + api.createAlertRule(input), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: queryKeys.alerts.rules }); + }, + }); +} + +export function useUpdateAlertRule() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: ({ + id, + ...input + }: { id: string } & Parameters[1]) => + api.updateAlertRule(id, input), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: queryKeys.alerts.rules }); + }, + }); +} + +export function useDeleteAlertRule() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (id: string) => api.deleteAlertRule(id), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: queryKeys.alerts.rules }); + }, + }); +} diff --git a/packages/core/src/ui/pages/alerts.tsx b/packages/core/src/ui/pages/alerts.tsx new file mode 100644 index 0000000..5154447 --- /dev/null +++ b/packages/core/src/ui/pages/alerts.tsx @@ -0,0 +1,1029 @@ +import { + Bell, + CheckCircle2, + Loader2, + Plus, + Send, + Trash2, + Webhook, + XCircle, +} from "lucide-react"; +import * as React from "react"; +import { toast } from "sonner"; +import { EmptyState } from "@/components/shared/empty-state"; +import { Alert, AlertDescription, AlertTitle } from "@/components/ui/alert"; +import { + AlertDialog, + AlertDialogAction, + AlertDialogCancel, + AlertDialogContent, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogTitle, + AlertDialogTrigger, +} from "@/components/ui/alert-dialog"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { + Dialog, + DialogContent, + DialogFooter, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import { Input } from "@/components/ui/input"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; +import { Skeleton } from "@/components/ui/skeleton"; +import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; +import type { + AlertContactPointPreset, + AlertContactPointPublic, + AlertRule, + AlertRuntimeStatus, + AlertSeverity, + AlertTrigger, +} from "@/core/types"; +import { + useAlertContactPoints, + useAlertRules, + useAlertStatus, + useConfig, + useCreateAlertContactPoint, + useCreateAlertRule, + useDeleteAlertContactPoint, + useDeleteAlertRule, + useTestAlertContactPoint, + useUpdateAlertContactPoint, + useUpdateAlertRule, +} from "@/lib/hooks"; +import { cn } from "@/lib/utils"; + +const TRIGGERS: { value: AlertTrigger; label: string; description: string }[] = + [ + { + value: "job_failed", + label: "Job failed", + description: "A job throws an error", + }, + { + value: "job_stalled", + label: "Job stalled", + description: "A job stops making progress", + }, + { + value: "retries_exhausted", + label: "Retries exhausted", + description: "A job used all retry attempts", + }, + { + value: "failed_backlog", + label: "Failed backlog", + description: "Failed job count crosses a threshold", + }, + { + value: "no_workers_with_backlog", + label: "No workers with backlog", + description: "Waiting jobs but no active workers", + }, + ]; + +const SEVERITIES: AlertSeverity[] = ["critical", "warning", "info"]; + +const DESTINATION_TYPES: { + value: AlertContactPointPreset; + label: string; + description: string; +}[] = [ + { + value: "slack", + label: "Slack", + description: "Incoming webhook to a channel", + }, + { + value: "webhook", + label: "Webhook", + description: "Custom HTTPS endpoint (JSON payload)", + }, +]; + +function destinationTypeLabel(preset: AlertContactPointPreset): string { + return DESTINATION_TYPES.find((d) => d.value === preset)?.label ?? preset; +} + +function triggerLabel(trigger: AlertTrigger): string { + return TRIGGERS.find((t) => t.value === trigger)?.label ?? trigger; +} + +function formatCooldown(ms: number | undefined): string { + const minutes = Math.round((ms ?? 300_000) / 60_000); + if (minutes < 60) return `${minutes} min`; + if (minutes % 60 === 0) return `${minutes / 60} hr`; + return `${(minutes / 60).toFixed(1)} hr`; +} + +function SeverityBadge({ severity }: { severity: AlertSeverity }) { + const variant = + severity === "critical" + ? "destructive" + : severity === "warning" + ? "secondary" + : "outline"; + return {severity}; +} + +function ContactPointForm({ + open, + onOpenChange, + initial, + onSubmit, +}: { + open: boolean; + onOpenChange: (open: boolean) => void; + initial?: AlertContactPointPublic; + onSubmit: (data: { + name: string; + preset: AlertContactPointPreset; + url: string; + displayName?: string; + enabled: boolean; + }) => void; +}) { + const [name, setName] = React.useState(initial?.name ?? ""); + const [preset, setPreset] = React.useState( + initial?.preset ?? "slack", + ); + const [url, setUrl] = React.useState(""); + const [displayName, setDisplayName] = React.useState( + initial?.displayName ?? "", + ); + + React.useEffect(() => { + if (open) { + setName(initial?.name ?? ""); + setPreset(initial?.preset ?? "slack"); + setUrl(""); + setDisplayName(initial?.displayName ?? ""); + } + }, [open, initial]); + + return ( + + + + + {initial ? "Edit destination" : "Add destination"} + + +
+
+

Name

+ setName(e.target.value)} + placeholder="Ops Slack" + /> +
+
+

Type

+ +

+ {DESTINATION_TYPES.find((d) => d.value === preset)?.description} +

+
+ {preset === "slack" && ( + + Slack incoming webhook + +
    +
  1. + Open{" "} + + api.slack.com/apps + {" "} + and create or select an app +
    2. +
  2. + Enable Incoming Webhooks and add a webhook + to your channel +
    3. +
  3. + Copy the URL ( + + https://hooks.slack.com/services/… + + ) and paste it below +
    4. +
  4. + Save, then use Test to verify delivery +
    5. +
+ + + )} +
+

+ {preset === "slack" ? "Slack webhook URL" : "Webhook URL"} + {initial ? " (leave blank to keep current)" : ""} +

+ setUrl(e.target.value)} + /> +
+
+

+ Sender name{" "} + (optional) +

+ setDisplayName(e.target.value)} + /> +

+ Shown on Slack messages when your workspace allows it. +

+
+
+ + + + + + + ); +} + +function RuleForm({ + open, + onOpenChange, + initial, + contactPoints, + queues, + onSubmit, +}: { + open: boolean; + onOpenChange: (open: boolean) => void; + initial?: AlertRule; + contactPoints: AlertContactPointPublic[]; + queues: string[]; + onSubmit: (data: Omit) => void; +}) { + const [name, setName] = React.useState(initial?.name ?? ""); + const [trigger, setTrigger] = React.useState( + initial?.trigger ?? "job_failed", + ); + const [severity, setSeverity] = React.useState( + initial?.severity ?? "warning", + ); + const [threshold, setThreshold] = React.useState( + String(initial?.threshold ?? 1), + ); + const [cooldownMinutes, setCooldownMinutes] = React.useState( + String(Math.round((initial?.cooldownMs ?? 300_000) / 60_000)), + ); + const [contactPointIds, setContactPointIds] = React.useState( + initial?.contactPointIds ?? [], + ); + const [queueFilter, setQueueFilter] = React.useState( + initial?.queues?.join(", ") ?? "", + ); + + React.useEffect(() => { + if (open) { + setName(initial?.name ?? ""); + setTrigger(initial?.trigger ?? "job_failed"); + setSeverity(initial?.severity ?? "warning"); + setThreshold(String(initial?.threshold ?? 1)); + setCooldownMinutes( + String(Math.round((initial?.cooldownMs ?? 300_000) / 60_000)), + ); + setContactPointIds(initial?.contactPointIds ?? []); + setQueueFilter(initial?.queues?.join(", ") ?? ""); + } + }, [open, initial]); + + const needsThreshold = + trigger === "failed_backlog" || trigger === "no_workers_with_backlog"; + + return ( + + + + + {initial ? "Edit alert rule" : "New alert rule"} + + +
+
+

Name

+ setName(e.target.value)} + placeholder="Failed jobs in production" + /> +
+
+
+

When

+ +
+
+

Severity

+ +
+
+ {needsThreshold && ( +
+

Threshold

+ setThreshold(e.target.value)} + /> +
+ )} +
+

+ Queues{" "} + (optional) +

+ setQueueFilter(e.target.value)} + /> +

+ Comma-separated names. Leave empty to watch every queue. +

+
+
+

+ Cooldown (minutes) +

+ setCooldownMinutes(e.target.value)} + /> +

+ Minimum wait before the same alert can fire again. +

+
+
+

Send to

+
+ {contactPoints.length === 0 ? ( +

+ Add a destination first, then link it here. +

+ ) : ( + contactPoints.map((cp) => ( + + )) + )} +
+
+
+ + + + + + + ); +} + +function SetupGuide({ + hasDestinations, + hasRules, +}: { + hasDestinations: boolean; + hasRules: boolean; +}) { + if (hasDestinations && hasRules) return null; + + const steps = [ + { + done: hasDestinations, + label: "Add a destination", + detail: "Slack channel or webhook URL", + }, + { + done: hasRules, + label: "Create an alert rule", + detail: "Choose when to notify", + }, + ]; + + return ( +
+

Getting started

+
    + {steps.map((step, i) => ( +
  1. + + {step.done ? "✓" : i + 1} + + + + {step.label} + + {!step.done && ( + + {step.detail} + + )} + +
    2. + ))} +
+
+ ); +} + +function DestinationIcon({ preset }: { preset: AlertContactPointPreset }) { + const Icon = preset === "slack" ? Bell : Webhook; + return ( +
+ +
+ ); +} + +function ActivityTab({ status }: { status: AlertRuntimeStatus }) { + return ( +
+
+

+ Event listeners +

+ {status.listeners.length === 0 ? ( +

+ No queues connected yet. +

+ ) : ( +
+ {status.listeners.map((l) => ( +
+ {l.queue} + {l.connected ? ( + + Active + + ) : ( + Disconnected + )} +
+ ))} +
+ )} +
+ +
+

+ Recent alerts +

+ {status.recentEvents.length === 0 ? ( +

+ Nothing has fired yet. Alerts appear here when a rule matches. +

+ ) : ( +
+ {status.recentEvents.slice(0, 20).map((ev) => ( +
+
+ + {ev.ruleName} + {ev.status} +
+

{ev.message}

+
+ ))} +
+ )} +
+ +
+

+ Delivery log +

+ {status.lastDeliveries.length === 0 ? ( +

+ Test messages and alert deliveries will show up here. +

+ ) : ( +
+ {status.lastDeliveries.slice(0, 10).map((d) => ( +
+ {d.success ? ( + + ) : ( + + )} + {d.contactPointName} + {d.error && {d.error}} +
+ ))} +
+ )} +
+
+ ); +} + +export function AlertsPage() { + const { data: config } = useConfig(); + const { data: status, isLoading: statusLoading } = useAlertStatus(); + const { data: contactPoints = [], isLoading: cpLoading } = + useAlertContactPoints(); + const { data: rules = [], isLoading: rulesLoading } = useAlertRules(); + const createCp = useCreateAlertContactPoint(); + const updateCp = useUpdateAlertContactPoint(); + const deleteCp = useDeleteAlertContactPoint(); + const testCp = useTestAlertContactPoint(); + const createRule = useCreateAlertRule(); + const updateRule = useUpdateAlertRule(); + const deleteRule = useDeleteAlertRule(); + + const [cpDialogOpen, setCpDialogOpen] = React.useState(false); + const [editCp, setEditCp] = React.useState(); + const [ruleDialogOpen, setRuleDialogOpen] = React.useState(false); + const [editRule, setEditRule] = React.useState(); + + const readonly = config?.readonly; + const queues = config?.queues ?? []; + + if (statusLoading || cpLoading || rulesLoading) { + return ( +
+ + +
+ ); + } + + return ( +
+ 0} + hasRules={rules.length > 0} + /> + + + + + Destinations + {contactPoints.length > 0 && ( + + {contactPoints.length} + + )} + + + Rules + {rules.length > 0 && ( + + {rules.length} + + )} + + Activity + + + + {!readonly && ( + + )} + {contactPoints.length === 0 ? ( + { + setEditCp(undefined); + setCpDialogOpen(true); + }} + > + + Add destination + + ) : undefined + } + /> + ) : ( +
+ {contactPoints.map((cp) => ( +
+ +
+
+ {cp.name} + + {destinationTypeLabel(cp.preset)} + + {!cp.enabled && Paused} +
+

+ {cp.urlMasked} + {cp.displayName ? ` · sends as ${cp.displayName}` : ""} +

+
+
+ {!readonly && ( + <> + + + + + + + + + + Delete destination? + + + Rules linked to {cp.name} will stop sending + there. + + + + Cancel + deleteCp.mutate(cp.id)} + > + Delete + + + + + + )} +
+
+ ))} +
+ )} + + + + {!readonly && ( + + )} + {contactPoints.length === 0 && !readonly && ( +

+ Add a destination before creating rules. +

+ )} + {rules.length === 0 ? ( + 0 ? ( + + ) : undefined + } + /> + ) : ( +
+ {rules.map((rule) => ( +
+
+
+ {rule.name} + + + {triggerLabel(rule.trigger)} + + {!rule.enabled && ( + Paused + )} +
+

+ {rule.queues?.length + ? `Queues: ${rule.queues.join(", ")}` + : "All queues"} + {rule.threshold !== undefined + ? ` · threshold ${rule.threshold}` + : ""} + {rule.cooldownMs !== undefined + ? ` · cooldown ${formatCooldown(rule.cooldownMs)}` + : ""} +

+
+ {!readonly && ( +
+ + + + + + + + + Delete rule? + + + Cancel + deleteRule.mutate(rule.id)} + > + Delete + + + + +
+ )} +
+ ))} +
+ )} + + + + {status ? ( + + ) : ( +

+ Unable to load status. +

+ )} + + + + { + if (editCp) { + updateCp.mutate({ + id: editCp.id, + name: data.name, + preset: data.preset, + displayName: data.displayName, + enabled: data.enabled, + ...(data.url ? { url: data.url } : {}), + }); + } else { + createCp.mutate(data); + } + }} + /> + + { + if (editRule) { + updateRule.mutate({ id: editRule.id, ...data }); + } else { + createRule.mutate(data); + } + }} + /> +
+ ); +} diff --git a/packages/core/src/ui/router.tsx b/packages/core/src/ui/router.tsx index e974db3..3c2d2d4 100644 --- a/packages/core/src/ui/router.tsx +++ b/packages/core/src/ui/router.tsx @@ -16,6 +16,7 @@ import { HeaderSearch } from "@/components/layout/header-search"; import { SidebarInset, SidebarProvider } from "@/components/ui/sidebar"; import { Toaster } from "@/components/ui/sonner"; import { useConfig, useQueueNames, useQueues } from "@/lib/hooks"; +import { AlertsPage } from "@/pages/alerts"; import { FlowPage } from "@/pages/flow"; import { FlowsPage } from "@/pages/flows"; import { JobPage } from "@/pages/job"; @@ -177,6 +178,9 @@ function RootLayout() { if (path === "/test") { return { activeNav: "test" as NavItem, activeQueue: undefined }; } + if (path === "/alerts") { + return { activeNav: "alerts" as NavItem, activeQueue: undefined }; + } if (path.startsWith("/queues/")) { const queueName = path.split("/")[2]; return { activeNav: "queues" as NavItem, activeQueue: queueName }; @@ -249,6 +253,9 @@ function RootLayout() { case "flows": navigate({ to: "/flows" }); break; + case "alerts": + navigate({ to: "/alerts" }); + break; case "test": navigate({ to: "/test" }); break; @@ -473,6 +480,14 @@ function TestRoute() { ); } +function AlertsRoute() { + return ( + + + + ); +} + function QueueRoute() { const { queueName } = useParams({ from: "/queues/$queueName" }); const { data: config } = useConfig(); @@ -591,6 +606,12 @@ const testRoute = createRoute({ validateSearch: testSearchSchema, }); +const alertsRoute = createRoute({ + getParentRoute: () => rootRoute, + path: "/alerts", + component: AlertsRoute, +}); + const queueRoute = createRoute({ getParentRoute: () => rootRoute, path: "/queues/$queueName", @@ -614,6 +635,7 @@ const routeTree = rootRoute.addChildren([ flowsRoute, flowDetailRoute, testRoute, + alertsRoute, queueRoute, jobRoute, ]); diff --git a/packages/core/tsconfig.json b/packages/core/tsconfig.json index 495e042..7bdafa0 100644 --- a/packages/core/tsconfig.json +++ b/packages/core/tsconfig.json @@ -12,5 +12,5 @@ } }, "include": ["src/**/*.ts", "src/**/*.tsx"], - "exclude": ["node_modules", "dist"] + "exclude": ["node_modules", "dist", "src/**/*.test.ts"] } From e19720c266a290dc79337a315ed38738dbf4f835 Mon Sep 17 00:00:00 2001 From: Pontus Abrahamsson Date: Sun, 31 May 2026 13:50:18 +0200 Subject: [PATCH 2/3] release: 0.8.0 self-hosted alerting Bump @getworkbench/core and all adapters to 0.8.0 for the dashboard Alerts feature. Co-authored-by: Cursor --- CHANGELOG.md | 10 ++++++++-- RELEASING.md | 13 ++++++++++--- packages/adonis/package.json | 2 +- packages/astro/package.json | 2 +- packages/bun/package.json | 2 +- packages/core/package.json | 2 +- packages/elysia/package.json | 2 +- packages/express/package.json | 2 +- packages/fastify/package.json | 2 +- packages/h3/package.json | 2 +- packages/hono/package.json | 2 +- packages/koa/package.json | 2 +- packages/nestjs/package.json | 2 +- packages/next/package.json | 2 +- packages/nuxt/package.json | 2 +- packages/tanstack-start/package.json | 2 +- 16 files changed, 32 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 83a70ec..0931f9a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,10 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.8.0] - 2026-05-30 + ### Added -- **`@getworkbench/core`** Self-hosted alerting: BullMQ `QueueEvents` listeners, Slack/webhook presets, dashboard **Alerts** page, and Redis-backed persistence for contact points and rules (keys under `{prefix}:workbench:alerts:*`). Alerts are enabled by default; configure contact points and rules in the dashboard to start receiving notifications. -- **`@getworkbench/core`** Slack incoming webhook validation, Block Kit payload fixes (HTTPS dashboard buttons, HTTP dev links), and in-dialog Slack setup instructions. +- **`@getworkbench/core`** Self-hosted alerting: BullMQ `QueueEvents` listeners, Slack/webhook presets, dashboard **Alerts** page, and Redis-backed persistence for contact points and rules (keys under `{prefix}:workbench:alerts:*`). Alerts are enabled by default; configure destinations and rules in the dashboard to start receiving notifications. +- **`@getworkbench/core`** Slack incoming webhook validation, Block Kit payload formatting (HTTPS dashboard buttons, HTTP dev links), and in-dialog Slack setup instructions. + +### Changed + +- All 14 publishable adapter packages (`@getworkbench/core`, `hono`, `elysia`, `express`, `fastify`, `nestjs`, `next`, `koa`, `astro`, `nuxt`, `bun`, `h3`, `adonis`, `tanstack-start`) bump together to `0.8.0`. ## [0.7.1] - 2026-05-28 diff --git a/RELEASING.md b/RELEASING.md index b5eea25..ea348e5 100644 --- a/RELEASING.md +++ b/RELEASING.md @@ -57,8 +57,8 @@ packages must bump together** because they share `@getworkbench/core` via # 1. Bump core + every adapter in lockstep. `npm version` does not handle # `workspace:*` deps, so edit the `"version"` field in each package.json # directly (or use a tiny sed script). -for p in core express fastify next nestjs elysia hono; do - sed -i.bak -E "s/(\"version\": )\"[^\"]+\"/\\1\"0.3.0\"/" "packages/$p/package.json" +for p in core express fastify hono elysia next nestjs koa astro nuxt bun h3 adonis tanstack-start; do + sed -i.bak -E "s/(\"version\": )\"[^\"]+\"/\\1\"0.8.0\"/" "packages/$p/package.json" rm "packages/$p/package.json.bak" done @@ -76,8 +76,15 @@ bun run typecheck (cd packages/fastify && bun publish --access public) (cd packages/hono && bun publish --access public) (cd packages/elysia && bun publish --access public) +(cd packages/koa && bun publish --access public) (cd packages/next && bun publish --access public) -(cd packages/nestjs && bun publish --access public) # depends on express + fastify, publish last +(cd packages/nestjs && bun publish --access public) +(cd packages/astro && bun publish --access public) +(cd packages/nuxt && bun publish --access public) +(cd packages/bun && bun publish --access public) +(cd packages/h3 && bun publish --access public) +(cd packages/adonis && bun publish --access public) +(cd packages/tanstack-start && bun publish --access public) # (cd packages/cli && bun publish --access public) # only if you bumped it # 5. Commit, tag, push. diff --git a/packages/adonis/package.json b/packages/adonis/package.json index 9890d17..78f28a4 100644 --- a/packages/adonis/package.json +++ b/packages/adonis/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/adonis", - "version": "0.7.0", + "version": "0.8.0", "description": "AdonisJS adapter for Workbench — modern BullMQ dashboard with broader framework coverage than bull-board.", "type": "module", "main": "./dist/index.js", diff --git a/packages/astro/package.json b/packages/astro/package.json index 8a8d0aa..f7914d4 100644 --- a/packages/astro/package.json +++ b/packages/astro/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/astro", - "version": "0.7.0", + "version": "0.8.0", "description": "Astro adapter for Workbench — modern BullMQ dashboard with broader framework coverage than bull-board.", "type": "module", "main": "./dist/index.js", diff --git a/packages/bun/package.json b/packages/bun/package.json index 95e9dd1..d2e296b 100644 --- a/packages/bun/package.json +++ b/packages/bun/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/bun", - "version": "0.7.0", + "version": "0.8.0", "description": "Bun.serve adapter for Workbench — modern BullMQ dashboard with broader framework coverage than bull-board.", "type": "module", "main": "./dist/index.js", diff --git a/packages/core/package.json b/packages/core/package.json index 94a6b08..2715e4f 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/core", - "version": "0.7.1", + "version": "0.8.0", "description": "Core of Workbench — QueueManager, API router, and bundled UI. Modern alternative to `@bull-board/api`.", "type": "module", "main": "./dist/index.js", diff --git a/packages/elysia/package.json b/packages/elysia/package.json index a82a341..a986a9a 100644 --- a/packages/elysia/package.json +++ b/packages/elysia/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/elysia", - "version": "0.7.0", + "version": "0.8.0", "description": "Elysia adapter for Workbench — modern BullMQ dashboard, drop-in alternative to `@bull-board/elysia`.", "type": "module", "main": "./dist/index.js", diff --git a/packages/express/package.json b/packages/express/package.json index 7fb0f7e..8d4e35b 100644 --- a/packages/express/package.json +++ b/packages/express/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/express", - "version": "0.7.0", + "version": "0.8.0", "description": "Express adapter for Workbench — modern BullMQ dashboard, drop-in alternative to `@bull-board/express`.", "type": "module", "main": "./dist/index.js", diff --git a/packages/fastify/package.json b/packages/fastify/package.json index 9b2a02e..1a23280 100644 --- a/packages/fastify/package.json +++ b/packages/fastify/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/fastify", - "version": "0.7.0", + "version": "0.8.0", "description": "Fastify adapter for Workbench — modern BullMQ dashboard, drop-in alternative to `@bull-board/fastify`.", "type": "module", "main": "./dist/index.js", diff --git a/packages/h3/package.json b/packages/h3/package.json index a37ad17..7c0fb02 100644 --- a/packages/h3/package.json +++ b/packages/h3/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/h3", - "version": "0.7.0", + "version": "0.8.0", "description": "h3 adapter for Workbench — modern BullMQ dashboard, drop-in alternative to `@bull-board/h3`.", "type": "module", "main": "./dist/index.js", diff --git a/packages/hono/package.json b/packages/hono/package.json index c215cc3..6911582 100644 --- a/packages/hono/package.json +++ b/packages/hono/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/hono", - "version": "0.7.0", + "version": "0.8.0", "description": "Hono adapter for Workbench — modern BullMQ dashboard, drop-in alternative to `@bull-board/hono`.", "type": "module", "main": "./dist/index.js", diff --git a/packages/koa/package.json b/packages/koa/package.json index 814ad6e..8255ed7 100644 --- a/packages/koa/package.json +++ b/packages/koa/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/koa", - "version": "0.7.0", + "version": "0.8.0", "description": "Koa adapter for Workbench — modern BullMQ dashboard, drop-in alternative to `@bull-board/koa`.", "type": "module", "main": "./dist/index.js", diff --git a/packages/nestjs/package.json b/packages/nestjs/package.json index 3afea61..7e11abb 100644 --- a/packages/nestjs/package.json +++ b/packages/nestjs/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/nestjs", - "version": "0.7.0", + "version": "0.8.0", "description": "NestJS adapter for Workbench — modern BullMQ dashboard, drop-in alternative to `@bull-board/nestjs`.", "type": "module", "main": "./dist/index.js", diff --git a/packages/next/package.json b/packages/next/package.json index 710a245..b743e4a 100644 --- a/packages/next/package.json +++ b/packages/next/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/next", - "version": "0.7.0", + "version": "0.8.0", "description": "Next.js App Router adapter for Workbench — modern BullMQ dashboard with broader framework coverage than bull-board.", "type": "module", "main": "./dist/index.js", diff --git a/packages/nuxt/package.json b/packages/nuxt/package.json index f28fe53..0bbea8c 100644 --- a/packages/nuxt/package.json +++ b/packages/nuxt/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/nuxt", - "version": "0.7.0", + "version": "0.8.0", "description": "Nuxt (Nitro/h3) adapter for Workbench — modern BullMQ dashboard with broader framework coverage than bull-board.", "type": "module", "main": "./dist/index.js", diff --git a/packages/tanstack-start/package.json b/packages/tanstack-start/package.json index c7417ea..9bee21f 100644 --- a/packages/tanstack-start/package.json +++ b/packages/tanstack-start/package.json @@ -1,6 +1,6 @@ { "name": "@getworkbench/tanstack-start", - "version": "0.7.0", + "version": "0.8.0", "description": "TanStack Start adapter for Workbench — modern BullMQ dashboard with broader framework coverage than bull-board.", "type": "module", "main": "./dist/index.js", From e1889f2d0a237a0ed53b7e5566ef043d11483061 Mon Sep 17 00:00:00 2001 From: Pontus Abrahamsson Date: Sun, 31 May 2026 13:51:51 +0200 Subject: [PATCH 3/3] chore(web): remove Bull Board comparison from marketing site Drop the comparison section, nav link, and footer link from the homepage and blog chrome. Co-authored-by: Cursor --- apps/web/src/app/page.tsx | 60 -------------------- apps/web/src/components/blog/blog-chrome.tsx | 7 --- 2 files changed, 67 deletions(-) diff --git a/apps/web/src/app/page.tsx b/apps/web/src/app/page.tsx index d2efd76..7fc02aa 100644 --- a/apps/web/src/app/page.tsx +++ b/apps/web/src/app/page.tsx @@ -14,9 +14,7 @@ import { Twitter, } from "lucide-react"; import Image from "next/image"; -import Link from "next/link"; import { ActionButton } from "../components/action-button"; -import { CompareTable } from "../components/blog/compare-table"; import { CopyCommand } from "../components/copy-command"; import { AdonisLogo, @@ -43,7 +41,6 @@ import { TestMockup, } from "../components/mockups"; import { ThemeToggle } from "../components/theme-toggle"; -import { COMPARISON_ROWS } from "../lib/blog/comparison"; /** Update once the desktop app has a tagged GitHub Release. */ const MAC_DOWNLOAD_URL = @@ -58,7 +55,6 @@ const DOCKER_RUN_COMMAND = `docker run --rm -p 3000:3000 \\ const SPONSORS_URL = "https://github.com/sponsors/pontusab"; const TWITTER_URL = "https://x.com/pontusab"; const DOCS_URL = "https://github.com/pontusab/workbench#readme"; -const COMPARISON_URL = "/blog/workbench-vs-bull-board"; const INSTALL_COMMAND = "npx @getworkbench/cli init"; /** @@ -277,8 +273,6 @@ export default function Page() { - -