wulf-pulse/app/api/status/workers/route.ts

/**
 * 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<SnapshotRow>(
      `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<BucketRow>(
      bucketsCte('analyzer_jobs', 'status', 'finished_at', ['complete'], ['failed']),
    ),
    postgresClient.query<SnapshotRow>(
      `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<BucketRow>(
      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<BucketRow>(
      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 });
}