Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/openstatus. 🫖 Status page with uptime monitoring & API monitoring as code 🫖 openstatus.dev
bun drizzle-orm monitoring monitoring-as-code nextjs observability on-call open-source shadcn-ui status-page statuspage synthetic-monitoring tinybird turso uptime uptime-checker uptime-monitor
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385import { getLogger } from "@logtape/logtape";import { type SQL, and, db, eq, sql } from "@openstatus/db";import type { NotificationOutboxPayload, MonitorStatus, NotificationProvider,} from "@openstatus/db/src/schema";import { monitorTransition, notificationOutbox, notificationOutboxEventType, monitorIncidentTable, monitor, monitorStatusTable, notification, notificationsToMonitors, selectMonitorSchema,} from "@openstatus/db/src/schema";import { withBusyRetry } from "@openstatus/services";
import { checkerAudit } from "../utils/audit-log";
const logger = getLogger(["workflow"]);import { quorumCountSql, quorumGuardSql } from "./quorum";
export type EventType = (typeof notificationOutboxEventType)[number];
export const EVENT_TYPE: Record<MonitorStatus, EventType> = { active: "recovery", degraded: "degraded", error: "alert",};
export type TransitionInput = { monitorId: number; region: string; status: MonitorStatus; cronTimestamp: number; statusCode?: number; message?: string; latency?: number; deadlineSeconds: number; /** Monitors outside this gate are still delivered by the inline sender. */ rolloutPct: number;};
export type OutboxRowRef = { id: number; notificationId: number; provider: NotificationProvider; /** `pending` means the drainer owns delivery; `settled` means the inline sender does. */ deliveryStatus: "pending" | "settled";};
export type TransitionResult = | { kind: "unchanged" } | { kind: "monitor-missing" } | { kind: "evaluated"; transitioned: boolean; quorumCount: number; regionCount: number; affectedRegions: string[]; outboxRows: OutboxRowRef[]; incidentId: number | null; incidentCreatedId: number | null; incidentResolvedIds: number[]; };
/** * A Cloud Tasks retry can land after a later check already reported. The stored * `cron_timestamp` only advances on a status change, so it cannot bound this on * its own. */export function isStaleCheck( cronTimestamp: number, maxAgeMs: number, now = Date.now(),): boolean { return now - cronTimestamp > maxAgeMs;}
type JournalRow = { quorum_count: number; transitioned: number };type OutboxInsertRow = { id: number; notification_id: number; incident_id: number | null; provider: NotificationProvider; delivery_status: "pending" | "settled";};
function journalStatement( input: TransitionInput, regionsJson: string, regionCount: number, guard: SQL,) { const count = quorumCountSql({ toStatus: input.status, regionsJson, }); const transitioning = sql`${guard} AND ${monitor.status} <> ${input.status}`;
return db.all<JournalRow>(sql` INSERT INTO ${monitorTransition} (monitor_id, region, cron_timestamp, from_status, to_status, quorum_count, region_count, transitioned, outbox_rows, created_at) SELECT ${monitor.id}, ${input.region}, ${input.cronTimestamp}, ${monitor.status}, ${input.status}, ${count}, ${regionCount}, CASE WHEN ${transitioning} THEN 1 ELSE 0 END, CASE WHEN ${transitioning} THEN (SELECT count(*) FROM ${notificationsToMonitors} WHERE ${notificationsToMonitors.monitorId} = ${monitor.id}) ELSE 0 END, unixepoch() FROM ${monitor} WHERE ${monitor.id} = ${input.monitorId} RETURNING quorum_count, transitioned `);}
function createIncidentStatement(input: TransitionInput, guard: SQL) { return db.all<{ id: number }>(sql` INSERT INTO ${monitorIncidentTable} (monitor_id, workspace_id, started_at) SELECT ${monitor.id}, ${monitor.workspaceId}, ${Math.floor(input.cronTimestamp / 1000)} FROM ${monitor} WHERE ${monitor.id} = ${input.monitorId} AND ${monitor.status} <> ${input.status} AND ${guard} AND NOT EXISTS ( SELECT 1 FROM ${monitorIncidentTable} WHERE ${monitorIncidentTable.monitorId} = ${monitor.id} AND ${monitorIncidentTable.resolvedAt} IS NULL) ON CONFLICT DO NOTHING RETURNING id `);}
function resolveIncidentStatement(input: TransitionInput, guard: SQL) { return db.all<{ id: number }>(sql` UPDATE ${monitorIncidentTable} SET resolved_at = ${Math.floor(input.cronTimestamp / 1000)}, auto_resolved = 1 WHERE ${monitorIncidentTable.monitorId} = ${input.monitorId} AND ${monitorIncidentTable.resolvedAt} IS NULL AND EXISTS ( SELECT 1 FROM ${monitor} WHERE ${monitor.id} = ${input.monitorId} AND ${monitor.status} <> ${input.status} AND ${guard}) RETURNING id `);}
function outboxStatement( input: TransitionInput, payload: NotificationOutboxPayload, guard: SQL,) { const dedupPrefix = `${input.cronTimestamp}:${input.monitorId}:${input.status}:`; // Outside the rollout the inline sender owns this delivery, so the row is // written already consumed: it still feeds the shadow diff, but raising // OUTBOX_ROLLOUT_PCT can never make the drainer re-send it. const owned = sql`(${monitor.id} % 100) < ${input.rolloutPct}`;
return db.all<OutboxInsertRow>(sql` INSERT INTO ${notificationOutbox} (dedup_key, monitor_id, workspace_id, notification_id, provider, event_type, from_status, to_status, cron_timestamp, incident_id, payload, delivery_status, outcome, next_attempt_at, deadline_at, created_at) SELECT ${dedupPrefix} || ${notification.id}, ${monitor.id}, ${monitor.workspaceId}, ${notification.id}, ${notification.provider}, ${EVENT_TYPE[input.status]}, ${monitor.status}, ${input.status}, ${input.cronTimestamp}, (SELECT id FROM ${monitorIncidentTable} WHERE ${monitorIncidentTable.monitorId} = ${monitor.id} AND ${monitorIncidentTable.resolvedAt} IS NULL ORDER BY id DESC LIMIT 1), ${JSON.stringify(payload)}, CASE WHEN ${owned} THEN 'pending' ELSE 'settled' END, CASE WHEN ${owned} THEN NULL ELSE 'inline' END, unixepoch(), unixepoch() + ${input.deadlineSeconds}, unixepoch() FROM ${monitor} JOIN ${notificationsToMonitors} ON ${notificationsToMonitors.monitorId} = ${monitor.id} JOIN ${notification} ON ${notification.id} = ${notificationsToMonitors.notificationId} WHERE ${monitor.id} = ${input.monitorId} AND ${monitor.status} <> ${input.status} AND ${guard} ON CONFLICT (dedup_key) DO NOTHING RETURNING id, notification_id, incident_id, provider, delivery_status `);}
function casStatement(input: TransitionInput, guard: SQL) { return db.all<{ id: number }>(sql` UPDATE ${monitor} SET status = ${input.status}, updated_at = unixepoch() WHERE ${monitor.id} = ${input.monitorId} AND ${monitor.status} <> ${input.status} AND ${guard} RETURNING id `);}
/** * Writes the region status and, only when that changed something, evaluates the * monitor transition as one atomic batch. An outbox row exists iff the * compare-and-swap matched, so a notification is owed exactly once. */export async function applyStatusTransition( input: TransitionInput,): Promise<TransitionResult> { const changed = await withBusyRetry(() => db.all<{ region: string }>(sql` INSERT INTO ${monitorStatusTable} (monitor_id, region, status, cron_timestamp, updated_at) VALUES (${input.monitorId}, ${input.region}, ${input.status}, ${input.cronTimestamp}, unixepoch()) ON CONFLICT (monitor_id, region) DO UPDATE SET status = excluded.status, cron_timestamp = excluded.cron_timestamp, updated_at = unixepoch() WHERE ${monitorStatusTable.status} <> excluded.status AND excluded.cron_timestamp > ${monitorStatusTable.cronTimestamp} RETURNING region `), );
if (changed.length === 0) return { kind: "unchanged" };
return evaluateTransition(input);}
/** * The transition half, without the region write. Exported as the repair entry * point: the region write and this evaluation are separate transactions, so a * crash or a failed batch between them leaves `monitor.status` behind, and the * fast path means replaying the same check will not re-evaluate it. */export async function evaluateTransition( input: TransitionInput,): Promise<TransitionResult> { const [monitorRows, statusRows] = await withBusyRetry(() => db.batch([ db.select().from(monitor).where(eq(monitor.id, input.monitorId)), db .select({ region: monitorStatusTable.region }) .from(monitorStatusTable) .where( and( eq(monitorStatusTable.monitorId, input.monitorId), eq(monitorStatusTable.status, input.status), ), ), ]), );
const parsed = selectMonitorSchema.safeParse(monitorRows[0]); if (!parsed.success) return { kind: "monitor-missing" };
const regions = parsed.data.regions; const regionsJson = JSON.stringify(regions); const regionCount = regions.length; const configuredRegions = new Set<string>(regions); const affectedRegions = statusRows .map((row) => row.region) .filter((region) => configuredRegions.has(region));
const payload: NotificationOutboxPayload = { regions: affectedRegions, statusCode: input.statusCode, message: input.message, latency: input.latency, };
const guard = quorumGuardSql({ toStatus: input.status, regionsJson, regionCount, }); const journal = journalStatement(input, regionsJson, regionCount, guard); const outbox = outboxStatement(input, payload, guard); const cas = casStatement(input, guard);
let journalRows: JournalRow[]; let outboxRows: OutboxInsertRow[]; let incidentRows: { id: number }[] = [];
// All three statuses enqueue notifications; only the incident statement moves. // The outbox row reads incident_id by subquery, so it must run after the // incident is created but before an existing one is resolved. if (input.status === "error") { const [journalResult, incidentResult, outboxResult] = await withBusyRetry( () => db.batch([journal, createIncidentStatement(input, guard), outbox, cas]), ); journalRows = journalResult; outboxRows = outboxResult; incidentRows = incidentResult; } else { const [journalResult, outboxResult, incidentResult] = await withBusyRetry( () => db.batch([ journal, outbox, resolveIncidentStatement(input, guard), cas, ]), ); journalRows = journalResult; outboxRows = outboxResult; incidentRows = incidentResult; }
const incidentCreatedId = input.status === "error" ? (incidentRows[0]?.id ?? null) : null; const incidentResolvedIds = input.status === "error" ? [] : incidentRows.map((row) => row.id);
// Published here rather than in the route so drift repair keeps the trail too. await publishIncidentAudit(input, incidentCreatedId, incidentResolvedIds);
return { kind: "evaluated", transitioned: (journalRows[0]?.transitioned ?? 0) === 1, quorumCount: journalRows[0]?.quorum_count ?? 0, regionCount, affectedRegions, outboxRows: outboxRows.map((row) => ({ id: row.id, notificationId: row.notification_id, provider: row.provider, deliveryStatus: row.delivery_status, })), incidentId: outboxRows[0]?.incident_id ?? null, incidentCreatedId, incidentResolvedIds, };}
async function publishIncidentAudit( input: TransitionInput, createdId: number | null, resolvedIds: number[],): Promise<void> { const targets = [{ id: String(input.monitorId), type: "monitor" as const }]; const entries: Promise<unknown>[] = [];
if (createdId !== null) { entries.push( checkerAudit.publishAuditLog({ id: `monitor:${input.monitorId}`, action: "incident.created", targets, metadata: { cronTimestamp: input.cronTimestamp, incidentId: createdId }, }), ); }
for (const incidentId of resolvedIds) { entries.push( checkerAudit.publishAuditLog({ id: `monitor:${input.monitorId}`, action: "incident.resolved", targets, metadata: { cronTimestamp: input.cronTimestamp, incidentId }, }), ); }
// Best-effort: the incident is already committed, and failing here would make // Cloud Tasks retry a transition that has landed. try { await Promise.all(entries); } catch (error) { logger.warn("Failed to publish incident audit log", { monitor_id: input.monitorId, error_message: error instanceof Error ? error.message : String(error), }); }}