/** * GET /api/status/workers * Heartbeat snapshot for the three in-process workers: * • analyzer — analyzer_jobs * • rmm — rmm_executions * • sync — sync_history (proxy for the scheduler; one row per * triggered task) * * For each: last activity timestamp, in-flight count, last-1h success/ * failure totals, and 24 hourly buckets of activity (success vs * failure counts) for sparkline rendering. */ import { NextResponse } from 'next/server'; import { requireAuth } from '@/lib/auth-utils'; import postgresClient from '@/lib/services/postgres-client'; export interface ActivityBucket { /** Hour-truncated UTC timestamp. */ hour: string; success: number; failure: number; } export interface WorkerSnapshot { name: string; lastActivity: string | null; inFlight: number; oneHour: { success: number; failure: number }; activity24h: ActivityBucket[]; } interface SnapshotRow { last_activity: string | null; in_flight: string; ok_1h: string; fail_1h: string; } interface BucketRow { hour: string; ok: string; fail: string; } /** * Build a 24-row series with zero-fill so the sparkline always has the * same number of points (even when the worker is idle). Generated as a * date_trunc('hour') series joined to the source table's status histogram. */ function bucketsCte(table: string, statusCol: string, tsCol: string, okValues: string[], failValues: string[]) { const okIn = okValues.map((v) => `'${v}'`).join(','); const failIn = failValues.map((v) => `'${v}'`).join(','); return ` WITH hours AS ( SELECT generate_series( date_trunc('hour', NOW()) - INTERVAL '23 hours', date_trunc('hour', NOW()), INTERVAL '1 hour' ) AS h ) SELECT hours.h::text AS hour, COALESCE(SUM(CASE WHEN ${statusCol} IN (${okIn}) THEN 1 ELSE 0 END), 0)::text AS ok, COALESCE(SUM(CASE WHEN ${statusCol} IN (${failIn}) THEN 1 ELSE 0 END), 0)::text AS fail FROM hours LEFT JOIN ${table} t ON date_trunc('hour', t.${tsCol}) = hours.h AND t.${tsCol} >= NOW() - INTERVAL '24 hours' GROUP BY hours.h ORDER BY hours.h `; } export async function GET() { const { error } = await requireAuth(); if (error) return error; const [ analyzerRes, analyzerBuckets, rmmRes, rmmBuckets, syncRes, syncBuckets, ] = await Promise.all([ postgresClient.query( `SELECT GREATEST(MAX(queued_at), MAX(started_at), MAX(finished_at))::text AS last_activity, COUNT(*) FILTER (WHERE status IN ('queued','fetching','triaging','itglue','analyzing','deep_review'))::text AS in_flight, COUNT(*) FILTER (WHERE status = 'complete' AND finished_at >= NOW() - INTERVAL '1 hour')::text AS ok_1h, COUNT(*) FILTER (WHERE status = 'failed' AND finished_at >= NOW() - INTERVAL '1 hour')::text AS fail_1h FROM analyzer_jobs`, ), postgresClient.query( bucketsCte('analyzer_jobs', 'status', 'finished_at', ['complete'], ['failed']), ), postgresClient.query( `SELECT GREATEST(MAX(queued_at), MAX(started_at), MAX(completed_at))::text AS last_activity, COUNT(*) FILTER (WHERE status IN ('queued','running'))::text AS in_flight, COUNT(*) FILTER (WHERE status = 'complete' AND completed_at >= NOW() - INTERVAL '1 hour')::text AS ok_1h, COUNT(*) FILTER (WHERE status IN ('failed','timeout') AND completed_at >= NOW() - INTERVAL '1 hour')::text AS fail_1h FROM rmm_executions`, ), postgresClient.query( bucketsCte('rmm_executions', 'status', 'completed_at', ['complete'], ['failed', 'timeout']), ), postgresClient.query<{ last_run: string | null; ok_1h: string; fail_1h: string }>( `SELECT MAX(last_run)::text AS last_run, COUNT(*) FILTER (WHERE last_status = 'success' AND last_run >= NOW() - INTERVAL '1 hour')::text AS ok_1h, COUNT(*) FILTER (WHERE last_status = 'failed' AND last_run >= NOW() - INTERVAL '1 hour')::text AS fail_1h FROM sync_schedules WHERE is_enabled = true`, ), postgresClient.query( bucketsCte('sync_history', 'status', 'started_at', ['completed'], ['failed']), ), ]); const toBuckets = (rows: BucketRow[]): ActivityBucket[] => rows.map((r) => ({ hour: r.hour, success: parseInt(r.ok, 10), failure: parseInt(r.fail, 10), })); const a = analyzerRes.rows[0]; const r = rmmRes.rows[0]; const s = syncRes.rows[0]; const workers: WorkerSnapshot[] = [ { name: 'Analyzer', lastActivity: a?.last_activity ?? null, inFlight: parseInt(a?.in_flight ?? '0', 10), oneHour: { success: parseInt(a?.ok_1h ?? '0', 10), failure: parseInt(a?.fail_1h ?? '0', 10), }, activity24h: toBuckets(analyzerBuckets.rows), }, { name: 'RMM Overshell', lastActivity: r?.last_activity ?? null, inFlight: parseInt(r?.in_flight ?? '0', 10), oneHour: { success: parseInt(r?.ok_1h ?? '0', 10), failure: parseInt(r?.fail_1h ?? '0', 10), }, activity24h: toBuckets(rmmBuckets.rows), }, { name: 'Sync scheduler', lastActivity: s?.last_run ?? null, inFlight: 0, oneHour: { success: parseInt(s?.ok_1h ?? '0', 10), failure: parseInt(s?.fail_1h ?? '0', 10), }, activity24h: toBuckets(syncBuckets.rows), }, ]; return NextResponse.json({ workers }); }