← back to Norma

app/api/pipeline/[id]/dispatch/route.ts

167 lines

import { NextRequest, NextResponse } from 'next/server';
import { query } from '@/lib/db';
import { requireRole } from '@/lib/require-role';
import { auditLog } from '@/lib/audit';
import { getOrgId } from '@/lib/orgId';
import { buildInternalAuthHeader } from '@/lib/cron-auth';

/**
 * POST /api/pipeline/[id]/dispatch
 * Dispatch an approved petition to a social/petition platform agent.
 * Body: { platform } — the target platform for dispatch.
 * Updates status to 'posting' and records dispatch intent in social_dispatched JSONB.
 */
export async function POST(
  request: NextRequest,
  { params }: { params: Promise<{ id: string }> }
) {
  const auth = requireRole(request, 'admin', 'staff');
  if (auth instanceof NextResponse) return auth;

  try {
    const { id } = await params;

    const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
    if (!uuidRegex.test(id)) {
      return NextResponse.json({ error: 'Invalid ID format' }, { status: 400 });
    }

    const body = await request.json();

    if (!body.platform || typeof body.platform !== 'string') {
      return NextResponse.json({ error: 'platform is required' }, { status: 400 });
    }

    const validPlatforms = ['moveon', 'actionnetwork', 'change_org', 'twitter', 'bluesky', 'reddit', 'discord', 'custom'];
    if (!validPlatforms.includes(body.platform)) {
      return NextResponse.json(
        { error: `platform must be one of: ${validPlatforms.join(', ')}` },
        { status: 400 }
      );
    }

    const orgId = auth.role === 'admin' ? getOrgId(request) : auth.orgId;

    // Check current status — must be approved or posting (re-dispatch)
    const existing = await query(
      `SELECT id, status, social_dispatched, title FROM petition_pipeline WHERE id = $1 AND ($2::text IS NULL OR org_id = $2)`,
      [id, orgId]
    );

    if (existing.rows.length === 0) {
      return NextResponse.json({ error: 'Pipeline item not found' }, { status: 404 });
    }

    const item = existing.rows[0];
    if (item.status !== 'approved' && item.status !== 'posting') {
      return NextResponse.json(
        { error: `Cannot dispatch item with status '${item.status}'. Item must be approved first.` },
        { status: 409 }
      );
    }

    // Build dispatch record
    const dispatchRecord = {
      dispatched_by: auth.username,
      dispatched_at: new Date().toISOString(),
      status: 'pending',
    };

    // Merge into existing social_dispatched JSONB
    const existingDispatched = item.social_dispatched || {};
    existingDispatched[body.platform] = dispatchRecord;

    // Update status to 'posting' and record dispatch
    const result = await query(
      `UPDATE petition_pipeline
       SET status = 'posting',
           social_dispatched = $1,
           updated_at = NOW()
       WHERE id = $2 AND ($3::text IS NULL OR org_id = $3)
       RETURNING *`,
      [JSON.stringify(existingDispatched), id, orgId]
    );

    const updated = result.rows[0];

    // Log the agent action
    await query(
      `INSERT INTO sdcc_agent_actions
         (agent, action_type, platform, target_id, content, petition_id, status)
       VALUES ($1, $2, $3, $4, $5, $6, $7)`,
      [
        'pipeline-dispatch',
        'dispatch',
        body.platform,
        id,
        `Dispatched petition "${item.title}" to ${body.platform}`,
        id,
        'pending',
      ]
    );

    // ── Actually dispatch to Pulse agent → social agents ──
    let pulseResult = null;
    try {
      const pulseRes = await fetch('http://127.0.0.1:9845/api/pulse/dispatch', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': buildInternalAuthHeader(),
        },
        body: JSON.stringify({
          petition_id: id,
          title: item.title,
          body: updated.body || '',
          target_agents: [body.platform],
        }),
      });
      pulseResult = await pulseRes.json();

      // Update dispatch status based on Pulse response
      const dispatchStatus = pulseResult?.summary?.dispatched > 0 ? 'dispatched' : 'agent_unavailable';
      existingDispatched[body.platform].status = dispatchStatus;
      existingDispatched[body.platform].pulse_result = pulseResult;

      await query(
        `UPDATE petition_pipeline SET social_dispatched = $1, updated_at = NOW() WHERE id = $2 AND ($3::text IS NULL OR org_id = $3)`,
        [JSON.stringify(existingDispatched), id, orgId],
      );
    } catch (pulseErr) {
      console.error('[dispatch] Pulse agent call failed:', (pulseErr as Error).message);
      existingDispatched[body.platform].status = 'pulse_unreachable';
      await query(
        `UPDATE petition_pipeline SET social_dispatched = $1, updated_at = NOW() WHERE id = $2 AND ($3::text IS NULL OR org_id = $3)`,
        [JSON.stringify(existingDispatched), id, orgId],
      );
    }

    const ip =
      request.headers.get('x-forwarded-for') ||
      request.headers.get('x-real-ip') ||
      undefined;
    await auditLog(
      'pipeline.dispatched',
      'petition_pipeline',
      updated.id,
      { platform: body.platform, dispatched_by: auth.username, title: item.title, pulse_result: pulseResult },
      ip
    );

    return NextResponse.json({
      item: updated,
      dispatch: {
        platform: body.platform,
        status: pulseResult?.summary?.dispatched > 0 ? 'dispatched' : 'pending',
        pulse_result: pulseResult,
        message: pulseResult?.summary?.dispatched > 0
          ? `Petition dispatched to ${body.platform} agent successfully.`
          : `Petition queued for ${body.platform}. Agent may be unavailable.`,
      },
    });
  } catch (err) {
    console.error('[api/pipeline/[id]/dispatch] error:', (err as Error).message);
    return NextResponse.json({ error: 'Failed to dispatch pipeline item' }, { status: 500 });
  }
}