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.
15 kB · 506 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507import { getLogger } from "@logtape/logtape";import { db } from "@openstatus/db";import { listExternalServices } from "@openstatus/services/external-service";import type { ExternalServiceRow } from "@openstatus/services/external-service";import { type UpsertExternalComponentInput, upsertExternalComponentsForService,} from "@openstatus/services/external-service-component";import { type UpsertExternalIncidentInput, upsertExternalIncidentsForService,} from "@openstatus/services/external-service-incident";import { FetchError, fetchers } from "@openstatus/status-fetcher";import type { NormalizedComponent, NormalizedIncident, StatusFetcher, StatusPageEntry, StatusResult,} from "@openstatus/status-fetcher";import { OSTinybird } from "@openstatus/tinybird";import { Effect } from "effect";import type { Context } from "hono";
import { env } from "../env";import { reportBackgroundError, reportFetchFailure, runSentryCron,} from "../lib/sentry";
const logger = getLogger(["workflow", "external-status"]);
const tb = new OSTinybird(env().TINY_BIRD_API_KEY);
// 10 per phase × 3 phases = peak 30 concurrent HTTP requests upstream; keeps// Atlassian/Incident.io CDNs comfortable while still parallelising heavily.const PHASE_CONCURRENCY = 10;
function toStatusPageEntry(row: ExternalServiceRow): StatusPageEntry { return { id: row.slug, name: row.name, url: row.url, status_page_url: row.statusPageUrl, provider: row.provider, industry: row.industry, description: row.description ?? undefined, api_config: row.apiConfig ?? undefined, };}
type Snapshot = { id: string; indicator: string; status: string; status_message: string; fetched_at: number; updated_at: number; time_zone: string;};
function buildSnapshot(args: { entry: StatusPageEntry; result: StatusResult; fetchedAt: number;}): Snapshot { const { entry, result, fetchedAt } = args; return { id: entry.id, indicator: result.severity, status: result.status, status_message: result.description, fetched_at: fetchedAt, updated_at: result.updated_at, time_zone: result.timezone ?? "", };}
function toUpsertInput( incident: NormalizedIncident,): UpsertExternalIncidentInput { return { providerIncidentId: incident.providerIncidentId, name: incident.name, status: incident.status, impact: incident.impact, shortlink: incident.shortlink, startedAt: incident.startedAt, createdAt: incident.createdAt, resolvedAt: incident.resolvedAt, affectedComponentIds: incident.affectedComponentIds, raw: incident.raw, };}
type ComponentSnapshot = { component_id: string; external_service_id: number; indicator: string; status: string; fetched_at: number;};
function toComponentUpsertInput( component: NormalizedComponent,): UpsertExternalComponentInput { return { upstreamComponentId: component.upstreamComponentId, name: component.name, description: component.description, groupName: component.groupName, position: component.position, indicator: component.severity, status: component.status, };}
type PhaseCounts = { successCount: number; failureCount: number; skippedCount: number; total: number;};
type StatusPhaseOutcome = | { kind: "ok"; snapshot: Snapshot } | { kind: "no-fetcher"; slug: string } | { kind: "fail"; slug: string; reason: string };
type IncidentPhaseOutcome = | { kind: "ok"; slug: string; count: number } | { kind: "skip"; slug: string } | { kind: "fail"; slug: string; reason: string };
type ComponentPhaseOutcome = | { kind: "ok"; slug: string; snapshots: ComponentSnapshot[] } | { kind: "skip"; slug: string } | { kind: "fail"; slug: string; reason: string };
type Triplet = { row: ExternalServiceRow; entry: StatusPageEntry; fetcher: StatusFetcher | null;};
function runStatusPhase( triplets: Triplet[], fetchedAt: number,): Effect.Effect<StatusPhaseOutcome[]> { return Effect.forEach( triplets, ({ entry, fetcher }) => { if (!fetcher) { return Effect.succeed<StatusPhaseOutcome>({ kind: "no-fetcher", slug: entry.id, }); } return fetcher.fetch(entry).pipe( Effect.map( (result): StatusPhaseOutcome => ({ kind: "ok", snapshot: buildSnapshot({ entry, result, fetchedAt }), }), ), Effect.catchAll((err: FetchError) => Effect.sync<StatusPhaseOutcome>(() => { reportFetchFailure({ phase: "status", slug: entry.id, error: err }); return { kind: "fail", slug: entry.id, reason: err.message }; }), ), ); }, { concurrency: PHASE_CONCURRENCY }, );}
function runIncidentPhase( triplets: Triplet[], tickStartedAt: Date,): Effect.Effect<IncidentPhaseOutcome[]> { return Effect.forEach( triplets, ({ row, entry, fetcher }) => { if (!fetcher || !fetcher.fetchIncidents) { return Effect.succeed<IncidentPhaseOutcome>({ kind: "skip", slug: entry.id, }); } return fetcher.fetchIncidents(entry).pipe( Effect.flatMap((incidents) => Effect.tryPromise({ try: () => upsertExternalIncidentsForService({ ctx: { db }, externalServiceId: row.id, incidents: incidents.map(toUpsertInput), now: tickStartedAt, }), catch: (e) => new FetchError({ url: entry.status_page_url, fetcherName: fetcher.name, entryId: entry.id, cause: e instanceof Error ? e : new Error(String(e)), }), }).pipe( Effect.map( (result): IncidentPhaseOutcome => ({ kind: "ok", slug: entry.id, count: result.upserted, }), ), ), ), Effect.catchAll((err: FetchError) => Effect.sync<IncidentPhaseOutcome>(() => { reportFetchFailure({ phase: "incidents", slug: entry.id, error: err, }); return { kind: "fail", slug: entry.id, reason: err.message }; }), ), ); }, { concurrency: PHASE_CONCURRENCY }, );}
function runComponentPhase( triplets: Triplet[], tickStartedAt: Date, fetchedAt: number,): Effect.Effect<ComponentPhaseOutcome[]> { return Effect.forEach( triplets, ({ row, entry, fetcher }) => { if (!fetcher || !fetcher.fetchComponents) { return Effect.succeed<ComponentPhaseOutcome>({ kind: "skip", slug: entry.id, }); } return fetcher.fetchComponents(entry).pipe( Effect.flatMap((components) => Effect.tryPromise({ try: () => upsertExternalComponentsForService({ ctx: { db }, externalServiceId: row.id, components: components.map(toComponentUpsertInput), now: tickStartedAt, }), catch: (e) => new FetchError({ url: entry.status_page_url, fetcherName: fetcher.name, entryId: entry.id, cause: e instanceof Error ? e : new Error(String(e)), }), }).pipe( Effect.map((result): ComponentPhaseOutcome => { // History rows key on our PK, so the upstream→PK map from the // upsert turns each normalized component into a snapshot. const byUpstream = new Map( components.map((c) => [c.upstreamComponentId, c]), ); const snapshots: ComponentSnapshot[] = []; for (const upserted of result.upserted) { const c = byUpstream.get(upserted.upstreamComponentId); if (!c) continue; snapshots.push({ component_id: String(upserted.id), external_service_id: row.id, indicator: c.severity, status: c.status, fetched_at: fetchedAt, }); } return { kind: "ok", slug: entry.id, snapshots }; }), ), ), Effect.catchAll((err: FetchError) => Effect.sync<ComponentPhaseOutcome>(() => { reportFetchFailure({ phase: "components", slug: entry.id, error: err, }); return { kind: "fail", slug: entry.id, reason: err.message }; }), ), ); }, { concurrency: PHASE_CONCURRENCY }, );}
function summarizeStatus(outcomes: StatusPhaseOutcome[]): { counts: PhaseCounts; snapshots: Snapshot[];} { const snapshots: Snapshot[] = []; let successCount = 0; let failureCount = 0; let skippedCount = 0; for (const o of outcomes) { if (o.kind === "ok") { snapshots.push(o.snapshot); successCount++; } else if (o.kind === "no-fetcher") { skippedCount++; logger.warn("external-status status: no fetcher matches slug={slug}", { slug: o.slug, }); } else { failureCount++; logger.warn( "external-status status: fetch failed for slug={slug}: {reason}", { slug: o.slug, reason: o.reason, }, ); } } return { counts: { successCount, failureCount, skippedCount, total: outcomes.length, }, snapshots, };}
function summarizeIncidents(outcomes: IncidentPhaseOutcome[]): PhaseCounts { let successCount = 0; let failureCount = 0; let skippedCount = 0; for (const o of outcomes) { if (o.kind === "ok") { successCount++; } else if (o.kind === "skip") { skippedCount++; } else { failureCount++; logger.warn( "external-status incidents: failed for slug={slug}: {reason}", { slug: o.slug, reason: o.reason, }, ); } } return { successCount, failureCount, skippedCount, total: outcomes.length, };}
function summarizeComponents(outcomes: ComponentPhaseOutcome[]): { counts: PhaseCounts; snapshots: ComponentSnapshot[];} { const snapshots: ComponentSnapshot[] = []; let successCount = 0; let failureCount = 0; let skippedCount = 0; for (const o of outcomes) { if (o.kind === "ok") { successCount++; snapshots.push(...o.snapshots); } else if (o.kind === "skip") { skippedCount++; } else { failureCount++; logger.warn( "external-status components: failed for slug={slug}: {reason}", { slug: o.slug, reason: o.reason, }, ); } } return { counts: { successCount, failureCount, skippedCount, total: outcomes.length, }, snapshots, };}
function buildTriplets(services: ExternalServiceRow[]): Triplet[] { return services.map((row) => { const entry = toStatusPageEntry(row); const fetcher = fetchers.find((f) => f.canHandle(entry)) ?? null; return { row, entry, fetcher }; });}
export async function runExternalStatusTick(): Promise<{ status: PhaseCounts; incidents: PhaseCounts; components: PhaseCounts;}> { const services = await listExternalServices({ ctx: { db } });
const triplets = buildTriplets(services); const tickStartedAt = new Date();
// The three phases are intentionally independent: each hits a different // upstream endpoint and store, so a failed status fetch must not suppress a // service's components (or vice versa). A tick can therefore persist // component history for a service whose status snapshot failed that tick. const [statusOutcomes, incidentOutcomes, componentOutcomes] = await Effect.runPromise( Effect.all( [ runStatusPhase(triplets, tickStartedAt.getTime()), runIncidentPhase(triplets, tickStartedAt), runComponentPhase(triplets, tickStartedAt, tickStartedAt.getTime()), ], { concurrency: 50 }, ), );
const status = summarizeStatus(statusOutcomes); const incidents = summarizeIncidents(incidentOutcomes); const components = summarizeComponents(componentOutcomes);
if (status.snapshots.length > 0) { await tb.publishExternalStatus(status.snapshots); } if (components.snapshots.length > 0) { await tb.publishExternalStatusComponent(components.snapshots); }
return { status: status.counts, incidents, components: components.counts };}
export async function handleExternalStatusCron(c: Context) { const { cronCompleted, cronFailed } = runSentryCron("external-status");
// Background chain: must not capture `c` or anything derived from it // (e.g. via getSentry(c)). The handler returns 200 before this resolves, and // a captured per-request Sentry hub stays pinned across retries — see // apps/workflows/plan.md. void Effect.runPromise( Effect.tryPromise({ try: () => runExternalStatusTick(), catch: (e) => new Error( `external-status tick failed: ${e instanceof Error ? e.message : String(e)}`, ), }).pipe( Effect.tap((res) => Effect.sync(() => { logger.info( "external-status tick complete: status={statusOk}/{statusTotal} ({statusFail} failures, {statusSkip} skipped), incidents={incOk}/{incTotal} ({incFail} failures, {incSkip} skipped), components={compOk}/{compTotal} ({compFail} failures, {compSkip} skipped)", { statusOk: res.status.successCount, statusTotal: res.status.total, statusFail: res.status.failureCount, statusSkip: res.status.skippedCount, incOk: res.incidents.successCount, incTotal: res.incidents.total, incFail: res.incidents.failureCount, incSkip: res.incidents.skippedCount, compOk: res.components.successCount, compTotal: res.components.total, compFail: res.components.failureCount, compSkip: res.components.skippedCount, }, ); void cronCompleted(); }), ), Effect.catchAll((e) => Effect.sync(() => { logger.error("external-status tick errored: {message}", { message: e.message, }); void reportBackgroundError(e.message); void cronFailed(); }), ), ), );
return c.json({ success: true }, 200);}