← back to Homesonspec

apps/admin/src/app/api/ingestion-stats/route.ts

80 lines

import { prisma } from "@homesonspec/database";
import { NextResponse } from "next/server";

export const dynamic = "force-dynamic";

// Live ingestion telemetry for the /ingestion dashboard. All counts cast ::int so the
// JSON has no BigInt. Read-only; safe to poll frequently.
export async function GET() {
  const [byStatus, throughput, newHomes, runs, health, staged, topSources] = await Promise.all([
    prisma.$queryRawUnsafe<{ status: string; n: number }[]>(
      `SELECT status, count(*)::int AS n FROM "InventoryHome" GROUP BY status ORDER BY n DESC`,
    ),
    prisma.$queryRawUnsafe<any[]>(
      `SELECT
         count(*)::int AS total,
         count(*) FILTER (WHERE "updatedAt" > now() - interval '1 min')::int  AS m1,
         count(*) FILTER (WHERE "updatedAt" > now() - interval '5 min')::int  AS m5,
         count(*) FILTER (WHERE "updatedAt" > now() - interval '30 min')::int AS m30,
         count(*) FILTER (WHERE "updatedAt" > now() - interval '60 min')::int AS m60,
         max("updatedAt") AS newest
       FROM "InventoryHome"`,
    ),
    // newly-DISCOVERED homes (createdAt) vs merely refreshed — the growth signal
    prisma.$queryRawUnsafe<any[]>(
      `SELECT
         count(*) FILTER (WHERE "createdAt" > now() - interval '30 min')::int AS n30,
         count(*) FILTER (WHERE "createdAt" > now() - interval '60 min')::int AS n60
       FROM "InventoryHome"`,
    ).catch(() => [{ n30: null, n60: null }]),
    prisma.$queryRawUnsafe<any[]>(
      `SELECT
         count(*) FILTER (WHERE "startedAt" > now() - interval '10 min')::int AS r10,
         count(*) FILTER (WHERE "startedAt" > now() - interval '60 min')::int AS r60,
         EXTRACT(EPOCH FROM (now() - max("startedAt")))::int AS newest_age_s
       FROM "SourceRun"`,
    ),
    // ingestion error signal over the last hour
    prisma.$queryRawUnsafe<any[]>(
      `SELECT
         count(*) FILTER (WHERE "ok" = false AND "startedAt" > now() - interval '60 min')::int AS err_runs,
         count(*) FILTER (WHERE "startedAt" > now() - interval '60 min')::int AS runs,
         COALESCE(sum("errorCount") FILTER (WHERE "startedAt" > now() - interval '60 min'),0)::int AS err_count
       FROM "SourceRun"`,
    ).catch(() => [{ err_runs: 0, runs: 0, err_count: 0 }]),
    prisma.$queryRawUnsafe<any[]>(
      `SELECT count(*)::int AS total,
         count(*) FILTER (WHERE "updatedAt" > now() - interval '30 min')::int AS s30
       FROM "StagedRecord"`,
    ),
    // per-source: recent runs, homes published, errors, and how long since its last run
    prisma.$queryRawUnsafe<any[]>(
      // LEFT JOIN from active sources so a source that STOPPED running entirely still
      // appears (as a stale/absent row) instead of silently vanishing — the catastrophic
      // failure this table exists to catch (Cody-gated fix, 2026-07-28). Stalest first.
      `SELECT sr."key" AS key,
         count(run."id") FILTER (WHERE run."startedAt" > now() - interval '60 min')::int AS runs,
         COALESCE(sum(run."published") FILTER (WHERE run."startedAt" > now() - interval '60 min'),0)::int AS published,
         count(run."id") FILTER (WHERE run."ok" = false AND run."startedAt" > now() - interval '60 min')::int AS failed_runs,
         COALESCE(sum(run."errorCount") FILTER (WHERE run."startedAt" > now() - interval '60 min'),0)::int AS err_records,
         EXTRACT(EPOCH FROM (now() - max(run."startedAt")))::int AS last_age_s
       FROM "SourceRegistry" sr
       LEFT JOIN "SourceRun" run ON run."sourceId" = sr."id"
       WHERE sr."active" = true
       GROUP BY sr."key"
       ORDER BY last_age_s DESC NULLS FIRST LIMIT 14`,
    ).catch(() => [] as any[]),
  ]);

  return NextResponse.json({
    ts: new Date().toISOString(),
    byStatus,
    throughput: throughput[0] ?? {},
    newHomes: newHomes[0] ?? {},
    runs: runs[0] ?? {},
    health: health[0] ?? {},
    staged: staged[0] ?? {},
    topSources,
  });
}