fix: improve the logic for reduce write row d1

This commit is contained in:
2026-09-02 20:13:00 +07:00
parent c525e8e3a2
commit 6d8c40b37d
36 changed files with 2310 additions and 194 deletions
+1
View File
@@ -0,0 +1 @@
DROP INDEX `monitors_enabled_last_checked_at_idx`;
File diff suppressed because it is too large Load Diff
+7
View File
@@ -85,6 +85,13 @@
"when": 1788187390249,
"tag": "0011_hard_mystique",
"breakpoints": true
},
{
"idx": 12,
"version": "6",
"when": 1788348540256,
"tag": "0012_free_tier_scale",
"breakpoints": true
}
]
}
-1
View File
@@ -60,7 +60,6 @@ export type CheckResult = {
export type Check = {
id: number;
monitorId: number;
ok: boolean;
degraded: boolean;
statusCode: number | null;
+1 -1
View File
@@ -73,7 +73,7 @@ export function StatusHistoryBar({ history }: { history: HistoryEntry[] }) {
startDay === endDay
? dayTitle(startDay, uptimePct)
: `${formatDay(startDay)} ${formatDay(endDay)}: ${
uptimePct === undefined ? 'No data' : `${uptimePct.toFixed(1)}% average uptime`
uptimePct == null ? 'No data' : `${uptimePct.toFixed(1)}% average uptime`
}`
}
/>
@@ -13,7 +13,7 @@ export const DEFAULT_MONITOR_INPUT: MonitorInput = {
url: 'https://',
method: 'GET',
expectedStatus: 200,
intervalSeconds: 300,
intervalSeconds: 900,
timeoutMs: 10_000,
retryCount: 1,
failureThreshold: 2,
@@ -28,6 +28,7 @@ export const DEFAULT_MONITOR_INPUT: MonitorInput = {
export const INTERVAL_OPTIONS = [
{ value: '300', label: '5 minutes' },
{ value: '600', label: '10 minutes' },
{ value: '900', label: '15 minutes' },
{ value: '1800', label: '30 minutes' },
{ value: '3600', label: '1 hour' },
@@ -135,7 +136,7 @@ export function MonitorFormDialog({ editing, onClose }: MonitorFormDialogProps)
<p className="overline">Configuration</p>
<DialogTitle>{editing ? `Edit ${editing.name}` : 'Add a monitor'}</DialogTitle>
<DialogDescription>
Checks run at least every five minutes. Failures are retried immediately and must repeat before an incident is published.
Checks run on the selected interval. Failures are retried immediately and must repeat before an incident is published.
</DialogDescription>
</DialogHeader>
<form className="monitor-form" onSubmit={handleSubmit}>
@@ -275,7 +276,7 @@ export function MonitorFormDialog({ editing, onClose }: MonitorFormDialogProps)
maxLength={200}
placeholder="healthy"
/>
<small>Case-insensitive match within the first 256 KB of the response.</small>
<small>Case-insensitive match within the first 64 KB of the response.</small>
</label>
<div className="toggle-field advanced-inverted">
<Switch
+1
View File
@@ -18,6 +18,7 @@ export const queryClient: QueryClient = new QueryClient({
defaultOptions: {
queries: {
retry: 1,
staleTime: 30_000,
refetchOnWindowFocus: false,
},
},
+2 -2
View File
@@ -15,7 +15,7 @@ function upsertMeta(selector: string, attributes: Record<string, string>, conten
if (!element) {
element = document.createElement('meta');
for (const [name, value] of Object.entries(attributes)) element.setAttribute(name, value);
document.head.append(element);
document.head.appendChild(element);
}
element.setAttribute('content', content);
element.setAttribute(managedAttribute, 'true');
@@ -53,7 +53,7 @@ export function useSeo({ title, description, noindex = false, canonicalPath }: S
if (!canonical) {
canonical = document.createElement('link');
canonical.rel = 'canonical';
document.head.append(canonical);
document.head.appendChild(canonical);
}
canonical.href = absoluteUrl(canonicalPath);
canonical.setAttribute(managedAttribute, 'true');
+6 -1
View File
@@ -15,7 +15,12 @@ export const channelKeys = {
deliveries: (id: number) => ['channels', id, 'deliveries'] as const,
};
export function useNotificationChannelsQuery() {
return useQuery({ queryKey: channelKeys.all, queryFn: ({ signal }) => getNotificationChannels(signal), refetchInterval: 30_000 });
return useQuery({
queryKey: channelKeys.all,
queryFn: ({ signal }) => getNotificationChannels(signal),
refetchInterval: 120_000,
refetchIntervalInBackground: false,
});
}
function invalidateChannels() {
return queryClient.invalidateQueries({ queryKey: channelKeys.all });
+6 -1
View File
@@ -21,7 +21,12 @@ export const incidentKeys = {
detail: (id: number) => [...incidentKeys.all, 'detail', id] as const,
};
export function useIncidentsQuery(status: 'open' | 'all' = 'open') {
return useQuery({ queryKey: incidentKeys.list(status), queryFn: ({ signal }) => listIncidents(status, signal), refetchInterval: 60_000 });
return useQuery({
queryKey: incidentKeys.list(status),
queryFn: ({ signal }) => listIncidents(status, signal),
refetchInterval: 120_000,
refetchIntervalInBackground: false,
});
}
export function useIncidentQuery(id: number) {
return useQuery({ queryKey: incidentKeys.detail(id), queryFn: ({ signal }) => getIncident(id, signal) });
+1 -1
View File
@@ -14,7 +14,7 @@ export function useMaintenanceWindowsQuery() {
return useQuery({
queryKey: maintenanceKeys.all,
queryFn: ({ signal }) => getMaintenanceWindows(signal),
refetchInterval: 60_000,
refetchInterval: 300_000,
refetchIntervalInBackground: false,
});
}
+6 -4
View File
@@ -26,7 +26,7 @@ export const monitorsQueryOptions = () =>
queryOptions({
queryKey: monitorKeys.list(),
queryFn: ({ signal }) => listMonitors(signal),
refetchInterval: 60_000,
refetchInterval: 120_000,
refetchIntervalInBackground: false,
});
@@ -35,7 +35,7 @@ export function useMonitorsQuery() {
}
const liveQueryDefaults = {
refetchInterval: 60_000,
refetchInterval: 120_000,
refetchIntervalInBackground: false,
} as const;
@@ -62,7 +62,8 @@ export function useMonitorStatsQuery(id: number) {
queryKey: monitorKeys.stats(id),
queryFn: ({ signal }) => getMonitorStats(id, signal),
enabled: Number.isSafeInteger(id) && id > 0,
...liveQueryDefaults,
refetchInterval: 300_000,
refetchIntervalInBackground: false,
});
}
@@ -71,7 +72,8 @@ export function useMonitorIncidentsQuery(id: number) {
queryKey: monitorKeys.incidents(id),
queryFn: ({ signal }) => listIncidents(id, 50, signal),
enabled: Number.isSafeInteger(id) && id > 0,
...liveQueryDefaults,
refetchInterval: 300_000,
refetchIntervalInBackground: false,
});
}
+2 -1
View File
@@ -20,7 +20,8 @@ export function useIncidentHistoryQuery() {
return useQuery({
queryKey: statusKeys.history,
queryFn: ({ signal }) => getIncidentHistory(signal),
refetchInterval: 60_000,
refetchInterval: 120_000,
refetchIntervalInBackground: false,
});
}
+39
View File
@@ -53,3 +53,42 @@ export async function recordAiEvent(env: Env, input: AiEventInput): Promise<void
console.warn(JSON.stringify({ message: 'AI event recording failed', error: error instanceof Error ? error.message : String(error) }));
}
}
export async function recordAiEvents(env: Env, inputs: AiEventInput[]): Promise<void> {
if (inputs.length === 0) return;
try {
await env.DB.prepare(
`INSERT INTO ai_events
(kind, incident_id, monitor_id, model, outcome, reason, latency_ms, prompt_tokens, completion_tokens,
context_preview, output_preview, created_at)
SELECT json_extract(value, '$.kind'), json_extract(value, '$.incidentId'), json_extract(value, '$.monitorId'),
json_extract(value, '$.model'), json_extract(value, '$.outcome'), json_extract(value, '$.reason'),
json_extract(value, '$.latencyMs'), json_extract(value, '$.promptTokens'), json_extract(value, '$.completionTokens'),
json_extract(value, '$.contextPreview'), json_extract(value, '$.outputPreview'), ?2
FROM json_each(?1)`,
)
.bind(
JSON.stringify(
inputs.map((input) => ({
kind: input.kind,
incidentId: input.incidentId ?? null,
monitorId: input.monitorId ?? null,
model: input.model ?? null,
outcome: input.outcome,
reason: preview(input.reason)?.slice(0, 200) ?? null,
latencyMs: input.latencyMs ?? null,
promptTokens: input.promptTokens ?? null,
completionTokens: input.completionTokens ?? null,
contextPreview: preview(input.contextPreview),
outputPreview: preview(input.outputPreview),
})),
),
Date.now(),
)
.run();
} catch (error) {
console.warn(
JSON.stringify({ message: 'AI event batch recording failed', error: error instanceof Error ? error.message : String(error) }),
);
}
}
+33 -15
View File
@@ -2,7 +2,7 @@ import { and, eq, isNull, sql } from 'drizzle-orm';
import { requestCompletionDetailed, type CompletionResult } from '../ai/client';
import { DEGRADED_RECOVERY_UPDATE_BODY, RECOVERY_UPDATE_BODY } from '../ai/fallback-message';
import { buildIncidentContext } from '../ai/incident-context';
import { recordAiEvent, type AiEventKind } from '../ai/events';
import { recordAiEvent, recordAiEvents, type AiEventInput, type AiEventKind } from '../ai/events';
import {
AUTOPILOT_STATUS_GUIDANCE,
INCIDENT_FOLLOWUP_SYSTEM_PROMPT,
@@ -21,7 +21,7 @@ import { loadIncidentSignal, findLatestAutoIncidentForMonitor } from './signal';
// Per-pass ceilings come from resolveRunLimits(env): AI_CALLS_PER_RUN / AI_FOLLOWUP_CALLS_PER_RUN,
// defaulting to DEFAULT_RUN_LIMITS. They keep one autopilot pass within the free-plan subrequest budget.
export const AUTOPILOT_CONCURRENCY = 4;
export const AUTOPILOT_DEADLINE_MS = 45_000;
export const AUTOPILOT_DEADLINE_MS = 30_000;
export type AiBudget = { remaining: number; deadline?: number };
export type AutopilotEvent = {
@@ -35,6 +35,7 @@ export type AutopilotSummary = { calls: number; written: number; rejected: numbe
type Settings = typeof aiSettings.$inferSelect;
type Task = { kind: AiEventKind; incidentId: number; monitorId: number; run: () => Promise<boolean> };
type SignalLoader = (incidentId: number) => ReturnType<typeof loadIncidentSignal>;
function completionEvent(result: CompletionResult) {
return {
@@ -77,11 +78,18 @@ async function writeCasUpdate(
return Number(results[0].meta.changes ?? 0) === 1 && Number(results[1].meta.changes ?? 0) === 1;
}
async function processOpening(env: Env, settings: Settings, event: AutopilotEvent, incidentId: number, kind: 'down' | 'degraded') {
async function processOpening(
env: Env,
settings: Settings,
event: AutopilotEvent,
incidentId: number,
kind: 'down' | 'degraded',
loadSignal: SignalLoader,
) {
const db = getDb(env);
const [incident] = await db.select().from(incidents).where(eq(incidents.id, incidentId)).limit(1);
if (!incident || incident.resolvedAt || incident.source !== 'auto') return false;
const signal = await loadIncidentSignal(db, incident.id);
const signal = await loadSignal(incident.id);
if (!signal) return false;
const context = await buildIncidentContext(db, event.monitor, event.result);
const result = await complete(settings, INCIDENT_OPEN_SYSTEM_PROMPT, context, 220);
@@ -183,9 +191,8 @@ async function processResolution(env: Env, settings: Settings, event: AutopilotE
return wrote;
}
async function processFollowup(env: Env, settings: Settings, incident: typeof incidents.$inferSelect) {
const db = getDb(env);
const signal = await loadIncidentSignal(db, incident.id);
async function processFollowup(env: Env, settings: Settings, incident: typeof incidents.$inferSelect, loadSignal: SignalLoader) {
const signal = await loadSignal(incident.id);
if (!signal) return false;
const status = settings.autopilotAdvanceStatus
? advanceStatus(incident.status as AutopilotIncidentStatus, signal)
@@ -214,7 +221,7 @@ async function processFollowup(env: Env, settings: Settings, incident: typeof in
return wrote;
}
async function loadFollowupTasks(env: Env, settings: Settings, excluded: Set<number>): Promise<Task[]> {
async function loadFollowupTasks(env: Env, settings: Settings, excluded: Set<number>, loadSignal: SignalLoader): Promise<Task[]> {
const db = getDb(env);
const maxFollowups = resolveRunLimits(env).aiFollowupCallsPerRun;
const rows = await db
@@ -256,15 +263,15 @@ async function loadFollowupTasks(env: Env, settings: Settings, excluded: Set<num
}
const count = Number(row.autoUpdateCount);
if (count >= settings.autopilotMaxUpdates) continue;
const last = row.lastUpdateAt instanceof Date ? row.lastUpdateAt.getTime() : Number(row.lastUpdateAt);
const last = Number(row.lastUpdateAt);
if (Date.now() < nextFollowupDueAt(last, count, settings.autopilotFollowupMinutes)) continue;
const signal = await loadIncidentSignal(db, row.incident.id);
const signal = await loadSignal(row.incident.id);
if (!signal?.monitor.alertsEnabled) continue;
tasks.push({
kind: 'incident_followup',
incidentId: row.incident.id,
monitorId: signal.monitor.id,
run: () => processFollowup(env, settings, row.incident),
run: () => processFollowup(env, settings, row.incident, loadSignal),
});
if (tasks.length >= maxFollowups) break;
}
@@ -281,6 +288,15 @@ export async function runAutopilot(
if (!settings?.enabled || !settings.autopilotEnabled || !settings.baseUrl || !settings.apiKey || !settings.model) return summary;
const budget = input.budget ?? { remaining: resolveRunLimits(env).aiCallsPerRun, deadline: Date.now() + AUTOPILOT_DEADLINE_MS };
budget.deadline ??= Date.now() + AUTOPILOT_DEADLINE_MS;
const signalCache = new Map<number, ReturnType<typeof loadIncidentSignal>>();
const loadSignal: SignalLoader = (incidentId) => {
let signal = signalCache.get(incidentId);
if (!signal) {
signal = loadIncidentSignal(db, incidentId);
signalCache.set(incidentId, signal);
}
return signal;
};
const tasks: Task[] = [];
const excluded = new Set<number>();
for (const event of input.events ?? []) {
@@ -293,7 +309,7 @@ export async function runAutopilot(
kind: 'incident_open',
incidentId: incident.id,
monitorId: event.monitor.id,
run: () => processOpening(env, settings, event, incident.id, 'down'),
run: () => processOpening(env, settings, event, incident.id, 'down', loadSignal),
});
}
}
@@ -316,7 +332,7 @@ export async function runAutopilot(
kind: 'degraded_open',
incidentId: incident.id,
monitorId: event.monitor.id,
run: () => processOpening(env, settings, event, incident.id, 'degraded'),
run: () => processOpening(env, settings, event, incident.id, 'degraded', loadSignal),
});
}
}
@@ -332,7 +348,7 @@ export async function runAutopilot(
}
}
}
if (!input.skipSweep) tasks.push(...(await loadFollowupTasks(env, settings, excluded)));
if (!input.skipSweep && budget.remaining > tasks.length) tasks.push(...(await loadFollowupTasks(env, settings, excluded, loadSignal)));
const priority: Record<AiEventKind, number> = {
incident_open: 0,
incident_resolve: 1,
@@ -343,13 +359,14 @@ export async function runAutopilot(
};
tasks.sort((left, right) => priority[left.kind] - priority[right.kind]);
const skippedEvents: AiEventInput[] = [];
for (let offset = 0; offset < tasks.length; offset += AUTOPILOT_CONCURRENCY) {
const batch = tasks.slice(offset, offset + AUTOPILOT_CONCURRENCY);
await Promise.all(
batch.map(async (task) => {
if (budget.remaining <= 0 || Date.now() >= budget.deadline!) {
summary.skipped += 1;
await recordAiEvent(env, {
skippedEvents.push({
kind: task.kind,
incidentId: task.incidentId,
monitorId: task.monitorId,
@@ -377,5 +394,6 @@ export async function runAutopilot(
}),
);
}
await recordAiEvents(env, skippedEvents);
return summary;
}
+241 -11
View File
@@ -10,6 +10,81 @@ export type AlertTransition = Extract<CheckTransition, 'opened' | 'resolved'>;
export type LatencyTransition = 'degraded' | 'recovered' | null;
type BatchStatement = Parameters<Database['batch']>[0][number];
type ResultState = {
transition: CheckTransition;
latencyTransition: LatencyTransition;
consecutiveFailures: number;
monitorUpdate: {
id: number;
lastOk?: boolean;
consecutiveFailures: number;
consecutiveSlow: number;
lastDegraded: boolean;
lastStatusCode: number | null;
lastLatencyMs: number;
lastError: string | null;
lastCheckedAt: number;
updatedAt: number;
};
};
export type ScheduledResult = {
monitor: Monitor;
result: CheckResult;
checkedAt: Date;
maintenance: boolean;
transition: CheckTransition;
latencyTransition: LatencyTransition;
consecutiveFailures: number;
};
function computeResultState(monitor: Monitor, result: CheckResult, checkedAt: Date, maintenance: boolean): ResultState {
const baseUpdate = {
id: monitor.id,
consecutiveFailures: monitor.consecutiveFailures,
consecutiveSlow: monitor.consecutiveSlow,
lastDegraded: monitor.lastDegraded,
lastStatusCode: result.statusCode,
lastLatencyMs: result.latencyMs,
lastError: result.error,
lastCheckedAt: checkedAt.getTime(),
updatedAt: checkedAt.getTime(),
};
if (maintenance) {
return {
transition: null,
latencyTransition: null,
consecutiveFailures: monitor.consecutiveFailures,
monitorUpdate: baseUpdate,
};
}
const threshold = Math.max(1, monitor.failureThreshold);
const previousFailures = monitor.consecutiveFailures;
const nextFailures = result.ok ? 0 : previousFailures + 1;
const nextSlow = result.degraded ? monitor.consecutiveSlow + 1 : 0;
const wasDown = monitor.lastOk === false;
const isDown = !result.ok && nextFailures >= threshold;
const confirmed = result.ok ? true : isDown ? false : undefined;
const confirmedDegraded = nextSlow >= threshold;
const latencyTransition: LatencyTransition =
!monitor.lastDegraded && confirmedDegraded ? 'degraded' : monitor.lastDegraded && !confirmedDegraded ? 'recovered' : null;
const transition: CheckTransition =
!wasDown && isDown ? 'opened' : wasDown && result.ok ? 'resolved' : !result.ok ? 'pending' : previousFailures > 0 ? 'cleared' : null;
return {
transition,
latencyTransition,
consecutiveFailures: nextFailures,
monitorUpdate: {
...baseUpdate,
...(confirmed === undefined ? {} : { lastOk: confirmed }),
consecutiveFailures: nextFailures,
consecutiveSlow: nextSlow,
lastDegraded: confirmedDegraded,
},
};
}
export function buildResultStatements(
db: Database,
@@ -19,6 +94,7 @@ export function buildResultStatements(
maintenance = false,
options: { degradedIncidents?: boolean } = {},
) {
const state = computeResultState(monitor, result, checkedAt, maintenance);
const statements: BatchStatement[] = [
db.insert(checks).values({
monitorId: monitor.id,
@@ -47,24 +123,22 @@ export function buildResultStatements(
);
return {
statements,
transition: null as CheckTransition,
latencyTransition: null as LatencyTransition,
consecutiveFailures: monitor.consecutiveFailures,
transition: state.transition,
latencyTransition: state.latencyTransition,
consecutiveFailures: state.consecutiveFailures,
};
}
const threshold = Math.max(1, monitor.failureThreshold);
const previousFailures = monitor.consecutiveFailures;
// This deliberately uses the monitor snapshot. Concurrent manual and scheduled checks may lose one increment,
// which delays confirmation by one check but cannot publish a false incident.
const nextFailures = result.ok ? 0 : previousFailures + 1;
const nextSlow = result.degraded ? monitor.consecutiveSlow + 1 : 0;
const nextFailures = state.monitorUpdate.consecutiveFailures;
const nextSlow = state.monitorUpdate.consecutiveSlow;
const wasDown = monitor.lastOk === false;
const isDown = !result.ok && nextFailures >= threshold;
const confirmed = result.ok ? true : isDown ? false : undefined;
const confirmedDegraded = nextSlow >= threshold;
const latencyTransition: LatencyTransition =
!monitor.lastDegraded && confirmedDegraded ? 'degraded' : monitor.lastDegraded && !confirmedDegraded ? 'recovered' : null;
const confirmed = state.monitorUpdate.lastOk;
const confirmedDegraded = state.monitorUpdate.lastDegraded;
const latencyTransition = state.latencyTransition;
statements.push(
db
@@ -196,7 +270,7 @@ export function buildResultStatements(
);
transition = 'resolved';
} else if (!wasDown && !result.ok) transition = 'pending';
else if (!wasDown && result.ok && previousFailures > 0) transition = 'cleared';
else if (!wasDown && result.ok && monitor.consecutiveFailures > 0) transition = 'cleared';
if (options.degradedIncidents && monitor.alertsEnabled && latencyTransition === 'degraded' && !isDown) {
statements.push(
@@ -255,3 +329,159 @@ export function buildResultStatements(
return { statements, transition, latencyTransition, consecutiveFailures: nextFailures };
}
/**
* Persists an entire scheduled run with a fixed number of D1 statements. The incident statements
* are only added for transitions that occurred, so a healthy 40-monitor run uses two statements.
*/
export async function persistScheduledResults(
env: Env,
items: Array<Pick<ScheduledResult, 'monitor' | 'result' | 'checkedAt' | 'maintenance'>>,
options: { degradedIncidents?: boolean } = {},
): Promise<ScheduledResult[]> {
if (items.length === 0) return [];
const persisted = items.map((item) => ({ ...item, ...computeResultState(item.monitor, item.result, item.checkedAt, item.maintenance) }));
const checksJson = JSON.stringify(
persisted.map(({ monitor, result, checkedAt, maintenance }) => ({
monitorId: monitor.id,
ok: result.ok ? 1 : 0,
degraded: result.degraded ? 1 : 0,
statusCode: result.statusCode,
latencyMs: result.latencyMs,
error: result.error,
checkedAt: checkedAt.getTime(),
maintenance: maintenance ? 1 : 0,
})),
);
const updatesJson = JSON.stringify(
persisted.map(({ monitorUpdate }) => ({
...monitorUpdate,
lastOk: 'lastOk' in monitorUpdate ? (monitorUpdate.lastOk ? 1 : 0) : null,
hasLastOk: 'lastOk' in monitorUpdate ? 1 : 0,
lastDegraded: monitorUpdate.lastDegraded ? 1 : 0,
})),
);
const statements: D1PreparedStatement[] = [
env.DB.prepare(
`INSERT INTO checks (monitor_id, ok, degraded, status_code, latency_ms, error, checked_at, maintenance)
SELECT json_extract(value, '$.monitorId'), json_extract(value, '$.ok'), json_extract(value, '$.degraded'),
json_extract(value, '$.statusCode'), json_extract(value, '$.latencyMs'), json_extract(value, '$.error'),
json_extract(value, '$.checkedAt'), json_extract(value, '$.maintenance')
FROM json_each(?1)`,
).bind(checksJson),
env.DB.prepare(
`UPDATE monitors SET
last_ok = CASE WHEN json_extract(v.value, '$.hasLastOk') = 1 THEN json_extract(v.value, '$.lastOk') ELSE monitors.last_ok END,
consecutive_failures = json_extract(v.value, '$.consecutiveFailures'),
consecutive_slow = json_extract(v.value, '$.consecutiveSlow'),
last_degraded = json_extract(v.value, '$.lastDegraded'),
last_status_code = json_extract(v.value, '$.lastStatusCode'),
last_latency_ms = json_extract(v.value, '$.lastLatencyMs'),
last_error = json_extract(v.value, '$.lastError'),
last_checked_at = json_extract(v.value, '$.lastCheckedAt'),
updated_at = json_extract(v.value, '$.updatedAt')
FROM json_each(?1) AS v WHERE monitors.id = json_extract(v.value, '$.id')`,
).bind(updatesJson),
];
const appendOpenedIncidents = (kind: 'down' | 'degraded', opened: typeof persisted) => {
if (opened.length === 0) return;
const rows = JSON.stringify(
opened.map(({ monitor, result, checkedAt, consecutiveFailures }) => ({
monitorId: monitor.id,
impact:
kind === 'degraded'
? 'minor'
: result.statusCode === null || result.statusCode >= 500 || consecutiveFailures >= 10
? 'major'
: 'minor',
startedAt: checkedAt.getTime(),
statusCode: result.statusCode,
error: result.error,
})),
);
statements.push(
env.DB.prepare(
`INSERT INTO incidents (status, impact, source, kind, started_at, start_status_code, start_error, created_at, updated_at)
SELECT 'investigating', json_extract(value, '$.impact'), 'auto', ?2, json_extract(value, '$.startedAt'),
json_extract(value, '$.statusCode'), json_extract(value, '$.error'), json_extract(value, '$.startedAt'), json_extract(value, '$.startedAt')
FROM json_each(?1) ORDER BY CAST(key AS INTEGER)`,
).bind(rows, kind),
env.DB.prepare(
`WITH inserted(last_id) AS MATERIALIZED (SELECT last_insert_rowid())
INSERT INTO incident_monitors (incident_id, monitor_id)
SELECT inserted.last_id - json_array_length(?1) + 1 + CAST(j.key AS INTEGER),
json_extract(j.value, '$.monitorId')
FROM json_each(?1) j CROSS JOIN inserted`,
).bind(rows),
);
};
const openedDown = persisted.filter((item) => !item.maintenance && item.transition === 'opened');
appendOpenedIncidents('down', openedDown);
if (options.degradedIncidents) {
appendOpenedIncidents(
'degraded',
persisted.filter(
(item) => !item.maintenance && item.monitor.alertsEnabled && item.latencyTransition === 'degraded' && item.transition !== 'opened',
),
);
}
const appendResolution = (kind: 'down' | 'degraded', resolved: typeof persisted, body: string) => {
if (resolved.length === 0) return;
const rows = JSON.stringify(resolved.map(({ monitor, checkedAt }) => ({ monitorId: monitor.id, checkedAt: checkedAt.getTime() })));
statements.push(
env.DB.prepare(
`INSERT INTO incident_updates (incident_id, status, body, source, created_at)
SELECT i.id, 'resolved', ?2, 'system', json_extract(j.value, '$.checkedAt')
FROM json_each(?1) j JOIN incident_monitors im ON im.monitor_id = json_extract(j.value, '$.monitorId')
JOIN incidents i ON i.id = im.incident_id
WHERE i.source = 'auto' AND i.kind = ?3 AND i.resolved_at IS NULL`,
).bind(rows, body, kind),
env.DB.prepare(
`UPDATE incidents SET
status = 'resolved', resolved_at = json_extract(j.value, '$.checkedAt'),
duration_ms = json_extract(j.value, '$.checkedAt') - incidents.started_at,
updated_at = json_extract(j.value, '$.checkedAt')
FROM json_each(?1) j JOIN incident_monitors im ON im.monitor_id = json_extract(j.value, '$.monitorId')
WHERE incidents.id = im.incident_id AND incidents.source = 'auto' AND incidents.kind = ?2 AND incidents.resolved_at IS NULL`,
).bind(rows, kind),
);
};
appendResolution(
'down',
persisted.filter((item) => !item.maintenance && item.transition === 'resolved'),
RECOVERY_UPDATE_BODY,
);
if (options.degradedIncidents) {
appendResolution(
'degraded',
persisted.filter((item) => !item.maintenance && item.latencyTransition === 'recovered'),
DEGRADED_RECOVERY_UPDATE_BODY,
);
if (openedDown.length > 0) {
const rows = JSON.stringify(openedDown.map(({ monitor, checkedAt }) => ({ monitorId: monitor.id, checkedAt: checkedAt.getTime() })));
statements.push(
env.DB.prepare(
`INSERT INTO incident_updates (incident_id, status, body, source, created_at)
SELECT i.id, 'resolved', ?2, 'system', json_extract(j.value, '$.checkedAt')
FROM json_each(?1) j JOIN incident_monitors im ON im.monitor_id = json_extract(j.value, '$.monitorId')
JOIN incidents i ON i.id = im.incident_id
WHERE i.source = 'auto' AND i.kind = 'degraded' AND i.resolved_at IS NULL`,
).bind(rows, DEGRADED_SUPERSEDED_UPDATE_BODY),
env.DB.prepare(
`UPDATE incidents SET status = 'resolved', resolved_at = json_extract(j.value, '$.checkedAt'),
duration_ms = json_extract(j.value, '$.checkedAt') - incidents.started_at,
updated_at = json_extract(j.value, '$.checkedAt')
FROM json_each(?1) j JOIN incident_monitors im ON im.monitor_id = json_extract(j.value, '$.monitorId')
WHERE incidents.id = im.incident_id AND incidents.source = 'auto' AND incidents.kind = 'degraded' AND incidents.resolved_at IS NULL`,
).bind(rows),
);
}
}
await env.DB.batch(statements);
return persisted.map(({ monitorUpdate: _monitorUpdate, ...item }) => item);
}
+35 -40
View File
@@ -1,20 +1,17 @@
import { and, eq, sql } from 'drizzle-orm';
import { and, eq, inArray, sql } from 'drizzle-orm';
import type { AutopilotEvent } from '../autopilot';
import { getDb } from '../db/client';
import { aiSettings, monitors } from '../db/schema';
import { DEFAULT_RUN_LIMITS, resolveRunLimits } from '../lib/runtime-config';
import { DEFAULT_RUN_LIMITS, resolveCheckRunConfig, resolveRunLimits } from '../lib/runtime-config';
import { loadActiveMaintenance } from '../maintenance/windows';
import { buildAlertEvent } from '../notifications/compose';
import { dispatchNotification, type NotificationBudget } from '../notifications/dispatch';
import { buildResultStatements } from './persist-result';
import { dispatchRunNotifications, type NotificationBudget } from '../notifications/dispatch';
import { persistScheduledResults } from './persist-result';
import { runCheck, runCheckWithRetries, type RetryBudget } from './run-check';
const MAX_MONITORS_PER_RUN = 40;
const CONCURRENCY = 10;
const MAX_BATCH_STATEMENTS = 100;
/** Default retry ceiling for one scheduled run; override with the RETRY_ATTEMPTS_PER_RUN env var. */
export const MAX_RETRY_ATTEMPTS_PER_RUN = DEFAULT_RUN_LIMITS.retryAttemptsPerRun;
const RETRY_DEADLINE_MS = 90_000;
const RETRY_DEADLINE_MS = 45_000;
export type DueCheckSummary = {
checked: number;
@@ -29,8 +26,10 @@ export type DueCheckSummary = {
export async function runDueChecks(env: Env, ctx?: Pick<ExecutionContext, 'waitUntil'>): Promise<DueCheckSummary> {
const db = getDb(env);
const now = Date.now();
const due = await db
.select()
const limits = resolveRunLimits(env);
const { maxMonitorsPerRun, concurrency } = resolveCheckRunConfig(env);
const dueIds = db
.select({ id: monitors.id })
.from(monitors)
.where(
and(
@@ -39,10 +38,16 @@ export async function runDueChecks(env: Env, ctx?: Pick<ExecutionContext, 'waitU
),
)
.orderBy(sql`${monitors.lastCheckedAt} ASC NULLS FIRST`)
.limit(MAX_MONITORS_PER_RUN);
.limit(maxMonitorsPerRun);
// Claim and return due monitors in one SQLite statement. Concurrent cron invocations serialize
// this write, so a later invocation sees the claimed timestamp and cannot run the same monitor.
const due = await db
.update(monitors)
.set({ lastCheckedAt: new Date(now) })
.where(inArray(monitors.id, dueIds))
.returning();
if (due.length === 0) return { checked: 0, up: 0, down: 0, pending: 0, opened: 0, retries: 0, events: [] };
const limits = resolveRunLimits(env);
const [activeMaintenance, [settings]] = await Promise.all([
loadActiveMaintenance(db, new Date()),
db.select().from(aiSettings).where(eq(aiSettings.id, 1)).limit(1),
@@ -56,8 +61,8 @@ export async function runDueChecks(env: Env, ctx?: Pick<ExecutionContext, 'waitU
maintenance: boolean;
}> = [];
for (let offset = 0; offset < due.length; offset += CONCURRENCY) {
const batch = due.slice(offset, offset + CONCURRENCY);
for (let offset = 0; offset < due.length; offset += concurrency) {
const batch = due.slice(offset, offset + concurrency);
const results = await Promise.all(
batch.map(async (monitor) => {
const maintenance = activeMaintenance.has(monitor.id);
@@ -69,39 +74,29 @@ export async function runDueChecks(env: Env, ctx?: Pick<ExecutionContext, 'waitU
completed.push(...results);
}
const persisted = completed.map(({ monitor, result, checkedAt, maintenance }) => ({
monitor,
result,
checkedAt,
...buildResultStatements(db, monitor, result, checkedAt, maintenance, {
degradedIncidents: Boolean(settings?.autopilotEnabled && settings.autopilotDegradedIncidents),
}),
}));
let statementChunk: (typeof persisted)[number]['statements'] = [];
for (const item of persisted) {
if (statementChunk.length > 0 && statementChunk.length + item.statements.length > MAX_BATCH_STATEMENTS) {
await db.batch(statementChunk as [(typeof statementChunk)[number], ...typeof statementChunk]);
statementChunk = [];
}
statementChunk.push(...item.statements);
}
if (statementChunk.length > 0) await db.batch(statementChunk as [(typeof statementChunk)[number], ...typeof statementChunk]);
const persisted = await persistScheduledResults(env, completed, {
degradedIncidents: Boolean(settings?.autopilotEnabled && settings.autopilotDegradedIncidents),
});
const notificationBudget: NotificationBudget = { remaining: limits.notificationsPerRun };
const notifications = persisted.flatMap((item) => {
const work: Promise<unknown>[] = [];
const notificationEvents = persisted.flatMap((item) => {
const events = [];
if (item.transition === 'opened' || item.transition === 'resolved') {
work.push(dispatchNotification(env, buildAlertEvent(item.monitor, item.result, item.transition, item.checkedAt), notificationBudget));
events.push({
event: buildAlertEvent(item.monitor, item.result, item.transition, item.checkedAt),
monitorAlertsEnabled: item.monitor.alertsEnabled,
});
}
if (item.latencyTransition) {
work.push(
dispatchNotification(env, buildAlertEvent(item.monitor, item.result, item.latencyTransition, item.checkedAt), notificationBudget),
);
events.push({
event: buildAlertEvent(item.monitor, item.result, item.latencyTransition, item.checkedAt),
monitorAlertsEnabled: item.monitor.alertsEnabled,
});
}
return work;
return events;
});
if (notifications.length > 0) {
const notificationWork = Promise.all(notifications).then(() => undefined);
if (notificationEvents.length > 0) {
const notificationWork = dispatchRunNotifications(env, notificationEvents, notificationBudget);
if (ctx) ctx.waitUntil(notificationWork);
else await notificationWork;
}
+34 -40
View File
@@ -47,44 +47,40 @@ export const loginAttempts = sqliteTable(
],
);
export const monitors = sqliteTable(
'monitors',
{
id: integer('id').primaryKey({ autoIncrement: true }),
name: text('name').notNull(),
url: text('url').notNull(),
method: text('method').notNull().default('GET'),
expectedStatus: integer('expected_status').notNull().default(200),
expectKeyword: text('expect_keyword'),
keywordInverted: integer('keyword_inverted', { mode: 'boolean' }).notNull().default(false),
requestHeaders: text('request_headers'),
requestBody: text('request_body'),
degradedLatencyMs: integer('degraded_latency_ms'),
intervalSeconds: integer('interval_seconds').notNull().default(300),
timeoutMs: integer('timeout_ms').notNull().default(10_000),
enabled: integer('enabled', { mode: 'boolean' }).notNull().default(true),
alertsEnabled: integer('alerts_enabled', { mode: 'boolean' }).notNull().default(true),
/** Number of immediate retries after a failed attempt. Zero disables retries. */
retryCount: integer('retry_count').notNull().default(1),
/** Consecutive failed checks required before confirming an outage. */
failureThreshold: integer('failure_threshold').notNull().default(2),
/** Failures since the last successful check. Maintenance checks do not change this value. */
consecutiveFailures: integer('consecutive_failures').notNull().default(0),
/** Slow successful checks since latency last recovered. Maintenance checks do not change this value. */
consecutiveSlow: integer('consecutive_slow').notNull().default(0),
/** Confirmed state, not the raw latest result. */
lastOk: integer('last_ok', { mode: 'boolean' }),
/** Confirmed degraded state. A down monitor is never degraded. */
lastDegraded: integer('last_degraded', { mode: 'boolean' }).notNull().default(false),
lastStatusCode: integer('last_status_code'),
lastLatencyMs: integer('last_latency_ms'),
lastError: text('last_error'),
lastCheckedAt: integer('last_checked_at', { mode: 'timestamp_ms' }),
createdAt: integer('created_at', { mode: 'timestamp_ms' }).notNull(),
updatedAt: integer('updated_at', { mode: 'timestamp_ms' }).notNull(),
},
(table) => [index('monitors_enabled_last_checked_at_idx').on(table.enabled, table.lastCheckedAt)],
);
export const monitors = sqliteTable('monitors', {
id: integer('id').primaryKey({ autoIncrement: true }),
name: text('name').notNull(),
url: text('url').notNull(),
method: text('method').notNull().default('GET'),
expectedStatus: integer('expected_status').notNull().default(200),
expectKeyword: text('expect_keyword'),
keywordInverted: integer('keyword_inverted', { mode: 'boolean' }).notNull().default(false),
requestHeaders: text('request_headers'),
requestBody: text('request_body'),
degradedLatencyMs: integer('degraded_latency_ms'),
intervalSeconds: integer('interval_seconds').notNull().default(300),
timeoutMs: integer('timeout_ms').notNull().default(10_000),
enabled: integer('enabled', { mode: 'boolean' }).notNull().default(true),
alertsEnabled: integer('alerts_enabled', { mode: 'boolean' }).notNull().default(true),
/** Number of immediate retries after a failed attempt. Zero disables retries. */
retryCount: integer('retry_count').notNull().default(1),
/** Consecutive failed checks required before confirming an outage. */
failureThreshold: integer('failure_threshold').notNull().default(2),
/** Failures since the last successful check. Maintenance checks do not change this value. */
consecutiveFailures: integer('consecutive_failures').notNull().default(0),
/** Slow successful checks since latency last recovered. Maintenance checks do not change this value. */
consecutiveSlow: integer('consecutive_slow').notNull().default(0),
/** Confirmed state, not the raw latest result. */
lastOk: integer('last_ok', { mode: 'boolean' }),
/** Confirmed degraded state. A down monitor is never degraded. */
lastDegraded: integer('last_degraded', { mode: 'boolean' }).notNull().default(false),
lastStatusCode: integer('last_status_code'),
lastLatencyMs: integer('last_latency_ms'),
lastError: text('last_error'),
lastCheckedAt: integer('last_checked_at', { mode: 'timestamp_ms' }),
createdAt: integer('created_at', { mode: 'timestamp_ms' }).notNull(),
updatedAt: integer('updated_at', { mode: 'timestamp_ms' }).notNull(),
});
export const maintenanceWindows = sqliteTable(
'maintenance_windows',
@@ -134,8 +130,6 @@ export const checks = sqliteTable(
},
(table) => [
index('checks_monitor_id_checked_at_idx').on(table.monitorId, table.checkedAt),
// Daily rollup and retention cleanup scan by checked_at across every monitor; without this
// index those become full-table scans, which dominate D1 "rows read" at scale.
index('checks_checked_at_idx').on(table.checkedAt),
],
);
+10 -4
View File
@@ -15,6 +15,10 @@ import { runDailyRollup } from './scheduled/rollup';
const app = new Hono<{ Bindings: Env }>();
export function shouldRunFiveMinuteWork(scheduledTime: number): boolean {
return new Date(scheduledTime).getUTCMinutes() % 5 === 0;
}
app.use('/api/*', csrf());
app.route(
@@ -46,8 +50,9 @@ export default {
fetch: app.fetch,
async scheduled(controller, env, ctx) {
if (controller.cron === '5 0 * * *') {
const result = await runDailyRollup(env, new Date(controller.scheduledTime));
await cleanupStaleData(env);
const scheduledAt = new Date(controller.scheduledTime);
const result = await runDailyRollup(env, scheduledAt);
await cleanupStaleData(env, scheduledAt);
console.log(
JSON.stringify({
message: 'daily rollup completed',
@@ -59,10 +64,11 @@ export default {
return;
}
await cleanupExpiredAuthRecords(env);
const runFiveMinuteWork = shouldRunFiveMinuteWork(controller.scheduledTime);
if (runFiveMinuteWork) await cleanupExpiredAuthRecords(env);
const result = await runDueChecks(env, ctx);
ctx.waitUntil(
runAutopilot(env, { events: result.events }).then((autopilot) => {
runAutopilot(env, { events: result.events, skipSweep: !runFiveMinuteWork }).then((autopilot) => {
console.log(JSON.stringify({ message: 'autopilot run completed', ...autopilot }));
}),
);
+22 -2
View File
@@ -2,7 +2,7 @@
* Environment-tunable runtime knobs.
*
* Cloudflare's free plan caps a Worker invocation at 50 subrequests and 10 ms CPU, and the
* 5-minute cron runs checks, retries, notifications and AI autopilot inside a single
* minute cron runs checks, retries, notifications and AI autopilot inside a single
* invocation. The defaults below keep a healthy run well under that ceiling and make a large
* correlated outage shed the least-critical work first (AI, then retries) instead of failing
* mid-batch. Raise them with same-named environment variables on the Workers Paid plan, where
@@ -20,6 +20,13 @@ export type RunLimits = {
aiFollowupCallsPerRun: number;
};
export type CheckRunConfig = {
/** Maximum monitor fetches started by one scheduled invocation. */
maxMonitorsPerRun: number;
/** Maximum target requests in flight at once. */
concurrency: number;
};
export const DEFAULT_RUN_LIMITS: RunLimits = {
retryAttemptsPerRun: 15,
notificationsPerRun: 12,
@@ -27,8 +34,13 @@ export const DEFAULT_RUN_LIMITS: RunLimits = {
aiFollowupCallsPerRun: 2,
};
export const DEFAULT_CHECK_RUN_CONFIG: CheckRunConfig = {
maxMonitorsPerRun: 40,
concurrency: 6,
};
/** Seconds the public status responses are held in the edge cache. 0 disables edge caching. */
export const DEFAULT_STATUS_CACHE_SECONDS = 30;
export const DEFAULT_STATUS_CACHE_SECONDS = 60;
type EnvVars = Record<string, string | undefined>;
@@ -48,6 +60,14 @@ export function resolveRunLimits(env: Env): RunLimits {
};
}
export function resolveCheckRunConfig(env: Env): CheckRunConfig {
const vars = env as unknown as EnvVars;
return {
maxMonitorsPerRun: boundedInt(vars.MAX_MONITORS_PER_RUN, DEFAULT_CHECK_RUN_CONFIG.maxMonitorsPerRun, 1),
concurrency: boundedInt(vars.CHECK_CONCURRENCY, DEFAULT_CHECK_RUN_CONFIG.concurrency, 1),
};
}
export function resolveStatusCacheSeconds(env: Env): number {
return boundedInt((env as unknown as EnvVars).STATUS_CACHE_SECONDS, DEFAULT_STATUS_CACHE_SECONDS, 0);
}
+132
View File
@@ -15,6 +15,7 @@ export const MAX_NOTIFICATIONS_PER_RUN = 40;
export type NotificationBudget = { remaining: number };
type DeliveryResult = { ok: boolean; statusCode: number | null; error: string | null; attempts: number };
type RunNotification = { event: NotificationEvent; monitorAlertsEnabled: boolean };
function isChannelType(value: string): value is ChannelType {
return CHANNEL_TYPES.some((type) => type === value);
@@ -66,6 +67,137 @@ async function persistDeliveries(env: Env, event: NotificationEvent, results: Ar
await db.batch(statements as [(typeof statements)[number], ...typeof statements]);
}
async function persistRunDeliveries(
env: Env,
rows: Array<{ event: NotificationEvent; channelId: number; result: DeliveryResult }>,
): Promise<void> {
if (rows.length === 0) return;
const createdAt = Date.now();
await env.DB.prepare(
`INSERT INTO notification_deliveries
(channel_id, incident_id, monitor_id, event, ok, status_code, error, attempts, created_at)
SELECT json_extract(value, '$.channelId'), json_extract(value, '$.incidentId'), json_extract(value, '$.monitorId'),
json_extract(value, '$.event'), json_extract(value, '$.ok'), json_extract(value, '$.statusCode'),
json_extract(value, '$.error'), json_extract(value, '$.attempts'), ?2
FROM json_each(?1)`,
)
.bind(
JSON.stringify(
rows.map(({ event, channelId, result }) => ({
channelId,
incidentId: event.incidentId,
monitorId: event.monitor?.id ?? null,
event: event.kind,
ok: result.ok ? 1 : 0,
statusCode: result.statusCode,
error: result.error,
attempts: result.attempts,
})),
),
createdAt,
)
.run();
}
function invalidDelivery(error: unknown): DeliveryResult {
return {
ok: false,
statusCode: null,
error: error instanceof Error ? error.message.slice(0, 500) : String(error).slice(0, 500),
attempts: 0,
};
}
/** Routes all scheduled-run events from one channel snapshot and persists deliveries once. */
export async function dispatchRunNotifications(env: Env, inputs: RunNotification[], budget: NotificationBudget): Promise<void> {
const events = inputs.filter((input) => input.monitorAlertsEnabled && input.event.monitor);
if (events.length === 0) return;
const db = getDb(env);
const channels = await db.select().from(notificationChannels).where(eq(notificationChannels.enabled, true));
if (channels.length === 0) return;
const assignments = await db
.select()
.from(notificationChannelMonitors)
.where(
inArray(
notificationChannelMonitors.channelId,
channels.map((channel) => channel.id),
),
);
const monitorIdsByChannel = new Map<number, Set<number>>();
for (const assignment of assignments) {
let monitorIds = monitorIdsByChannel.get(assignment.channelId);
if (!monitorIds) {
monitorIds = new Set<number>();
monitorIdsByChannel.set(assignment.channelId, monitorIds);
}
monitorIds.add(assignment.monitorId);
}
const downEvents = events.filter(({ event }) => event.kind === 'down' || event.kind === 'recovered');
const incidentByMonitorAndState = new Map<string, number>();
if (downEvents.length > 0) {
const monitorIds = [...new Set(downEvents.map(({ event }) => event.monitor!.id))];
const incidents = await env.DB.prepare(
`SELECT im.monitor_id, i.id, CASE WHEN i.resolved_at IS NULL THEN 0 ELSE 1 END AS resolved
FROM incident_monitors im JOIN incidents i ON i.id = im.incident_id
JOIN json_each(?1) targets ON im.monitor_id = targets.value
WHERE i.source = 'auto' AND i.kind = 'down'
ORDER BY i.started_at DESC`,
)
.bind(JSON.stringify(monitorIds))
.all<{ monitor_id: number; id: number; resolved: number }>();
for (const incident of incidents.results) {
const key = `${incident.monitor_id}:${incident.resolved}`;
if (!incidentByMonitorAndState.has(key)) incidentByMonitorAndState.set(key, incident.id);
}
}
const work: Array<{ event: NotificationEvent; channel: (typeof channels)[number] }> = [];
for (const { event } of events) {
const effectiveEvent =
event.monitor && (event.kind === 'down' || event.kind === 'recovered')
? {
...event,
incidentId: incidentByMonitorAndState.get(`${event.monitor.id}:${event.kind === 'recovered' ? 1 : 0}`) ?? null,
}
: event;
for (const channel of channels) {
const monitorIds = monitorIdsByChannel.get(channel.id);
if (!monitorIds || monitorIds.size === 0 || monitorIds.has(event.monitor!.id)) work.push({ event: effectiveEvent, channel });
}
}
const available = Math.max(0, budget.remaining);
const sendable = work.slice(0, available);
budget.remaining -= sendable.length;
const skipped = work.slice(sendable.length).map(({ event, channel }) => ({
event,
channelId: channel.id,
result: { ok: false, statusCode: null, error: 'skipped: per-run limit', attempts: 0 } satisfies DeliveryResult,
}));
const settled = await Promise.allSettled(
sendable.map(async ({ event, channel }) => {
if (!isChannelType(channel.type)) throw new Error(`Unsupported channel type: ${channel.type}`);
let rawConfig: unknown;
try {
rawConfig = JSON.parse(channel.config);
} catch {
throw new Error('Invalid stored channel configuration');
}
const config = parseChannelConfig(channel.type, rawConfig);
if (typeof config === 'string') throw new Error(config);
return { event, channelId: channel.id, result: await sendRequest(formatChannel(channel.type, config, event)) };
}),
);
const delivered = settled.map((result, index) =>
result.status === 'fulfilled'
? result.value
: { event: sendable[index].event, channelId: sendable[index].channel.id, result: invalidDelivery(result.reason) },
);
await persistRunDeliveries(env, [...delivered, ...skipped]);
}
export async function dispatchNotification(env: Env, event: NotificationEvent, budget?: NotificationBudget): Promise<void> {
const db = getDb(env);
let effectiveEvent = event;
+6 -6
View File
@@ -142,12 +142,12 @@ channelRoutes.post('/', async (context) => {
})
.returning();
if (monitorIds.length > 0) {
await db.batch(
monitorIds.map((monitorId) => db.insert(notificationChannelMonitors).values({ channelId: channel.id, monitorId })) as [
ReturnType<typeof db.insert>,
...ReturnType<typeof db.insert>[],
],
);
await context.env.DB.prepare(
`INSERT INTO notification_channel_monitors (channel_id, monitor_id)
SELECT ?2, value FROM json_each(?1)`,
)
.bind(JSON.stringify(monitorIds), channel.id)
.run();
}
return context.json({ channel: publicChannel(channel, monitorIds, null) }, 201);
});
+20 -3
View File
@@ -371,7 +371,21 @@ monitorRoutes.get('/:id/checks', async (context) => {
const id = parseId(context.req.param('id'));
if (id === null) return context.json({ message: 'Monitor not found' }, 404);
const limit = parseLimit(context.req.query('limit'), 100, 500);
const rows = await getDb(context.env).select().from(checks).where(eq(checks.monitorId, id)).orderBy(desc(checks.checkedAt)).limit(limit);
const rows = await getDb(context.env)
.select({
id: checks.id,
ok: checks.ok,
degraded: checks.degraded,
statusCode: checks.statusCode,
latencyMs: checks.latencyMs,
error: checks.error,
checkedAt: checks.checkedAt,
maintenance: checks.maintenance,
})
.from(checks)
.where(eq(checks.monitorId, id))
.orderBy(desc(checks.checkedAt))
.limit(limit);
return context.json({ checks: rows });
});
@@ -424,7 +438,7 @@ monitorRoutes.get('/:id/stats', async (context) => {
const currentDayMs = Date.UTC(new Date(now).getUTCFullYear(), new Date(now).getUTCMonth(), new Date(now).getUTCDate());
const windows = [
{ key: '24h', start: now - 24 * 60 * 60 * 1000, raw: true },
{ key: '7d', start: now - 7 * 24 * 60 * 60 * 1000, raw: true },
{ key: '7d', start: currentDayMs - 6 * 24 * 60 * 60 * 1000, raw: false },
{ key: '30d', start: currentDayMs - 29 * 24 * 60 * 60 * 1000, raw: false },
{ key: '90d', start: currentDayMs - 89 * 24 * 60 * 60 * 1000, raw: false },
] as const;
@@ -517,6 +531,8 @@ monitorRoutes.post('/', async (context) => {
if (!parsed.ok) return context.json({ message: parsed.message }, 400);
const now = new Date();
const intervalSeconds = parsed.value.intervalSeconds!;
const lastCheckedAt = new Date(now.getTime() - Math.floor(Math.random() * intervalSeconds) * 1000);
const [monitor] = await getDb(context.env)
.insert(monitors)
.values({
@@ -529,12 +545,13 @@ monitorRoutes.post('/', async (context) => {
requestHeaders: parsed.value.requestHeaders ?? null,
requestBody: parsed.value.requestBody ?? null,
degradedLatencyMs: parsed.value.degradedLatencyMs ?? null,
intervalSeconds: parsed.value.intervalSeconds!,
intervalSeconds,
timeoutMs: parsed.value.timeoutMs!,
retryCount: parsed.value.retryCount ?? 1,
failureThreshold: parsed.value.failureThreshold ?? 2,
enabled: parsed.value.enabled ?? true,
alertsEnabled: parsed.value.alertsEnabled ?? true,
lastCheckedAt,
createdAt: now,
updatedAt: now,
})
+1 -1
View File
@@ -4,7 +4,7 @@ import { resolveStatusCacheSeconds } from '../lib/runtime-config';
import { rewriteHead } from '../seo/html';
import { absoluteBase, escapeHtml, incidentHead, statusHead } from '../seo/meta';
type AppFetch = (request: Request, env: Env, executionCtx: ExecutionContext) => Response | Promise<Response>;
type AppFetch = (request: Request, env: Env, executionCtx: Context<{ Bindings: Env }>['executionCtx']) => Response | Promise<Response>;
type EdgeCache = {
match(request: RequestInfo | URL): Promise<Response | undefined>;
put(request: RequestInfo | URL, response: Response): Promise<void>;
+3 -1
View File
@@ -321,6 +321,8 @@ statusRoutes.get('/incidents', async (context) => {
});
statusRoutes.get('/incidents/:id', async (context) => {
const cached = await cachedStatusResponse(context);
if (cached) return cached;
const id = parseId(context.req.param('id'));
if (id === null) return context.json({ message: 'Incident not found' }, 404);
const db = getDb(context.env);
@@ -338,7 +340,7 @@ statusRoutes.get('/incidents/:id', async (context) => {
updates.length > 0
? updates
: [{ status: incident.status, body: deterministicIncidentMessage(incident.startStatusCode), createdAt: incident.startedAt }];
return context.json({
return jsonWithEdgeCache(context, {
incident: {
id: incident.id,
title: publicIncidentTitle(incident),
+3 -3
View File
@@ -32,12 +32,12 @@ export async function cleanupExpiredAuthRecords(env: Env) {
* so the delete only scans the expiring slice, and running daily keeps that scan to ~one day
* of rows.
*/
export async function cleanupStaleData(env: Env) {
export async function cleanupStaleData(env: Env, now = new Date()) {
const db = getDb(env);
const now = new Date();
const checkCutoff = Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()) - CHECK_RETENTION_MS;
await db.batch([
db.delete(checks).where(lt(checks.checkedAt, new Date(now.getTime() - CHECK_RETENTION_MS))),
db.delete(checks).where(lt(checks.checkedAt, new Date(checkCutoff))),
db.delete(monitorDailyStats).where(lt(monitorDailyStats.day, new Date(now.getTime() - DAILY_STATS_RETENTION_MS))),
db.delete(notificationDeliveries).where(lt(notificationDeliveries.createdAt, new Date(now.getTime() - DELIVERY_RETENTION_MS))),
db.delete(aiEvents).where(lt(aiEvents.createdAt, new Date(now.getTime() - AI_EVENT_RETENTION_MS))),
+24 -48
View File
@@ -1,57 +1,33 @@
import { and, eq, gte, lt, sql } from 'drizzle-orm';
import { getDb } from '../db/client';
import { checks, monitorDailyStats } from '../db/schema';
export type DailyRollupSummary = {
day: string;
monitors: number;
};
export async function runDailyRollup(env: Env, now = new Date()): Promise<DailyRollupSummary> {
const db = getDb(env);
const currentUtcDay = Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate());
const dayStart = new Date(currentUtcDay - 24 * 60 * 60 * 1000);
const dayEnd = new Date(currentUtcDay);
const dayStart = currentUtcDay - 24 * 60 * 60 * 1000;
const dayEnd = currentUtcDay;
const count = await env.DB.prepare(
`SELECT count(DISTINCT monitor_id) AS count FROM checks
WHERE checked_at >= ?1 AND checked_at < ?2 AND maintenance = 0`,
)
.bind(dayStart, dayEnd)
.first<{ count: number }>();
await env.DB.prepare(
`INSERT INTO monitor_daily_stats
(monitor_id, day, total_checks, up_checks, avg_latency_ms, min_latency_ms, max_latency_ms)
SELECT monitor_id, ?3, count(*), sum(CASE WHEN ok = 1 THEN 1 ELSE 0 END),
round(avg(latency_ms)), min(latency_ms), max(latency_ms)
FROM checks
WHERE checked_at >= ?1 AND checked_at < ?2 AND maintenance = 0
GROUP BY monitor_id
ON CONFLICT(monitor_id, day) DO UPDATE SET
total_checks = excluded.total_checks, up_checks = excluded.up_checks,
avg_latency_ms = excluded.avg_latency_ms, min_latency_ms = excluded.min_latency_ms,
max_latency_ms = excluded.max_latency_ms`,
)
.bind(dayStart, dayEnd, dayStart)
.run();
const rows = await db
.select({
monitorId: checks.monitorId,
totalChecks: sql<number>`count(*)`,
upChecks: sql<number>`sum(case when ${checks.ok} = 1 then 1 else 0 end)`,
avgLatencyMs: sql<number | null>`round(avg(${checks.latencyMs}))`,
minLatencyMs: sql<number | null>`min(${checks.latencyMs})`,
maxLatencyMs: sql<number | null>`max(${checks.latencyMs})`,
})
.from(checks)
.where(and(gte(checks.checkedAt, dayStart), lt(checks.checkedAt, dayEnd), eq(checks.maintenance, false)))
.groupBy(checks.monitorId);
if (rows.length > 0) {
const statements = rows.map((row) =>
db
.insert(monitorDailyStats)
.values({
monitorId: row.monitorId,
day: dayStart,
totalChecks: row.totalChecks,
upChecks: row.upChecks,
avgLatencyMs: row.avgLatencyMs,
minLatencyMs: row.minLatencyMs,
maxLatencyMs: row.maxLatencyMs,
})
.onConflictDoUpdate({
target: [monitorDailyStats.monitorId, monitorDailyStats.day],
set: {
totalChecks: row.totalChecks,
upChecks: row.upChecks,
avgLatencyMs: row.avgLatencyMs,
minLatencyMs: row.minLatencyMs,
maxLatencyMs: row.maxLatencyMs,
},
}),
);
await db.batch(statements as [(typeof statements)[number], ...typeof statements]);
}
return { day: dayStart.toISOString().slice(0, 10), monitors: rows.length };
return { day: new Date(dayStart).toISOString().slice(0, 10), monitors: Number(count?.count ?? 0) };
}
+30 -1
View File
@@ -1,7 +1,7 @@
import { applyD1Migrations, type D1Migration } from 'cloudflare:test';
import { env } from 'cloudflare:workers';
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { dispatchNotification } from '../src/worker/notifications/dispatch';
import { dispatchNotification, dispatchRunNotifications } from '../src/worker/notifications/dispatch';
import { discordProvider, slackProvider, telegramProvider, webhookProvider } from '../src/worker/notifications/providers';
import { parseChannelInput } from '../src/worker/routes/channels';
@@ -110,6 +110,35 @@ describe('notification channels', () => {
expect(delivery).toEqual({ ok: 1, status_code: 204, attempts: 2 });
});
it('routes a scheduled event batch and writes all delivery rows together', async () => {
await insertChannel('All services', '{"url":"https://hooks.example.test/events"}');
const assignedChannel = await insertChannel('API only', '{"url":"https://hooks.example.test/api"}');
await env.DB.prepare('INSERT INTO notification_channel_monitors (channel_id, monitor_id) VALUES (?, 7)').bind(assignedChannel).run();
const fetchMock = vi.fn(async () => new Response(null, { status: 204 }));
vi.stubGlobal('fetch', fetchMock);
await dispatchRunNotifications(
env,
[
{ event, monitorAlertsEnabled: true },
{ event: { ...event, kind: 'recovered', title: 'API recovered' }, monitorAlertsEnabled: true },
],
{ remaining: 3 },
);
expect(fetchMock).toHaveBeenCalledTimes(3);
const deliveries = await env.DB.prepare('SELECT event, ok FROM notification_deliveries ORDER BY id').all<{
event: string;
ok: number;
}>();
expect(deliveries.results).toEqual([
{ event: 'down', ok: 1 },
{ event: 'down', ok: 1 },
{ event: 'recovered', ok: 1 },
{ event: 'recovered', ok: 0 },
]);
});
it('does not retry a 400 response', async () => {
await insertChannel('All services', '{"url":"https://hooks.example.test/events"}');
const fetchMock = vi.fn(async () => new Response(null, { status: 400 }));
+59
View File
@@ -2,7 +2,10 @@ import { applyD1Migrations, type D1Migration } from 'cloudflare:test';
import { env } from 'cloudflare:workers';
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { runAutopilot } from '../src/worker/autopilot';
import { persistScheduledResults } from '../src/worker/checks/persist-result';
import { MAX_RETRY_ATTEMPTS_PER_RUN, runDueChecks } from '../src/worker/checks/run-due-checks';
import { getDb } from '../src/worker/db/client';
import { monitors } from '../src/worker/db/schema';
async function clearMonitoringTables() {
await env.DB.batch([
@@ -117,6 +120,35 @@ describe('scheduled monitor checks', () => {
expect(checkCount?.count).toBe(1);
});
it('atomically claims monitors so overlapping runs do not check them twice', async () => {
await insertMonitor();
let releaseFetch!: () => void;
let markFetchStarted!: () => void;
const fetchStarted = new Promise<void>((resolve) => {
markFetchStarted = resolve;
});
const blockedFetch = new Promise<void>((resolve) => {
releaseFetch = resolve;
});
const fetchMock = vi.fn(async () => {
markFetchStarted();
await blockedFetch;
return new Response(null, { status: 200 });
});
vi.stubGlobal('fetch', fetchMock);
const firstRun = runDueChecks(env);
await fetchStarted;
const secondSummary = await runDueChecks(env);
releaseFetch();
const firstSummary = await firstRun;
expect(firstSummary.checked).toBe(1);
expect(secondSummary.checked).toBe(0);
expect(fetchMock).toHaveBeenCalledTimes(1);
expect((await env.DB.prepare('SELECT count(*) AS count FROM checks').first<{ count: number }>())?.count).toBe(1);
});
it('passes a case-insensitive keyword assertion and sends configured request data', async () => {
await insertMonitor({
method: 'POST',
@@ -424,6 +456,33 @@ describe('scheduled monitor checks', () => {
expect(new Set(assignments.results.map((row) => row.incident_id)).size).toBe(2);
});
it('does not associate opened incidents through colliding timestamps', async () => {
const firstId = await insertMonitor({ name: 'First', last_ok: 1, failure_threshold: 1 });
const secondId = await insertMonitor({ name: 'Second', last_ok: 1, failure_threshold: 1 });
const checkedAt = new Date('2026-09-02T12:00:00.000Z');
const existing = await env.DB.prepare(
"INSERT INTO incidents (status, impact, source, kind, started_at, created_at, updated_at) VALUES ('investigating', 'major', 'auto', 'down', ?, ?, ?)",
)
.bind(checkedAt.getTime(), checkedAt.getTime(), checkedAt.getTime())
.run();
const monitorRows = await getDb(env).select().from(monitors);
const byId = new Map(monitorRows.map((monitor) => [monitor.id, monitor]));
const result = { ok: false, degraded: false, statusCode: 503, latencyMs: 10, error: 'Unavailable', attempts: 1 };
await persistScheduledResults(env, [
{ monitor: byId.get(firstId)!, result, checkedAt, maintenance: false },
{ monitor: byId.get(secondId)!, result, checkedAt, maintenance: false },
]);
const assignments = await env.DB.prepare('SELECT incident_id, monitor_id FROM incident_monitors ORDER BY incident_id').all<{
incident_id: number;
monitor_id: number;
}>();
expect(assignments.results).toHaveLength(2);
expect(assignments.results.map((row) => row.monitor_id)).toEqual([firstId, secondId]);
expect(assignments.results.every((row) => row.incident_id !== Number(existing.meta.last_row_id))).toBe(true);
});
it('resolves the open incident on recovery', async () => {
const id = await insertMonitor({ last_ok: 0, consecutive_failures: 2 });
const startedAt = Date.now() - 60_000;
+21
View File
@@ -118,5 +118,26 @@ describe('scheduled cleanup', () => {
expect(await count('notification_deliveries')).toBe(1);
expect(await count('ai_events')).toBe(1);
});
it('retains fresh checks even when a stale check has a higher id', async () => {
const now = new Date('2026-09-02T12:00:00Z');
const monitorId = await insertMonitor();
await env.DB.batch([
// Insert the fresh check first so primary-key and timestamp order disagree.
env.DB.prepare('INSERT INTO checks (monitor_id, ok, latency_ms, checked_at) VALUES (?, 1, 100, ?)').bind(
monitorId,
now.getTime() - DAY_MS,
),
env.DB.prepare('INSERT INTO checks (monitor_id, ok, latency_ms, checked_at) VALUES (?, 0, 300, ?)').bind(
monitorId,
now.getTime() - 8 * DAY_MS,
),
]);
await cleanupStaleData(env, now);
const checks = await env.DB.prepare('SELECT ok FROM checks ORDER BY id').all<{ ok: number }>();
expect(checks.results).toEqual([{ ok: 1 }]);
});
});
});
+6
View File
@@ -1,7 +1,13 @@
import { exports as worker } from 'cloudflare:workers';
import { describe, expect, it } from 'vitest';
import { shouldRunFiveMinuteWork } from '../src/worker';
describe('uptime monitoring Worker', () => {
it('runs housekeeping only on five-minute UTC boundaries', () => {
expect(shouldRunFiveMinuteWork(Date.parse('2026-09-02T12:10:00Z'))).toBe(true);
expect(shouldRunFiveMinuteWork(Date.parse('2026-09-02T12:11:00Z'))).toBe(false);
});
it('returns a successful D1 health response', async () => {
const response = await worker.default.fetch('https://example.com/api/health');
const body = await response.json<{
+21 -2
View File
@@ -173,6 +173,19 @@ describe('monitor API', () => {
expect(list.monitors[0]).toMatchObject({ id: created.monitor.id, url: 'https://example.com/health' });
});
it('jitters a new monitor across its first interval', async () => {
vi.spyOn(Math, 'random').mockReturnValue(0.5);
const before = Date.now();
const created = await (
await createMonitor(await authenticatedCookie(), { intervalSeconds: 600 })
).json<{
monitor: { lastCheckedAt: string };
}>();
const lastCheckedAt = Date.parse(created.monitor.lastCheckedAt);
expect(lastCheckedAt).toBeGreaterThanOrEqual(before - 301_000);
expect(lastCheckedAt).toBeLessThanOrEqual(Date.now() - 299_000);
});
it.each([
[{ url: 'file:///etc/passwd' }, 'Enter a valid http or https URL'],
[{ intervalSeconds: 60 }, 'intervalSeconds must be an integer between 300 and 86400'],
@@ -349,7 +362,7 @@ describe('monitor API', () => {
expect(stats.windows['24h']).toEqual({ uptimePct: 50, totalChecks: 2, upChecks: 1, avgLatencyMs: 200, incidentCount: 1 });
});
it('combines daily rollups with the current partial day for long-range stats', async () => {
it('combines daily rollups with the current partial day for seven-day and long-range stats', async () => {
const cookie = await authenticatedCookie();
const created = await (await createMonitor(cookie)).json<{ monitor: { id: number } }>();
const id = created.monitor.id;
@@ -371,8 +384,14 @@ describe('monitor API', () => {
const response = await apiFetch(`/api/monitors/${id}/stats`, 'GET', cookie);
const stats = await response.json<{
windows: { '30d': { uptimePct: number; totalChecks: number; upChecks: number; avgLatencyMs: number } };
windows: Record<'7d' | '30d', { uptimePct: number; totalChecks: number; upChecks: number; avgLatencyMs: number }>;
}>();
expect(stats.windows['7d']).toMatchObject({
uptimePct: 80,
totalChecks: 10,
upChecks: 8,
avgLatencyMs: 120,
});
expect(stats.windows['30d']).toMatchObject({
uptimePct: 80,
totalChecks: 10,
+32
View File
@@ -64,4 +64,36 @@ describe('daily monitor rollups', () => {
max_latency_ms: 300,
});
});
it('rolls up checks correctly when ids and timestamps are out of order', async () => {
const createdAt = Date.parse('2026-08-28T00:05:00Z');
const inserted = await env.DB.prepare(
"INSERT INTO monitors (name, url, method, expected_status, interval_seconds, timeout_ms, enabled, alerts_enabled, created_at, updated_at) VALUES ('API', 'https://example.com', 'GET', 200, 300, 10000, 1, 1, ?, ?)",
)
.bind(createdAt, createdAt)
.run();
const monitorId = Number(inserted.meta.last_row_id);
// The newer timestamp deliberately receives the lower primary key.
await env.DB.batch([
env.DB.prepare('INSERT INTO checks (monitor_id, ok, latency_ms, checked_at) VALUES (?, 1, 100, ?)').bind(
monitorId,
Date.parse('2026-08-28T12:00:00Z'),
),
env.DB.prepare('INSERT INTO checks (monitor_id, ok, latency_ms, checked_at) VALUES (?, 0, 300, ?)').bind(
monitorId,
Date.parse('2026-08-27T12:00:00Z'),
),
]);
await runDailyRollup(env, new Date('2026-08-28T00:05:00Z'));
await runDailyRollup(env, new Date('2026-08-29T00:05:00Z'));
const rows = await env.DB.prepare('SELECT day, total_checks, up_checks FROM monitor_daily_stats WHERE monitor_id = ? ORDER BY day')
.bind(monitorId)
.all();
expect(rows.results).toEqual([
{ day: Date.parse('2026-08-27T00:00:00Z'), total_checks: 1, up_checks: 0 },
{ day: Date.parse('2026-08-28T00:00:00Z'), total_checks: 1, up_checks: 1 },
]);
});
});
+16
View File
@@ -1,7 +1,9 @@
import { describe, expect, it } from 'vitest';
import {
DEFAULT_CHECK_RUN_CONFIG,
DEFAULT_RUN_LIMITS,
DEFAULT_STATUS_CACHE_SECONDS,
resolveCheckRunConfig,
resolveRunLimits,
resolveStatusCacheSeconds,
} from '../src/worker/lib/runtime-config';
@@ -36,6 +38,20 @@ describe('resolveRunLimits', () => {
});
});
describe('resolveCheckRunConfig', () => {
it('defaults to free-tier-safe check sizing', () => {
expect(resolveCheckRunConfig(asEnv({}))).toEqual(DEFAULT_CHECK_RUN_CONFIG);
});
it('accepts positive integer overrides and rejects invalid values', () => {
expect(resolveCheckRunConfig(asEnv({ MAX_MONITORS_PER_RUN: '25', CHECK_CONCURRENCY: '4' }))).toEqual({
maxMonitorsPerRun: 25,
concurrency: 4,
});
expect(resolveCheckRunConfig(asEnv({ MAX_MONITORS_PER_RUN: '0', CHECK_CONCURRENCY: '2.5' }))).toEqual(DEFAULT_CHECK_RUN_CONFIG);
});
});
describe('resolveStatusCacheSeconds', () => {
it('defaults to DEFAULT_STATUS_CACHE_SECONDS', () => {
expect(resolveStatusCacheSeconds(asEnv({}))).toBe(DEFAULT_STATUS_CACHE_SECONDS);
+1 -1
View File
@@ -10,7 +10,7 @@ export default defineConfig({
miniflare: {
// STATUS_CACHE_SECONDS '0' disables the public-status edge cache so assertions see
// fresh D1 reads instead of a response cached by an earlier test.
bindings: { TEST_MIGRATIONS: migrations, STATUS_CACHE_SECONDS: '0' },
bindings: { TEST_MIGRATIONS: migrations, STATUS_CACHE_SECONDS: '0', PUBLIC_BASE_URL: '' },
},
}),
],
+1 -1
View File
@@ -11,7 +11,7 @@
"run_worker_first": ["/api/*", "/", "/incidents/*", "/robots.txt", "/sitemap.xml"],
},
"triggers": {
"crons": ["*/5 * * * *", "5 0 * * *"],
"crons": ["* * * * *", "5 0 * * *"],
},
"keep_vars": true,
"vars": {