← back to Norma

app/api/pipeline/route.ts

178 lines

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

/**
 * GET /api/pipeline
 * List pipeline items with optional filters:
 *   ?status=       — filter by status (draft, approved, posting, posted, rejected)
 *   ?platform=     — filter by platform (moveon, actionnetwork, change_org, custom)
 *   ?source_type=  — filter by source (pulse, manual, import)
 *   ?limit=        — results per page (default 50, max 200)
 *   ?offset=       — pagination offset (default 0)
 * Ordered by created_at DESC.
 */
export async function GET(request: NextRequest) {
  const auth = requireRole(request, 'admin', 'staff');
  if (auth instanceof NextResponse) return auth;

  try {
    const { searchParams } = new URL(request.url);
    const status = searchParams.get('status');
    const platform = searchParams.get('platform');
    const sourceType = searchParams.get('source_type');

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

    const conditions: string[] = [];
    const values: unknown[] = [];
    let paramIndex = 1;

    if (orgId) {
      conditions.push(`org_id = $${paramIndex}`);
      values.push(orgId);
      paramIndex++;
    }

    if (status) {
      conditions.push(`status = $${paramIndex}`);
      values.push(status);
      paramIndex++;
    }

    if (platform) {
      conditions.push(`platform = $${paramIndex}`);
      values.push(platform);
      paramIndex++;
    }

    if (sourceType) {
      conditions.push(`source_type = $${paramIndex}`);
      values.push(sourceType);
      paramIndex++;
    }

    let limit = parseInt(searchParams.get('limit') || '50', 10);
    if (isNaN(limit) || limit < 1) limit = 50;
    if (limit > 200) limit = 200;

    let offset = parseInt(searchParams.get('offset') || '0', 10);
    if (isNaN(offset) || offset < 0) offset = 0;

    const whereClause =
      conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : '';

    const countResult = await query(
      `SELECT COUNT(*)::int AS total FROM petition_pipeline ${whereClause}`,
      values
    );
    const total = countResult.rows[0]?.total ?? 0;

    const result = await query(
      `SELECT * FROM petition_pipeline
       ${whereClause}
       ORDER BY position ASC, created_at DESC
       LIMIT $${paramIndex} OFFSET $${paramIndex + 1}`,
      [...values, limit, offset]
    );

    return NextResponse.json({
      items: result.rows,
      total,
      limit,
      offset,
    });
  } catch (err) {
    console.error('[api/pipeline] GET error:', (err as Error).message);
    return NextResponse.json({ error: 'Failed to fetch pipeline items' }, { status: 500 });
  }
}

/**
 * POST /api/pipeline
 * Create a new pipeline item.
 * Body: { title, body, target?, category?, platform?, tone?, talking_points?, geo_states?,
 *         source_type?, source_data?, pulse_topic_id?, ai_confidence?, signature_goal? }
 */
export async function POST(request: NextRequest) {
  const auth = requireRole(request, 'admin', 'staff');
  if (auth instanceof NextResponse) return auth;

  try {
    const data = await request.json();

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

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

    const validPlatforms = ['moveon', 'actionnetwork', 'change_org', 'custom'];
    const platform = validPlatforms.includes(data.platform) ? data.platform : 'moveon';

    const validSourceTypes = ['pulse', 'manual', 'import', 'ai'];
    const sourceType = validSourceTypes.includes(data.source_type) ? data.source_type : 'manual';

    // Normalize talking_points to array
    let talkingPoints: string[] = [];
    if (Array.isArray(data.talking_points)) {
      talkingPoints = data.talking_points.map((t: string) => String(t).trim()).filter(Boolean);
    }

    // Normalize geo_states to array
    let geoStates: string[] = [];
    if (Array.isArray(data.geo_states)) {
      geoStates = data.geo_states.map((s: string) => String(s).trim().toUpperCase()).filter(Boolean);
    }

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

    const result = await query(
      `INSERT INTO petition_pipeline
         (title, body, target, category, platform, tone, talking_points, geo_states,
          source_type, source_data, pulse_topic_id, ai_confidence, signature_goal, status, org_id)
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, 'draft', $14)
       RETURNING *`,
      [
        data.title.trim(),
        data.body.trim(),
        data.target || null,
        data.category || null,
        platform,
        data.tone || null,
        talkingPoints,
        geoStates,
        sourceType,
        data.source_data ? JSON.stringify(data.source_data) : null,
        data.pulse_topic_id || null,
        data.ai_confidence != null ? Number(data.ai_confidence) : null,
        data.signature_goal != null ? Number(data.signature_goal) : null,
        orgId,
      ]
    );

    const item = result.rows[0];

    const ip =
      request.headers.get('x-forwarded-for') ||
      request.headers.get('x-real-ip') ||
      undefined;
    await auditLog(
      'pipeline.created',
      'petition_pipeline',
      item.id,
      { title: item.title, platform: item.platform, source_type: item.source_type },
      ip
    );

    return NextResponse.json({ item }, { status: 201 });
  } catch (err) {
    console.error('[api/pipeline] POST error:', (err as Error).message);
    return NextResponse.json({ error: 'Failed to create pipeline item' }, { status: 500 });
  }
}