godcrm/backend/routes/v3/ai-agents/autopilotDashboardController.js

/**
 * Autopilot Dashboard Controller
 *
 * GET /autopilot/dashboard — Aggregated view of agent status, job pipeline, throughput.
 * POST /autopilot/jobs/:id/cancel — Cancel a running job
 * POST /autopilot/jobs/:id/reassign — Reassign a job to another agent
 */

import { Router } from 'express';
import { dbAll, dbGet, isPostgres } from '../../../database/connection.js';
import { success, error, notFound, badRequest } from '../../../utils/response.js';
import { apiLogger } from '../../../utils/logger.js';
import AgentJobService from '../../../services/AgentJobService.js';

const router = Router();

/**
 * GET /autopilot/dashboard — Main dashboard payload
 *
 * Returns:
 * - agents: list of agents with current job info
 * - pipeline: { backlog, in_progress, review, done, failed } counts
 * - recentJobs: last 50 jobs with status/timing
 * - throughput: completed/failed counts for last 24h, 7d
 * - errors: recent failed jobs with error details
 */
router.get('/autopilot/dashboard', async (req, res) => {
  try {
    const now = isPostgres() ? 'NOW()' : "datetime('now')";
    const interval24h = isPostgres()
      ? `NOW() - INTERVAL '24 hours'`
      : "datetime('now', '-24 hours')";
    const interval7d = isPostgres()
      ? `NOW() - INTERVAL '7 days'`
      : "datetime('now', '-7 days')";

    // Run all queries in parallel
    const [
      pipelineCounts,
      activeJobs,
      recentJobs,
      throughput24h,
      throughput7d,
      recentErrors,
      agentStats,
    ] = await Promise.all([
      // 1. Pipeline counts by status
      dbAll(`
        SELECT status, COUNT(*)::int as count
        FROM agent_jobs
        GROUP BY status
      `),

      // 2. Currently active (processing) jobs with agent info
      dbAll(`
        SELECT
          aj.id, aj.job_id, aj.agent_name, aj.agent_user_id, aj.agent_row_id,
          aj.status, aj.created_at, aj.started_at, aj.timeout_at,
          aj.attempts, aj.max_attempts, aj.conversation_id,
          aj.context
        FROM agent_jobs aj
        WHERE aj.status IN ('pending', 'processing')
        ORDER BY aj.created_at ASC
      `),

      // 3. Recent 50 jobs (all statuses)
      dbAll(
        isPostgres()
          ? `SELECT id, job_id, agent_name, agent_user_id, status,
                    created_at, started_at, completed_at, attempts, error_message,
                    conversation_id
             FROM agent_jobs
             ORDER BY created_at DESC
             LIMIT $1`
          : `SELECT id, job_id, agent_name, agent_user_id, status,
                    created_at, started_at, completed_at, attempts, error_message,
                    conversation_id
             FROM agent_jobs
             ORDER BY created_at DESC
             LIMIT ?`,
        [50]
      ),

      // 4. Throughput last 24h
      dbAll(`
        SELECT status, COUNT(*)::int as count
        FROM agent_jobs
        WHERE created_at >= ${interval24h}
        GROUP BY status
      `),

      // 5. Throughput last 7d
      dbAll(`
        SELECT status, COUNT(*)::int as count
        FROM agent_jobs
        WHERE created_at >= ${interval7d}
        GROUP BY status
      `),

      // 6. Recent errors (last 20 failed jobs)
      dbAll(
        isPostgres()
          ? `SELECT id, job_id, agent_name, agent_user_id, error_message,
                    created_at, completed_at, attempts, conversation_id
             FROM agent_jobs
             WHERE status = 'failed'
             ORDER BY completed_at DESC NULLS LAST
             LIMIT $1`
          : `SELECT id, job_id, agent_name, agent_user_id, error_message,
                    created_at, completed_at, attempts, conversation_id
             FROM agent_jobs
             WHERE status = 'failed'
             ORDER BY completed_at DESC
             LIMIT ?`,
        [20]
      ),

      // 7. Per-agent stats (last 7 days)
      dbAll(`
        SELECT
          agent_name,
          agent_user_id,
          COUNT(*)::int as total_jobs,
          COUNT(*) FILTER (WHERE status = 'completed')::int as completed,
          COUNT(*) FILTER (WHERE status = 'failed')::int as failed,
          COUNT(*) FILTER (WHERE status = 'processing')::int as active,
          COUNT(*) FILTER (WHERE status = 'pending')::int as pending,
          ROUND(AVG(EXTRACT(EPOCH FROM (completed_at - started_at)))::numeric, 1) as avg_duration_sec
        FROM agent_jobs
        WHERE created_at >= ${interval7d}
        GROUP BY agent_name, agent_user_id
        ORDER BY total_jobs DESC
      `),
    ]);

    // Build pipeline summary
    const pipeline = { pending: 0, processing: 0, completed: 0, failed: 0, cancelled: 0 };
    for (const row of pipelineCounts) {
      if (pipeline[row.status] !== undefined) {
        pipeline[row.status] = row.count;
      }
    }

    // Build throughput summaries
    const buildThroughput = (rows) => {
      const t = { total: 0, completed: 0, failed: 0, cancelled: 0, pending: 0, processing: 0 };
      for (const row of rows) {
        t[row.status] = row.count || 0;
        t.total += row.count || 0;
      }
      return t;
    };

    return success(res, {
      pipeline,
      agents: agentStats,
      activeJobs,
      recentJobs,
      throughput: {
        last_24h: buildThroughput(throughput24h),
        last_7d: buildThroughput(throughput7d),
      },
      errors: recentErrors,
      generated_at: new Date().toISOString(),
    });
  } catch (err) {
    apiLogger.error({ err }, 'Error fetching autopilot dashboard');
    return error(res, 'AUTOPILOT_DASHBOARD_ERROR', err.message, 500);
  }
});

/**
 * POST /autopilot/jobs/:id/cancel — Cancel a job (delegates to AgentJobService)
 */
router.post('/autopilot/jobs/:id/cancel', async (req, res) => {
  try {
    const result = await AgentJobService.cancelJob(Number(req.params.id));
    if (!result.success) {
      return badRequest(res, result.error);
    }
    return success(res, { message: 'Job cancelled', job_id: Number(req.params.id) });
  } catch (err) {
    apiLogger.error({ err }, 'Error cancelling job from autopilot');
    return error(res, 'AUTOPILOT_CANCEL_ERROR', err.message, 500);
  }
});

export default router;