diff --git a/apps/workflows/src/checker/incident-utils.ts b/apps/workflows/src/checker/incident-utils.ts new file mode 100644 index 00000000..17ec152e --- /dev/null +++ b/apps/workflows/src/checker/incident-utils.ts @@ -0,0 +1,103 @@ +import { getLogger } from "@logtape/logtape"; +import { and, db, eq, inArray, isNull, schema } from "@openstatus/db"; + +import { checkerAudit } from "../utils/audit-log"; + +const logger = getLogger(["workflow"]); + +/** + * Finds an open incident (not resolved) for the given monitor. + */ +export async function findOpenIncident(monitorId: number) { + return db + .select() + .from(schema.incidentTable) + .where( + and( + eq(schema.incidentTable.monitorId, monitorId), + isNull(schema.incidentTable.resolvedAt), + ), + ) + .get(); +} + +/** + * Finds all open incidents (not resolved) for the given monitor. + */ +export async function findAllOpenIncidents(monitorId: number) { + return db + .select() + .from(schema.incidentTable) + .where( + and( + eq(schema.incidentTable.monitorId, monitorId), + isNull(schema.incidentTable.resolvedAt), + ), + ) + .all(); +} + +/** + * Resolves all open incidents by setting resolvedAt and autoResolved flag. + * Uses a single atomic bulk update to prevent partial state on failures. + * Returns array of successfully resolved incidents. + */ +export async function resolveIncident(params: { + monitorId: string; + cronTimestamp: number; +}): Promise<(typeof schema.incidentTable.$inferSelect)[]> { + const { monitorId, cronTimestamp } = params; + + // Find ALL open incidents for this monitor + const incidents = await findAllOpenIncidents(Number(monitorId)); + + if (incidents.length === 0) { + return []; // No open incidents + } + + // Extract all incident IDs for bulk update + const incidentIds = incidents.map((i) => i.id); + + // ATOMIC BULK UPDATE: Resolve all incidents in a single query + // This prevents partial state if operation fails midway + const resolvedIncidents = await db + .update(schema.incidentTable) + .set({ + resolvedAt: new Date(cronTimestamp), + autoResolved: true, + }) + .where( + and( + inArray(schema.incidentTable.id, incidentIds), + isNull(schema.incidentTable.resolvedAt), // Still prevents race conditions + ), + ) + .returning(); + + // Emit audit logs for each resolved incident + // These are best-effort; failure here doesn't affect data integrity + for (const incident of resolvedIncidents) { + logger.info("Recovered incident", { + incident_id: incident.id, + monitor_id: monitorId, + }); + + try { + await checkerAudit.publishAuditLog({ + id: `monitor:${monitorId}`, + action: "incident.resolved", + targets: [{ id: monitorId, type: "monitor" }], + metadata: { cronTimestamp, incidentId: incident.id }, + }); + } catch (error) { + logger.error("Failed to publish audit log for incident resolution", { + incident_id: incident.id, + monitor_id: monitorId, + error, + }); + // Don't throw - incident is already resolved + } + } + + return resolvedIncidents; +} diff --git a/apps/workflows/src/checker/index.ts b/apps/workflows/src/checker/index.ts index a657189d..7feeac00 100644 --- a/apps/workflows/src/checker/index.ts +++ b/apps/workflows/src/checker/index.ts @@ -1,5 +1,5 @@ import { getLogger } from "@logtape/logtape"; -import { and, db, eq, inArray, isNull, schema } from "@openstatus/db"; +import { and, db, eq, inArray, schema } from "@openstatus/db"; import { incidentTable } from "@openstatus/db/src/schema"; import { monitorRegions } from "@openstatus/db/src/schema/constants"; import { @@ -13,6 +13,7 @@ import { env } from "../env"; import type { Env } from "../index"; import { checkerAudit } from "../utils/audit-log"; import { triggerNotifications, upsertMonitorStatus } from "./alerting"; +import { findOpenIncident, resolveIncident } from "./incident-utils"; import { updateStatusPrivate } from "./private-location"; export const checkerRoute = new Hono(); @@ -31,60 +32,6 @@ const payloadSchema = z.object({ const logger = getLogger(["workflow"]); -/** - * Finds an open incident (not resolved and not acknowledged) for the given monitor. - */ -async function findOpenIncident(monitorId: number) { - return db - .select() - .from(incidentTable) - .where( - and( - eq(incidentTable.monitorId, monitorId), - isNull(incidentTable.resolvedAt), - ), - ) - .get(); -} - -/** - * Resolves an open incident by setting resolvedAt and autoResolved flag. - */ -async function resolveIncident(params: { - monitorId: string; - cronTimestamp: number; -}) { - const { monitorId, cronTimestamp } = params; - const incident = await findOpenIncident(Number(monitorId)); - - if (!incident || incident.resolvedAt) { - return null; - } - - logger.info("Recovering incident", { - incident_id: incident.id, - monitor_id: monitorId, - }); - - await db - .update(incidentTable) - .set({ - resolvedAt: new Date(cronTimestamp), - autoResolved: true, - }) - .where(eq(incidentTable.id, incident.id)) - .run(); - - await checkerAudit.publishAuditLog({ - id: `monitor:${monitorId}`, - action: "incident.resolved", - targets: [{ id: monitorId, type: "monitor" }], - metadata: { cronTimestamp, incidentId: incident.id }, - }); - - return incident; -} - checkerRoute.post("/updateStatus", async (c) => { const auth = c.req.header("Authorization"); if (auth !== `Basic ${env().CRON_SECRET}`) { @@ -233,7 +180,8 @@ checkerRoute.post("/updateStatus", async (c) => { let incident = null; if (monitor.status === "error") { - incident = await resolveIncident({ monitorId, cronTimestamp }); + const incidents = await resolveIncident({ monitorId, cronTimestamp }); + incident = incidents[0] ?? null; } triggeredNotifications = await triggerNotifications({ @@ -266,10 +214,11 @@ checkerRoute.post("/updateStatus", async (c) => { let incident = null; if (monitor.status === "error") { - incident = await resolveIncident({ + const incidents = await resolveIncident({ monitorId, cronTimestamp, }); + incident = incidents[0] ?? null; } triggeredNotifications = await triggerNotifications({ diff --git a/apps/workflows/src/checker/private-location.ts b/apps/workflows/src/checker/private-location.ts index 38a2dc7e..2907b8d6 100644 --- a/apps/workflows/src/checker/private-location.ts +++ b/apps/workflows/src/checker/private-location.ts @@ -1,5 +1,15 @@ import { getLogger } from "@logtape/logtape"; -import { and, db, eq, gte, isNull, lte, schema, sql } from "@openstatus/db"; +import { + and, + db, + eq, + gte, + inArray, + isNull, + lte, + schema, + sql, +} from "@openstatus/db"; import { monitorStatusSchema } from "@openstatus/db/src/schema/monitors/validation"; import type { Context } from "hono"; import { z } from "zod"; @@ -8,6 +18,7 @@ import { env } from "../env"; import type { Env } from "../index"; import { checkerAudit } from "../utils/audit-log"; import { triggerNotifications } from "./alerting"; +import { findOpenIncident, resolveIncident } from "./incident-utils"; const logger = getLogger(["workflow"]); @@ -170,11 +181,123 @@ export async function updateStatusPrivate(c: Context) { const regions = [attachment.name]; + // Check if monitor has cloud regions + const hasCloudRegions = + monitor.regions && monitor.regions.trim().length > 0; + + // Query all private locations for threshold check + const allLocations = await db + .select({ id: schema.privateLocationToMonitors.privateLocationId }) + .from(schema.privateLocationToMonitors) + .where( + and( + eq(schema.privateLocationToMonitors.monitorId, monitorIdNumber), + isNull(schema.privateLocationToMonitors.deletedAt), + ), + ) + .all(); + + const numberOfLocations = allLocations.length; + const locationIds = allLocations + .map((loc) => loc.id) + .filter((id): id is number => id !== null); + + // Count how many locations report this status + const locationsWithStatus = await db + .select({ + privateLocationId: + schema.privateLocationMonitorStatus.privateLocationId, + }) + .from(schema.privateLocationMonitorStatus) + .where( + and( + eq(schema.privateLocationMonitorStatus.monitorId, monitorIdNumber), + eq(schema.privateLocationMonitorStatus.status, status), + inArray( + schema.privateLocationMonitorStatus.privateLocationId, + locationIds, + ), + ), + ) + .all(); + + const affectedLocationCount = locationsWithStatus.length; + + // Apply ≥50% threshold (matching cloud checker logic) + const shouldTriggerIncident = + !hasCloudRegions && + (affectedLocationCount >= numberOfLocations / 2 || + numberOfLocations === 1); + + let incident = null; let triggeredNotifications: { notificationId: number; provider: string }[] = []; switch (status) { case "error": + // Create incident only if private-only monitor AND threshold met + if (shouldTriggerIncident) { + try { + const existingIncident = await findOpenIncident(monitorIdNumber); + if (!existingIncident) { + const [newIncident] = await db + .insert(schema.incidentTable) + .values({ + monitorId: monitorIdNumber, + workspaceId: monitor.workspaceId, + startedAt: new Date(cronTimestamp), + }) + .returning(); + + if (newIncident?.id) { + incident = newIncident; + await checkerAudit.publishAuditLog({ + id: `monitor:${monitorId}`, + action: "incident.created", + targets: [{ id: monitorId, type: "monitor" }], + metadata: { cronTimestamp, incidentId: newIncident.id }, + }); + logger.info("Created incident", { + incident_id: newIncident.id, + monitor_id: monitorId, + affected_location_count: affectedLocationCount, + total_locations: numberOfLocations, + }); + } + } else { + incident = existingIncident; + logger.info("Already in incident", { + incident_id: existingIncident.id, + }); + } + } catch (error) { + // Check if this is a constraint violation (race condition) + const errorMessage = + error instanceof Error ? error.message : String(error); + if ( + errorMessage.includes("UNIQUE constraint") || + errorMessage.includes("unique") + ) { + // Another request created the incident concurrently, fetch it + logger.info( + "Concurrent incident creation detected, fetching existing", + { + monitor_id: monitorId, + }, + ); + const existingIncident = await findOpenIncident(monitorIdNumber); + if (existingIncident) { + incident = existingIncident; + } + } else { + logger.error("Failed to create incident", { + monitor_id: monitorId, + error_message: errorMessage, + }); + } + } + } + await checkerAudit.publishAuditLog({ id: `monitor:${monitorId}`, action: "monitor.failed", @@ -195,9 +318,31 @@ export async function updateStatusPrivate(c: Context) { cronTimestamp, regions, latency, + incidentId: incident?.id, }); break; case "degraded": + // Resolve incident if private-only monitor AND threshold met + if (shouldTriggerIncident) { + try { + const incidents = await resolveIncident({ + monitorId, + cronTimestamp, + }); + incident = incidents[0] ?? null; + } catch (error) { + logger.warning( + "Failed to resolve incident on degraded transition", + { + monitor_id: monitorId, + error_message: + error instanceof Error ? error.message : String(error), + }, + ); + // Continue with notifications even if resolution fails + } + } + await checkerAudit.publishAuditLog({ id: `monitor:${monitorId}`, action: "monitor.degraded", @@ -217,9 +362,28 @@ export async function updateStatusPrivate(c: Context) { cronTimestamp, regions, latency, + incidentId: incident?.id, }); break; case "active": + // Resolve incident if private-only monitor AND threshold met + if (shouldTriggerIncident) { + try { + const incidents = await resolveIncident({ + monitorId, + cronTimestamp, + }); + incident = incidents[0] ?? null; + } catch (error) { + logger.warning("Failed to resolve incident on active transition", { + monitor_id: monitorId, + error_message: + error instanceof Error ? error.message : String(error), + }); + // Continue with notifications even if resolution fails + } + } + await checkerAudit.publishAuditLog({ id: `monitor:${monitorId}`, action: "monitor.recovered", @@ -239,6 +403,7 @@ export async function updateStatusPrivate(c: Context) { cronTimestamp, regions, latency, + incidentId: incident?.id, }); break; }