feat: refactor stage 2

apps/api/src/index.ts

@@ -3,6 +3,7 @@ import { ingest } from "./routes/pings";
 import { monitors } from "./routes/monitors";
 import { account } from "./routes/auth";
 import { internal } from "./routes/internal";
+import { channels } from "./routes/channels";
 import { migrate } from "./db";
 import { SECURITY_HEADERS } from "../../shared/auth";
 
@@ -16,6 +17,7 @@ const elysia = new Elysia()
   }))
   .use(account)
   .use(monitors)
+  .use(channels)
   .use(ingest)
   .use(internal);
 

apps/api/src/notifications/index.ts

@@ -0,0 +1,39 @@
+import sql from "../db";
+import type { NotificationProvider, ChannelRow, NotificationEvent } from "./types";
+import { webhook } from "./webhook";
+
+// Modular provider registry. Add new providers by importing them here and
+// dropping them in the map; nothing else needs to change.
+const providers: Record<string, NotificationProvider> = {
+  webhook,
+};
+
+export function knownProviderKinds(): string[] {
+  return Object.keys(providers);
+}
+
+export async function dispatch(channel: ChannelRow, event: NotificationEvent): Promise<void> {
+  const p = providers[channel.kind];
+  if (!p) {
+    console.warn(`[notify] unknown provider kind: ${channel.kind}`);
+    return;
+  }
+  try {
+    await p.send(channel, event);
+  } catch (e) {
+    console.warn(`[notify] ${channel.kind} send failed for channel ${channel.id}:`, e);
+  }
+}
+
+export async function dispatchForMonitor(monitorId: string, event: NotificationEvent): Promise<void> {
+  const channels = await sql<ChannelRow[]>`
+    SELECT c.id, c.account_id, c.name, c.kind, c.config, c.enabled
+    FROM notification_channels c
+    JOIN monitor_notifications mn ON mn.channel_id = c.id
+    WHERE mn.monitor_id = ${monitorId} AND c.enabled = true
+  `;
+  if (channels.length === 0) return;
+  await Promise.all(channels.map((c) => dispatch(c, event)));
+}
+
+export type { NotificationEvent, ChannelRow, MonitorContext, PingContext } from "./types";

apps/api/src/notifications/types.ts

@@ -0,0 +1,33 @@
+export interface ChannelRow {
+  id:         string;
+  account_id: string;
+  name:       string;
+  kind:       string;
+  config:     any;
+  enabled:    boolean;
+}
+
+export interface MonitorContext {
+  id:     string;
+  name:   string;
+  url:    string;
+  region: string;
+}
+
+export interface PingContext {
+  status_code: number | null;
+  latency_ms:  number | null;
+  error:       string | null;
+  checked_at:  string;
+}
+
+export type NotificationEvent =
+  | { kind: "down"; monitor: MonitorContext; ping: PingContext }
+  | { kind: "up";   monitor: MonitorContext; ping: PingContext }
+  | { kind: "cert"; monitor: MonitorContext; days: number }
+  | { kind: "test"; monitor: MonitorContext };
+
+export interface NotificationProvider {
+  kind: string;
+  send(channel: ChannelRow, event: NotificationEvent): Promise<void>;
+}

apps/api/src/notifications/webhook.ts

@@ -0,0 +1,38 @@
+import { createHmac } from "crypto";
+import type { NotificationProvider, ChannelRow, NotificationEvent } from "./types";
+
+export const webhook: NotificationProvider = {
+  kind: "webhook",
+  async send(channel: ChannelRow, event: NotificationEvent) {
+    const cfg = channel.config || {};
+    const url: string | undefined = cfg.url;
+    if (!url) throw new Error("webhook channel missing config.url");
+
+    const payload = JSON.stringify({
+      channel: { id: channel.id, name: channel.name },
+      event,
+    });
+
+    const headers: Record<string, string> = {
+      "content-type": "application/json",
+      "user-agent":   "PingQL-Notifier/1",
+    };
+    if (cfg.headers && typeof cfg.headers === "object") {
+      for (const [k, v] of Object.entries(cfg.headers)) {
+        if (typeof v === "string") headers[k.toLowerCase()] = v;
+      }
+    }
+    if (typeof cfg.secret === "string" && cfg.secret.length) {
+      headers["x-pingql-signature"] = createHmac("sha256", cfg.secret).update(payload).digest("hex");
+    }
+
+    const ctrl = new AbortController();
+    const timer = setTimeout(() => ctrl.abort(), 5000);
+    try {
+      const resp = await fetch(url, { method: "POST", headers, body: payload, signal: ctrl.signal });
+      if (!resp.ok) throw new Error(`webhook ${url} returned ${resp.status}`);
+    } finally {
+      clearTimeout(timer);
+    }
+  },
+};

apps/api/src/routes/channels.ts

@@ -0,0 +1,112 @@
+import { Elysia, t } from "elysia";
+import { requireAuth } from "./auth";
+import sql from "../db";
+import { dispatch, knownProviderKinds, type ChannelRow } from "../notifications";
+
+const ChannelBody = t.Object({
+  name:    t.String({ minLength: 1, maxLength: 200 }),
+  kind:    t.String({ description: "Provider kind, e.g. 'webhook'" }),
+  config:  t.Any({ description: "Provider-specific config object" }),
+  enabled: t.Optional(t.Boolean()),
+});
+
+function validateKind(kind: string): string | null {
+  if (!knownProviderKinds().includes(kind)) {
+    return `Unknown provider kind '${kind}'. Known: ${knownProviderKinds().join(", ")}`;
+  }
+  return null;
+}
+
+function validateWebhookConfig(config: any): string | null {
+  if (!config || typeof config !== "object") return "config must be an object";
+  if (typeof config.url !== "string" || !/^https?:\/\//.test(config.url)) {
+    return "webhook config.url must be a http(s) URL";
+  }
+  return null;
+}
+
+function validateConfig(kind: string, config: any): string | null {
+  if (kind === "webhook") return validateWebhookConfig(config);
+  return null;
+}
+
+export const channels = new Elysia({ prefix: "/notifications/channels" })
+  .use(requireAuth)
+
+  .get("/", async ({ accountId }) => {
+    return sql`
+      SELECT id, name, kind, config, enabled, created_at
+      FROM notification_channels
+      WHERE account_id = ${accountId}
+      ORDER BY created_at DESC
+    `;
+  }, { detail: { summary: "List notification channels", tags: ["notifications"] } })
+
+  .post("/", async ({ accountId, body, set }) => {
+    const kindErr = validateKind(body.kind);
+    if (kindErr) { set.status = 400; return { error: kindErr }; }
+    const cfgErr = validateConfig(body.kind, body.config);
+    if (cfgErr) { set.status = 400; return { error: cfgErr }; }
+
+    const [row] = await sql`
+      INSERT INTO notification_channels (account_id, name, kind, config, enabled)
+      VALUES (${accountId}, ${body.name}, ${body.kind}, ${sql.json(body.config)}, ${body.enabled ?? true})
+      RETURNING id, name, kind, config, enabled, created_at
+    `;
+    return row;
+  }, { body: ChannelBody, detail: { summary: "Create notification channel", tags: ["notifications"] } })
+
+  .patch("/:id", async ({ accountId, params, body, set }) => {
+    if (body.kind != null) {
+      const kindErr = validateKind(body.kind);
+      if (kindErr) { set.status = 400; return { error: kindErr }; }
+    }
+    if (body.config != null) {
+      const kind = body.kind ?? (await sql`SELECT kind FROM notification_channels WHERE id = ${params.id} AND account_id = ${accountId}`)[0]?.kind;
+      if (kind) {
+        const cfgErr = validateConfig(kind, body.config);
+        if (cfgErr) { set.status = 400; return { error: cfgErr }; }
+      }
+    }
+
+    const [row] = await sql`
+      UPDATE notification_channels SET
+        name    = COALESCE(${body.name ?? null},                                       name),
+        kind    = COALESCE(${body.kind ?? null},                                       kind),
+        config  = COALESCE(${body.config != null ? sql.json(body.config) : null},      config),
+        enabled = COALESCE(${body.enabled ?? null},                                    enabled)
+      WHERE id = ${params.id} AND account_id = ${accountId}
+      RETURNING id, name, kind, config, enabled, created_at
+    `;
+    if (!row) { set.status = 404; return { error: "Not found" }; }
+    return row;
+  }, { body: t.Partial(ChannelBody), detail: { summary: "Update notification channel", tags: ["notifications"] } })
+
+  .delete("/:id", async ({ accountId, params, set }) => {
+    const [row] = await sql`
+      DELETE FROM notification_channels
+      WHERE id = ${params.id} AND account_id = ${accountId}
+      RETURNING id
+    `;
+    if (!row) { set.status = 404; return { error: "Not found" }; }
+    return { deleted: true };
+  }, { detail: { summary: "Delete notification channel", tags: ["notifications"] } })
+
+  .post("/:id/test", async ({ accountId, params, set }) => {
+    const [row] = await sql<ChannelRow[]>`
+      SELECT id, account_id, name, kind, config, enabled
+      FROM notification_channels
+      WHERE id = ${params.id} AND account_id = ${accountId}
+    `;
+    if (!row) { set.status = 404; return { error: "Not found" }; }
+    try {
+      await dispatch(row, {
+        kind:    "test",
+        monitor: { id: "test", name: "Test event", url: "https://example.com", region: "" },
+      });
+      return { ok: true };
+    } catch (e: any) {
+      set.status = 502;
+      return { error: e?.message || String(e) };
+    }
+  }, { detail: { summary: "Send a test event to a channel", tags: ["notifications"] } });

apps/api/src/routes/monitors.ts

@@ -15,10 +15,27 @@ const MonitorBody = t.Object({
   max_retries:     t.Optional(t.Number({ minimum: 0, maximum: 10, default: 0, description: "Retry a failing check up to N times before declaring DOWN" })),
   retry_interval_s: t.Optional(t.Number({ minimum: 1, maximum: 600, default: 30, description: "Seconds between retries" })),
   resend_interval: t.Optional(t.Number({ minimum: 0, maximum: 1000, default: 0, description: "Re-alert every Nth consecutive down beat. 0 = never resend." })),
+  cert_alert_days: t.Optional(t.Number({ minimum: 0, maximum: 365, default: 14, description: "Alert when TLS cert is within N days of expiry. 0 disables." })),
   query:           t.Optional(t.Any({ description: "PingQL query — filter conditions for up/down" })),
   regions:         t.Optional(t.Array(t.String(), { description: "Regions to run checks from. Empty array = all regions." })),
+  channel_ids:     t.Optional(t.Array(t.String(), { description: "Notification channel IDs to attach to this monitor." })),
 });
 
+async function replaceMonitorChannels(monitorId: string, accountId: string, channelIds: string[]) {
+  await sql`DELETE FROM monitor_notifications WHERE monitor_id = ${monitorId}`;
+  if (channelIds.length === 0) return;
+  // Only attach channels that belong to the same account.
+  const owned = await sql<{ id: string }[]>`
+    SELECT id FROM notification_channels
+    WHERE account_id = ${accountId} AND id = ANY(${sql.array(channelIds)})
+  `;
+  if (owned.length === 0) return;
+  await sql`
+    INSERT INTO monitor_notifications (monitor_id, channel_id)
+    SELECT ${monitorId}, id FROM UNNEST(${sql.array(owned.map((o) => o.id))}::uuid[]) AS id
+  `;
+}
+
 export const monitors = new Elysia({ prefix: "/monitors" })
   .use(requireAuth)
 
@@ -50,7 +67,7 @@ export const monitors = new Elysia({ prefix: "/monitors" })
     const ssrfError = await validateMonitorUrl(body.url);
     if (ssrfError) { set.status = 400; return { error: ssrfError }; }
     const [monitor] = await sql`
-      INSERT INTO monitors (account_id, name, url, method, request_headers, request_body, timeout_ms, interval_s, max_retries, retry_interval_s, resend_interval, query, regions)
+      INSERT INTO monitors (account_id, name, url, method, request_headers, request_body, timeout_ms, interval_s, max_retries, retry_interval_s, resend_interval, cert_alert_days, query, regions)
       VALUES (
         ${accountId}, ${body.name}, ${body.url},
         ${(body.method ?? 'GET').toUpperCase()},
@@ -61,11 +78,13 @@ export const monitors = new Elysia({ prefix: "/monitors" })
         ${body.max_retries ?? 0},
         ${body.retry_interval_s ?? 30},
         ${body.resend_interval ?? 0},
+        ${body.cert_alert_days ?? 14},
         ${body.query ? sql.json(body.query) : null},
         ${sql.array(regions)}
       )
       RETURNING *
     `;
+    if (body.channel_ids) await replaceMonitorChannels(monitor.id, accountId, body.channel_ids);
     return monitor;
   }, { body: MonitorBody, detail: { summary: "Create monitor", tags: ["monitors"] } })
 
@@ -79,7 +98,10 @@ export const monitors = new Elysia({ prefix: "/monitors" })
       SELECT * FROM pings WHERE monitor_id = ${params.id}
       ORDER BY checked_at DESC LIMIT 100
     `;
-    return { ...monitor, results };
+    const channels = await sql<{ channel_id: string }[]>`
+      SELECT channel_id FROM monitor_notifications WHERE monitor_id = ${params.id}
+    `;
+    return { ...monitor, results, channel_ids: channels.map((c) => c.channel_id) };
   }, { detail: { summary: "Get monitor with results", tags: ["monitors"] } })
 
   .patch("/:id", async ({ accountId, plan, params, body, set }) => {
@@ -112,12 +134,14 @@ export const monitors = new Elysia({ prefix: "/monitors" })
         max_retries     = COALESCE(${body.max_retries ?? null}, max_retries),
         retry_interval_s = COALESCE(${body.retry_interval_s ?? null}, retry_interval_s),
         resend_interval = COALESCE(${body.resend_interval ?? null}, resend_interval),
+        cert_alert_days = COALESCE(${body.cert_alert_days ?? null}, cert_alert_days),
         query           = COALESCE(${body.query ? sql.json(body.query) : null}, query),
         regions         = COALESCE(${body.regions ? sql.array(body.regions) : null}, regions)
       WHERE id = ${params.id} AND account_id = ${accountId}
       RETURNING *
     `;
     if (!monitor) { set.status = 404; return { error: "Not found" }; }
+    if (body.channel_ids) await replaceMonitorChannels(monitor.id, accountId, body.channel_ids);
     return monitor;
   }, { body: t.Partial(MonitorBody), detail: { summary: "Update monitor", tags: ["monitors"] } })
 

apps/api/src/routes/pings.ts

@@ -2,6 +2,7 @@ import { Elysia, t } from "elysia";
 import sql from "../db";
 import { resolveKey } from "./auth";
 import { extractAuthKey, safeTokenCompare } from "../../../shared/auth";
+import { dispatchForMonitor, type NotificationEvent } from "../notifications";
 
 type SSEController = ReadableStreamDefaultController<Uint8Array>;
 const bus = new Map<string, Set<SSEController>>();  // keyed by accountId
@@ -56,7 +57,7 @@ export const ingest = new Elysia()
     if (!safeTokenCompare(token, process.env.MONITOR_TOKEN)) { set.status = 401; return { error: "Unauthorized" }; }
 
     const [monitor_check] = await sql`
-      SELECT id, account_id, last_state, consecutive_down, resend_interval
+      SELECT id, account_id, name, url, resend_interval, cert_alert_days
       FROM monitors WHERE id = ${body.monitor_id}
     `;
     if (!monitor_check) { set.status = 404; return { error: "Monitor not found" }; }
@@ -71,11 +72,20 @@ export const ingest = new Elysia()
     const scheduledAt = body.scheduled_at ? new Date(body.scheduled_at) : null;
     const jitterMs = body.jitter_ms ?? null;
 
-    // Important-beat + resend bookkeeping. Important = transition or resend tick.
+    // Per-region transition state. Empty string = unspecified/single-region.
+    const region = body.region ?? '';
+    const [stateRow] = await sql`
+      SELECT last_state, consecutive_down, cert_alert_sent
+      FROM monitor_region_state
+      WHERE monitor_id = ${body.monitor_id} AND region = ${region}
+    `;
+
     const newState = body.up ? 'up' : 'down';
-    const prevState: string | null = monitor_check.last_state;
-    let consecutiveDown: number = monitor_check.consecutive_down ?? 0;
+    const prevState: string | null = stateRow?.last_state ?? null;
+    let consecutiveDown: number = stateRow?.consecutive_down ?? 0;
+    let certAlertSent: boolean = stateRow?.cert_alert_sent ?? false;
     const resendInterval: number = monitor_check.resend_interval ?? 0;
+    const certAlertDays: number = monitor_check.cert_alert_days ?? 0;
     let important = false;
 
     if (prevState !== newState) {
@@ -90,10 +100,26 @@ export const ingest = new Elysia()
       consecutiveDown = 0;
     }
 
+    // Cert dedupe flag: flip on threshold cross, clear on renewal. Per-region.
+    let certEvent = false;
+    const days = body.cert_expiry_days;
+    if (days != null && certAlertDays > 0) {
+      if (days <= certAlertDays && !certAlertSent) {
+        certEvent = true;
+        certAlertSent = true;
+      } else if (days > certAlertDays && certAlertSent) {
+        certAlertSent = false;
+      }
+    }
+
     await sql`
-      UPDATE monitors
-      SET last_state = ${newState}, consecutive_down = ${consecutiveDown}
-      WHERE id = ${body.monitor_id}
+      INSERT INTO monitor_region_state (monitor_id, region, last_state, consecutive_down, cert_alert_sent, updated_at)
+      VALUES (${body.monitor_id}, ${region}, ${newState}, ${consecutiveDown}, ${certAlertSent}, now())
+      ON CONFLICT (monitor_id, region) DO UPDATE SET
+        last_state       = EXCLUDED.last_state,
+        consecutive_down = EXCLUDED.consecutive_down,
+        cert_alert_sent  = EXCLUDED.cert_alert_sent,
+        updated_at       = now()
     `;
 
     const [ping] = await sql`
@@ -121,6 +147,33 @@ export const ingest = new Elysia()
 
     publish(monitor_check.account_id, ping);
 
+    // Fire-and-forget notifications. Failures log only and never block ingest.
+    const monitorCtx = {
+      id:     monitor_check.id,
+      name:   monitor_check.name,
+      url:    monitor_check.url,
+      region,
+    };
+    const pingCtx = {
+      status_code: body.status_code ?? null,
+      latency_ms:  body.latency_ms ?? null,
+      error:       body.error ?? null,
+      checked_at:  ping?.checked_at?.toISOString?.() ?? new Date().toISOString(),
+    };
+
+    if (important) {
+      const event: NotificationEvent = body.up
+        ? { kind: "up",   monitor: monitorCtx, ping: pingCtx }
+        : { kind: "down", monitor: monitorCtx, ping: pingCtx };
+      // Suppress the synthetic "up" on a brand-new monitor (no prior state).
+      if (!(body.up && prevState === null)) {
+        void dispatchForMonitor(monitor_check.id, event);
+      }
+    }
+    if (certEvent && days != null) {
+      void dispatchForMonitor(monitor_check.id, { kind: "cert", monitor: monitorCtx, days });
+    }
+
     return { ok: true };
   }, {
     body: t.Object({