monitor

index.ts

 1import { isAuthorized, unauthorized } from "./auth";
 2import { errorPage, historyPage, homePage, html } from "./html";
 3import { MonitorDO, REGIONS, endpoints, type Check, type Region, type RegionSnapshot } from "./monitor";
 4
 5export { MonitorDO };
 6
 7const PAGE_SIZE = 50;
 8const RPC_TIMEOUT_MS = 5000;
 9
10export default {
11  async fetch(req, env): Promise<Response> {
12    if (!(await isAuthorized(req, env))) return unauthorized();
13    const url = new URL(req.url);
14    if (req.method !== "GET") return new Response("Method not allowed", { status: 405 });
15    if (url.pathname === "/") return home(env);
16    if (url.pathname === "/endpoint") return history(env, url.searchParams);
17    return errorPage(404, "Not found.");
18  },
19
20  // Watchdog: creates each regional monitor on first run and restarts any stopped alarm loop.
21  async scheduled(_controller, env): Promise<void> {
22    const results = await Promise.allSettled(REGIONS.map((r) => monitor(env, r).ensureAlarm(r)));
23    results.forEach((r, i) => {
24      if (r.status === "rejected") console.error("ensureAlarm failed", { region: REGIONS[i], error: String(r.reason) });
25    });
26  },
27} satisfies ExportedHandler<Env>;
28
29export function monitor(env: Env, region: Region): DurableObjectStub<MonitorDO> {
30  const ns = env.MONITOR as DurableObjectNamespace<MonitorDO>;
31  return ns.get(ns.idFromName(region), { locationHint: region });
32}
33
34/** Calls every regional monitor in parallel; a region that errors or is slow yields null. */
35async function fanOut<T>(env: Env, call: (stub: DurableObjectStub<MonitorDO>, region: Region) => Promise<T>): Promise<(T | null)[]> {
36  return Promise.all(
37    REGIONS.map(async (region) => {
38      try {
39        return await Promise.race([
40          call(monitor(env, region), region),
41          new Promise<never>((_, reject) => setTimeout(() => reject(new Error("timeout")), RPC_TIMEOUT_MS)),
42        ]);
43      } catch (e) {
44        console.error("region RPC failed", { region, error: String(e) });
45        return null;
46      }
47    }),
48  );
49}
50
51async function home(env: Env): Promise<Response> {
52  const snapshots = await fanOut<RegionSnapshot>(env, (stub, region) => stub.latest(region));
53  return html(homePage(env, endpoints(env), snapshots, Date.now()));
54}
55
56async function history(env: Env, params: URLSearchParams): Promise<Response> {
57  const endpoint = params.get("url") ?? "";
58  if (!endpoints(env).includes(endpoint)) return errorPage(404, "Unknown endpoint.");
59  const regionParam = params.get("region") ?? "all";
60  const region = (REGIONS as readonly string[]).includes(regionParam) ? (regionParam as Region) : "all";
61  const failuresOnly = params.get("failures") === "1";
62  const beforeParam = Number(params.get("before"));
63  const before = Number.isFinite(beforeParam) && beforeParam > 0 ? beforeParam : null;
64
65  const results = await fanOut<Check[]>(env, (stub, r) =>
66    region === "all" || region === r
67      ? stub.history(endpoint, before ?? Number.MAX_SAFE_INTEGER, PAGE_SIZE + 1, failuresOnly)
68      : Promise.resolve([]),
69  );
70  const { checks, nextBefore } = mergePage(results.map((r) => r ?? []), PAGE_SIZE);
71  const unreachable = REGIONS.filter((_, i) => results[i] === null);
72  return html(historyPage({ endpoint, region, failuresOnly, before, checks, nextBefore, unreachable }));
73}
74
75/**
76 * Merges per-region pages (each newest first, each fetched with limit > pageSize) into one page.
77 * Rows sharing the cut-off timestamp stay together so the `ts < before` cursor never skips any.
78 */
79export function mergePage(lists: Check[][], pageSize: number): { checks: Check[]; nextBefore: number | null } {
80  const all = lists.flat().sort((a, b) => b.ts - a.ts);
81  if (all.length <= pageSize) return { checks: all, nextBefore: null };
82  let end = pageSize;
83  while (end < all.length && all[end].ts === all[end - 1].ts) end++;
84  return { checks: all.slice(0, end), nextBefore: all[end - 1].ts };
85}