← back to Norma
app/api/ingest/hend/route.ts
313 lines
import { NextRequest, NextResponse } from 'next/server';
import { requireRole } from '@/lib/require-role';
import { query } from '@/lib/db';
import {
searchHENDResources,
getHENDCoalitionOrgs,
type HENDResource,
} from '@/lib/hend-data';
// ─── Helpers ──────────────────────────────────────────────────────────────────
/**
* Ensure the hend_resources table exists before any read/write.
* Runs only DDL if the table is absent — safe to call on every request.
*/
async function ensureTable(): Promise<void> {
await query(`
CREATE TABLE IF NOT EXISTS hend_resources (
id SERIAL PRIMARY KEY,
title TEXT NOT NULL,
url TEXT NOT NULL UNIQUE,
source_type TEXT NOT NULL,
published_date DATE,
author TEXT,
organization TEXT,
summary TEXT,
key_findings JSONB NOT NULL DEFAULT '[]',
states_mentioned TEXT[] NOT NULL DEFAULT '{}',
schools_mentioned TEXT[] NOT NULL DEFAULT '{}',
policy_topics TEXT[] NOT NULL DEFAULT '{}',
raw_data JSONB,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
)
`);
// Index on url is guaranteed by UNIQUE constraint; add GIN indexes for arrays.
await query(`
CREATE INDEX IF NOT EXISTS idx_hend_resources_policy_topics
ON hend_resources USING gin(policy_topics)
`);
await query(`
CREATE INDEX IF NOT EXISTS idx_hend_resources_source_type
ON hend_resources(source_type)
`);
}
/**
* Upsert a single HENDResource into the database.
* ON CONFLICT (url) updates mutable fields so existing rows stay fresh.
*
* Returns 'inserted' | 'updated' | 'skipped'.
*/
async function upsertResource(
resource: HENDResource,
): Promise<'inserted' | 'updated' | 'skipped'> {
const rawData = {
keyFindings: resource.keyFindings,
statesMentioned: resource.statesMentioned,
schoolsMentioned: resource.schoolsMentioned,
policyTopics: resource.policyTopics,
};
const result = await query<{ id: number; xmax: string }>(
`INSERT INTO hend_resources
(title, url, source_type, published_date, author, organization,
summary, key_findings, states_mentioned, schools_mentioned,
policy_topics, raw_data)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
ON CONFLICT (url) DO UPDATE SET
title = EXCLUDED.title,
source_type = EXCLUDED.source_type,
published_date = EXCLUDED.published_date,
author = EXCLUDED.author,
organization = EXCLUDED.organization,
summary = EXCLUDED.summary,
key_findings = EXCLUDED.key_findings,
states_mentioned = EXCLUDED.states_mentioned,
schools_mentioned = EXCLUDED.schools_mentioned,
policy_topics = EXCLUDED.policy_topics,
raw_data = EXCLUDED.raw_data,
updated_at = NOW()
RETURNING id, xmax::text`,
[
resource.title,
resource.url,
resource.sourceType,
resource.publishedDate ?? null,
resource.author ?? null,
resource.organization ?? null,
resource.summary ?? null,
JSON.stringify(resource.keyFindings),
resource.statesMentioned,
resource.schoolsMentioned,
resource.policyTopics,
JSON.stringify(rawData),
],
);
const row = result.rows[0];
if (!row) return 'skipped';
// PostgreSQL sets xmax = 0 on INSERT, non-zero on UPDATE
return row.xmax === '0' ? 'inserted' : 'updated';
}
// ─── GET /api/ingest/hend ─────────────────────────────────────────────────────
/**
* GET /api/ingest/hend
*
* Returns current stats about the hend_resources table:
* - total row count
* - breakdown by source_type
* - breakdown by policy_topic (unnested)
* - most recent updated_at timestamp
*
* Requires authenticated session cookie.
*/
export async function GET(request: NextRequest) {
const auth = requireRole(request, 'admin', 'staff');
if (auth instanceof NextResponse) return auth;
try {
await ensureTable();
const [countResult, byTypeResult, byTopicResult, recentResult] =
await Promise.all([
query<{ total: string }>(`SELECT COUNT(*) AS total FROM hend_resources`),
query<{ source_type: string; count: string }>(
`SELECT source_type, COUNT(*) AS count
FROM hend_resources
GROUP BY source_type
ORDER BY count DESC`,
),
query<{ topic: string; count: string }>(
`SELECT unnest(policy_topics) AS topic, COUNT(*) AS count
FROM hend_resources
GROUP BY topic
ORDER BY count DESC
LIMIT 20`,
),
query<{ updated_at: string }>(
`SELECT updated_at FROM hend_resources ORDER BY updated_at DESC LIMIT 1`,
),
]);
const byType: Record<string, number> = {};
for (const row of byTypeResult.rows) {
byType[row.source_type] = parseInt(row.count, 10);
}
const byTopic: Record<string, number> = {};
for (const row of byTopicResult.rows) {
byTopic[row.topic] = parseInt(row.count, 10);
}
return NextResponse.json(
{
total: parseInt(countResult.rows[0]?.total ?? '0', 10),
byType,
byTopic,
lastUpdated: recentResult.rows[0]?.updated_at ?? null,
},
{ status: 200 },
);
} catch (err) {
console.error('[api/ingest/hend] GET error:', (err as Error).message);
return NextResponse.json(
{ error: `Failed to fetch HEND stats: ${(err as Error).message}` },
{ status: 500 },
);
}
}
// ─── POST /api/ingest/hend ────────────────────────────────────────────────────
interface IngestOptions {
/** Include static coalition org records (default: true) */
includeCoalitionOrgs?: boolean;
/** Fetch from Federal Register + Congress.gov APIs (default: true) */
fetchLive?: boolean;
}
/**
* POST /api/ingest/hend
*
* Fetches HEND-related resources from external APIs and the static coalition
* org registry, then upserts all records into the hend_resources table.
*
* Optional JSON body:
* { includeCoalitionOrgs?: boolean, fetchLive?: boolean }
*
* Response:
* 200 { inserted, updated, skipped, total, errors, durationMs }
* 400 { error: string }
* 401 { error: string }
* 500 { error: string }
*
* Requires authenticated session cookie.
*/
export async function POST(request: NextRequest) {
const auth = requireRole(request, 'admin', 'staff');
if (auth instanceof NextResponse) return auth;
// Parse optional body
const opts: IngestOptions = {
includeCoalitionOrgs: true,
fetchLive: true,
};
const contentType = request.headers.get('content-type') ?? '';
if (contentType.includes('application/json')) {
try {
const body: unknown = await request.json();
if (body && typeof body === 'object') {
const b = body as Record<string, unknown>;
if (typeof b.includeCoalitionOrgs === 'boolean') {
opts.includeCoalitionOrgs = b.includeCoalitionOrgs;
}
if (typeof b.fetchLive === 'boolean') {
opts.fetchLive = b.fetchLive;
}
}
} catch {
// Non-JSON body — use defaults
}
}
const startTime = Date.now();
try {
await ensureTable();
// Collect resources from all enabled sources in parallel
const sourceTasks: Promise<HENDResource[]>[] = [];
if (opts.fetchLive) {
sourceTasks.push(
searchHENDResources().catch((err) => {
console.warn('[api/ingest/hend] searchHENDResources error:', (err as Error).message);
return [];
}),
);
}
if (opts.includeCoalitionOrgs) {
// getHENDCoalitionOrgs is synchronous; wrap in a resolved promise
sourceTasks.push(Promise.resolve(getHENDCoalitionOrgs()));
}
const batches = await Promise.all(sourceTasks);
const allResources: HENDResource[] = batches.flat();
if (allResources.length === 0) {
return NextResponse.json(
{
inserted: 0,
updated: 0,
skipped: 0,
total: 0,
errors: [],
durationMs: Date.now() - startTime,
message: 'No resources to ingest (both sources disabled or returned empty)',
},
{ status: 200 },
);
}
// Upsert each resource, collecting results
let inserted = 0;
let updated = 0;
let skipped = 0;
const errors: string[] = [];
for (const resource of allResources) {
try {
const outcome = await upsertResource(resource);
if (outcome === 'inserted') inserted++;
else if (outcome === 'updated') updated++;
else skipped++;
} catch (err) {
const msg = `Failed to upsert "${resource.title}": ${(err as Error).message}`;
console.error('[api/ingest/hend]', msg);
errors.push(msg);
skipped++;
}
}
const durationMs = Date.now() - startTime;
return NextResponse.json(
{
inserted,
updated,
skipped,
total: allResources.length,
errors,
durationMs,
},
{ status: 200 },
);
} catch (err) {
console.error('[api/ingest/hend] POST error:', (err as Error).message);
return NextResponse.json(
{ error: `HEND ingest failed: ${(err as Error).message}` },
{ status: 500 },
);
}
}