refactor tier 3

apps/api/src/jobs/rollup.ts

@@ -0,0 +1,92 @@
+import sql from "../db";
+
+// Aggregates raw pings into monitor_uptime_rollup so status pages and dashboard
+// widgets can compute uptime % over arbitrary windows without ever scanning the
+// pings table at read time. Three resolutions: hourly, daily, weekly.
+//
+// Each pass aggregates the *current* bucket only. The query is bounded by the
+// bucket size, not the table size, so it's cheap regardless of history depth.
+
+type BucketType = "hourly" | "daily" | "weekly";
+
+const BUCKET_TRUNC: Record<BucketType, string> = {
+  hourly: "hour",
+  daily:  "day",
+  weekly: "week",
+};
+
+async function rollupCurrent(bucket: BucketType): Promise<number> {
+  const trunc = BUCKET_TRUNC[bucket];
+  // Aggregate the bucket containing now(). ON CONFLICT updates if the row exists,
+  // so this is safe to run repeatedly during the bucket's lifetime.
+  const result = await sql`
+    INSERT INTO monitor_uptime_rollup (monitor_id, region, bucket_type, bucket_start, total, up_count, avg_latency)
+    SELECT
+      monitor_id,
+      COALESCE(region, 'default') AS region,
+      ${bucket} AS bucket_type,
+      date_trunc(${trunc}, checked_at) AS bucket_start,
+      count(*)::int AS total,
+      count(*) FILTER (WHERE up)::int AS up_count,
+      avg(latency_ms)::real AS avg_latency
+    FROM pings
+    WHERE checked_at >= date_trunc(${trunc}, now())
+    GROUP BY monitor_id, COALESCE(region, 'default'), date_trunc(${trunc}, checked_at)
+    ON CONFLICT (monitor_id, region, bucket_type, bucket_start) DO UPDATE SET
+      total       = EXCLUDED.total,
+      up_count    = EXCLUDED.up_count,
+      avg_latency = EXCLUDED.avg_latency
+  `;
+  return result.count ?? 0;
+}
+
+// Walk back N units and aggregate any buckets that don't exist yet. Used at
+// startup so a freshly-deployed system has historical data immediately.
+async function backfillRecent(bucket: BucketType, units: number): Promise<number> {
+  const trunc = BUCKET_TRUNC[bucket];
+  // Build the interval string entirely in JS so postgres.js binds a single text
+  // parameter. Avoids the int || text type-mismatch trap inside SQL.
+  const intervalLiteral = `${units} ${trunc}s`;
+  const result = await sql`
+    INSERT INTO monitor_uptime_rollup (monitor_id, region, bucket_type, bucket_start, total, up_count, avg_latency)
+    SELECT
+      monitor_id,
+      COALESCE(region, 'default') AS region,
+      ${bucket} AS bucket_type,
+      date_trunc(${trunc}, checked_at) AS bucket_start,
+      count(*)::int AS total,
+      count(*) FILTER (WHERE up)::int AS up_count,
+      avg(latency_ms)::real AS avg_latency
+    FROM pings
+    WHERE checked_at >= date_trunc(${trunc}, now()) - ${intervalLiteral}::interval
+    GROUP BY monitor_id, COALESCE(region, 'default'), date_trunc(${trunc}, checked_at)
+    ON CONFLICT (monitor_id, region, bucket_type, bucket_start) DO NOTHING
+  `;
+  return result.count ?? 0;
+}
+
+let started = false;
+
+export async function startRollupJob() {
+  if (started) return;
+  started = true;
+
+  // Startup backfill: gives existing accounts immediate history without waiting
+  // for the periodic timers to wander backwards. Cheap because pings is indexed
+  // on checked_at and the units are bounded.
+  try {
+    const [h, d, w] = await Promise.all([
+      backfillRecent("hourly", 48),    // 48h of hourly buckets
+      backfillRecent("daily",  90),    // 90 days of daily buckets
+      backfillRecent("weekly", 26),    // 26 weeks of weekly buckets
+    ]);
+    console.log(`[rollup] backfilled rows: hourly=${h} daily=${d} weekly=${w}`);
+  } catch (e) {
+    console.warn("[rollup] backfill failed:", e);
+  }
+
+  // Periodic refreshes for the *current* bucket of each resolution.
+  setInterval(() => { rollupCurrent("hourly").catch((e) => console.warn("[rollup] hourly failed:", e)); },  5 * 60 * 1000);
+  setInterval(() => { rollupCurrent("daily").catch((e) => console.warn("[rollup] daily failed:",  e)); }, 30 * 60 * 1000);
+  setInterval(() => { rollupCurrent("weekly").catch((e) => console.warn("[rollup] weekly failed:", e)); },  6 * 60 * 60 * 1000);
+}