diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..3f29ae6 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,4 @@ +node_modules +dist +.git +.env diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..1d90882 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,16 @@ +FROM node:22-alpine AS build +WORKDIR /app +COPY package*.json ./ +RUN npm install +COPY tsconfig.json ./ +COPY src ./src +RUN npm run build + +FROM node:22-alpine +WORKDIR /app +ENV NODE_ENV=production +COPY package*.json ./ +RUN npm install --omit=dev && npm cache clean --force +COPY --from=build /app/dist ./dist +EXPOSE 3000 +CMD ["node", "dist/index.js"] diff --git a/README.md b/README.md index f7fd2a0..ed0f5b0 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,48 @@ -# media-intelligence-deploy +# Media Intelligence -Public deployment mirror for Media Intelligence; canonical repo remains private \ No newline at end of file +Human-led media intelligence MVP hosted on Arrowstack, exposed through its own MCP. + +## MVP scope + +- Ingest four open NZ public sources: + - GeoNet earthquake GeoJSON + - NZ Parliament evidence/submissions Atom feed + - Stats NZ information releases + - data.govt.nz CKAN catalogue +- Normalize and hash incoming items. +- Retain source provenance/evidence. +- Surface candidate opportunities using deterministic scoring. +- Human triage via MCP. +- No drafting, journalist database, email distribution, or autonomous publication. + +## MCP + +Endpoint: `/api/mcp` + +Tools: + +- `sources_list` +- `sources_status` +- `sources_sync` +- `opportunities_list` +- `opportunity_get` +- `opportunity_mark` +- `evidence_get` +- `evidence_search` + +The MCP is a separate application boundary from the Arrowstack control-plane MCP. + +## Environment + +- `DATABASE_URL` — injected by Arrowstack tenant DB provisioning. +- `PUBLIC_URL` — public origin, e.g. `https://media.arrowstack.dev`. +- `KEYCLOAK_ISSUER` — Keycloak realm issuer. +- `KEYCLOAK_CLIENT_ID` — dedicated MCP OAuth client. +- `AUTH_DISABLED=true` — local/bootstrap only; do not use for public production. +- `WATCH_TERMS` — optional comma-separated relevance terms. +- `OPPORTUNITY_MIN_SCORE` — default 1. +- `SYNC_INTERVAL_MINUTES` — default 15, minimum 5. + +## Human gate + +The MVP cannot draft or distribute a media release. The only state-changing media action is human triage of an opportunity. diff --git a/package.json b/package.json new file mode 100644 index 0000000..f0cbff4 --- /dev/null +++ b/package.json @@ -0,0 +1,25 @@ +{ + "name": "media-intelligence", + "version": "0.1.0", + "private": true, + "type": "module", + "scripts": { + "build": "tsc -p tsconfig.json", + "start": "node dist/index.js", + "dev": "tsx src/index.ts" + }, + "dependencies": { + "cheerio": "^1.1.2", + "express": "^5.1.0", + "fast-xml-parser": "^5.2.5", + "jose": "^6.1.0", + "pg": "^8.16.3" + }, + "devDependencies": { + "@types/express": "^5.0.3", + "@types/node": "^24.7.0", + "@types/pg": "^8.15.5", + "tsx": "^4.20.6", + "typescript": "^5.9.3" + } +} diff --git a/src/auth.ts b/src/auth.ts new file mode 100644 index 0000000..29143cf --- /dev/null +++ b/src/auth.ts @@ -0,0 +1,45 @@ +import type { NextFunction, Request, Response } from "express"; +import { createRemoteJWKSet, jwtVerify } from "jose"; + +let jwks: ReturnType | null = null; + +export function protectedResourceMetadata(req: Request) { + const base = process.env.PUBLIC_URL || `${req.protocol}://${req.get("host")}`; + return { + resource: `${base.replace(/\/$/, "")}/api/mcp`, + authorization_servers: [process.env.KEYCLOAK_ISSUER || "https://auth.arrowstack.dev/realms/dev"], + bearer_methods_supported: ["header"], + scopes_supported: ["openid", "profile", "email"] + }; +} + +export async function requireMcpAuth(req: Request, res: Response, next: NextFunction) { + if (process.env.AUTH_DISABLED === "true") return next(); + + const issuer = process.env.KEYCLOAK_ISSUER; + const clientId = process.env.KEYCLOAK_CLIENT_ID; + if (!issuer || !clientId) { + return res.status(503).json({ error: "MCP authentication is not configured" }); + } + + const header = req.get("authorization") ?? ""; + if (!header.startsWith("Bearer ")) { + res.setHeader("WWW-Authenticate", `Bearer resource_metadata="${(process.env.PUBLIC_URL || `${req.protocol}://${req.get("host")}`).replace(/\/$/, "")}/.well-known/oauth-protected-resource/api/mcp"`); + return res.status(401).json({ error: "Bearer token required" }); + } + + try { + if (!jwks) { + jwks = createRemoteJWKSet(new URL(`${issuer.replace(/\/$/, "")}/protocol/openid-connect/certs`)); + } + const { payload } = await jwtVerify(header.slice(7), jwks, { issuer }); + const aud = Array.isArray(payload.aud) ? payload.aud : payload.aud ? [payload.aud] : []; + if (payload.azp !== clientId && !aud.includes(clientId)) { + return res.status(403).json({ error: "Token was not issued for this MCP client" }); + } + (req as any).auth = payload; + next(); + } catch (error) { + return res.status(401).json({ error: "Invalid bearer token" }); + } +} diff --git a/src/db.ts b/src/db.ts new file mode 100644 index 0000000..de68cd4 --- /dev/null +++ b/src/db.ts @@ -0,0 +1,261 @@ +import pg from "pg"; +import { randomUUID } from "node:crypto"; +import { createHash } from "node:crypto"; +import type { NormalizedItem, SourceDefinition } from "./types.js"; +import type { ScoredItem } from "./types.js"; + +const { Pool } = pg; + +if (!process.env.DATABASE_URL) { + console.warn("DATABASE_URL is not set; database operations will fail until Arrowstack provisions it."); +} + +export const pool = new Pool({ + connectionString: process.env.DATABASE_URL, + ssl: process.env.DATABASE_SSL === "require" ? { rejectUnauthorized: false } : undefined, + max: Number(process.env.DB_POOL_MAX ?? 8) +}); + +export async function migrate() { + await pool.query(` + CREATE TABLE IF NOT EXISTS sources ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + source_type TEXT NOT NULL, + url TEXT NOT NULL, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + last_synced_at TIMESTAMPTZ, + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + + CREATE TABLE IF NOT EXISTS source_runs ( + id TEXT PRIMARY KEY, + source_id TEXT NOT NULL REFERENCES sources(id), + started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + finished_at TIMESTAMPTZ, + status TEXT NOT NULL, + item_count INTEGER NOT NULL DEFAULT 0, + new_signal_count INTEGER NOT NULL DEFAULT 0, + error TEXT + ); + + CREATE TABLE IF NOT EXISTS signals ( + id TEXT PRIMARY KEY, + source_id TEXT NOT NULL REFERENCES sources(id), + external_id TEXT NOT NULL, + title TEXT NOT NULL, + summary TEXT NOT NULL DEFAULT '', + url TEXT NOT NULL, + published_at TIMESTAMPTZ, + detected_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + content_hash TEXT NOT NULL, + topics JSONB NOT NULL DEFAULT '[]'::jsonb, + raw_json JSONB NOT NULL, + UNIQUE(source_id, external_id, content_hash) + ); + + CREATE INDEX IF NOT EXISTS idx_signals_source_detected ON signals(source_id, detected_at DESC); + CREATE INDEX IF NOT EXISTS idx_signals_external ON signals(source_id, external_id); + + CREATE TABLE IF NOT EXISTS evidence ( + id TEXT PRIMARY KEY, + signal_id TEXT NOT NULL REFERENCES signals(id) ON DELETE CASCADE, + source_url TEXT NOT NULL, + retrieved_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + content_hash TEXT NOT NULL, + excerpt TEXT NOT NULL DEFAULT '' + ); + + CREATE TABLE IF NOT EXISTS opportunities ( + id TEXT PRIMARY KEY, + signal_id TEXT NOT NULL UNIQUE REFERENCES signals(id) ON DELETE CASCADE, + status TEXT NOT NULL DEFAULT 'new', + score INTEGER NOT NULL DEFAULT 0, + reasons JSONB NOT NULL DEFAULT '[]'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + + CREATE INDEX IF NOT EXISTS idx_opportunities_status_created ON opportunities(status, created_at DESC); + + CREATE TABLE IF NOT EXISTS notes ( + id TEXT PRIMARY KEY, + opportunity_id TEXT NOT NULL REFERENCES opportunities(id) ON DELETE CASCADE, + note TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + + CREATE TABLE IF NOT EXISTS audit_events ( + id TEXT PRIMARY KEY, + event_type TEXT NOT NULL, + object_type TEXT NOT NULL, + object_id TEXT NOT NULL, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + `); +} + +export async function seedSources(sources: SourceDefinition[]) { + for (const source of sources) { + await pool.query( + `INSERT INTO sources(id,name,source_type,url) + VALUES($1,$2,$3,$4) + ON CONFLICT(id) DO UPDATE SET name=EXCLUDED.name, source_type=EXCLUDED.source_type, url=EXCLUDED.url`, + [source.id, source.name, source.type, source.url] + ); + } +} + +export function contentHash(item: NormalizedItem) { + return createHash("sha256") + .update(JSON.stringify({ + title: item.title, + summary: item.summary, + url: item.url, + publishedAt: item.publishedAt, + raw: item.raw + })) + .digest("hex"); +} + +export async function storeItem(item: NormalizedItem, scored: ScoredItem): Promise<{ created: boolean; signalId?: string; opportunityId?: string }> { + const hash = contentHash(item); + const existing = await pool.query( + "SELECT id FROM signals WHERE source_id=$1 AND external_id=$2 AND content_hash=$3 LIMIT 1", + [item.sourceId, item.externalId, hash] + ); + if (existing.rowCount) return { created: false }; + + const signalId = randomUUID(); + const opportunityId = randomUUID(); + const evidenceId = randomUUID(); + + await pool.query("BEGIN"); + try { + await pool.query( + `INSERT INTO signals(id,source_id,external_id,title,summary,url,published_at,content_hash,topics,raw_json) + VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9::jsonb,$10::jsonb)`, + [ + signalId, item.sourceId, item.externalId, item.title, item.summary, item.url, + item.publishedAt, hash, JSON.stringify(item.topics), JSON.stringify(item.raw) + ] + ); + await pool.query( + `INSERT INTO evidence(id,signal_id,source_url,content_hash,excerpt) + VALUES($1,$2,$3,$4,$5)`, + [evidenceId, signalId, item.url, hash, `${item.title}\n${item.summary}`.slice(0, 2000)] + ); + + if (scored.score >= Number(process.env.OPPORTUNITY_MIN_SCORE ?? 1)) { + await pool.query( + `INSERT INTO opportunities(id,signal_id,status,score,reasons) + VALUES($1,$2,'new',$3,$4::jsonb)`, + [opportunityId, signalId, scored.score, JSON.stringify(scored.reasons)] + ); + } + + await pool.query( + `INSERT INTO audit_events(id,event_type,object_type,object_id,payload) + VALUES($1,'signal.ingested','signal',$2,$3::jsonb)`, + [randomUUID(), signalId, JSON.stringify({ sourceId: item.sourceId, externalId: item.externalId, score: scored.score })] + ); + await pool.query("COMMIT"); + return { created: true, signalId, opportunityId: scored.score >= Number(process.env.OPPORTUNITY_MIN_SCORE ?? 1) ? opportunityId : undefined }; + } catch (error) { + await pool.query("ROLLBACK"); + throw error; + } +} + +export async function listSources() { + const result = await pool.query("SELECT * FROM sources ORDER BY id"); + return result.rows; +} + +export async function sourceStatus() { + const result = await pool.query(` + SELECT s.*, + (SELECT COUNT(*)::int FROM signals sig WHERE sig.source_id=s.id) AS signal_count, + (SELECT COUNT(*)::int FROM opportunities o JOIN signals sig ON sig.id=o.signal_id WHERE sig.source_id=s.id) AS opportunity_count + FROM sources s + ORDER BY s.id + `); + return result.rows; +} + +export async function listOpportunities(args: { status?: string; since?: string; limit?: number }) { + const values: unknown[] = []; + const where: string[] = []; + if (args.status) { + values.push(args.status); + where.push(`o.status=$${values.length}`); + } + if (args.since) { + values.push(args.since); + where.push(`o.created_at >= $${values.length}::timestamptz`); + } + const limit = Math.min(Math.max(Number(args.limit ?? 25), 1), 100); + values.push(limit); + const result = await pool.query( + `SELECT o.id,o.status,o.score,o.reasons,o.created_at,o.updated_at, + s.id AS signal_id,s.source_id,s.title,s.summary,s.url,s.published_at,s.detected_at,s.topics + FROM opportunities o + JOIN signals s ON s.id=o.signal_id + ${where.length ? "WHERE " + where.join(" AND ") : ""} + ORDER BY o.created_at DESC + LIMIT $${values.length}`, + values + ); + return result.rows; +} + +export async function getOpportunity(id: string) { + const result = await pool.query( + `SELECT o.*, s.source_id,s.external_id,s.title,s.summary,s.url,s.published_at,s.detected_at,s.topics,s.raw_json + FROM opportunities o JOIN signals s ON s.id=o.signal_id WHERE o.id=$1`, + [id] + ); + if (!result.rowCount) return null; + const row = result.rows[0]; + const evidence = await pool.query("SELECT * FROM evidence WHERE signal_id=$1 ORDER BY retrieved_at DESC", [row.signal_id]); + const notes = await pool.query("SELECT * FROM notes WHERE opportunity_id=$1 ORDER BY created_at ASC", [id]); + return { ...row, evidence: evidence.rows, notes: notes.rows }; +} + +export async function markOpportunity(id: string, status: string) { + const allowed = new Set(["new", "watch", "investigate", "develop", "ignore", "closed"]); + if (!allowed.has(status)) throw new Error(`Invalid status: ${status}`); + const result = await pool.query( + "UPDATE opportunities SET status=$2, updated_at=NOW() WHERE id=$1 RETURNING *", + [id, status] + ); + if (!result.rowCount) return null; + await pool.query( + `INSERT INTO audit_events(id,event_type,object_type,object_id,payload) + VALUES($1,'opportunity.status_changed','opportunity',$2,$3::jsonb)`, + [randomUUID(), id, JSON.stringify({ status })] + ); + return result.rows[0]; +} + +export async function getEvidence(id: string) { + const result = await pool.query( + `SELECT e.*, s.source_id,s.title,s.summary,s.url,s.published_at + FROM evidence e JOIN signals s ON s.id=e.signal_id WHERE e.id=$1`, + [id] + ); + return result.rows[0] ?? null; +} + +export async function searchEvidence(query: string, limit = 20) { + const result = await pool.query( + `SELECT e.*, s.source_id,s.title,s.summary,s.url,s.published_at + FROM evidence e JOIN signals s ON s.id=e.signal_id + WHERE e.excerpt ILIKE $1 OR s.title ILIKE $1 OR s.summary ILIKE $1 + ORDER BY e.retrieved_at DESC LIMIT $2`, + [`%${query}%`, Math.min(Math.max(limit, 1), 100)] + ); + return result.rows; +} diff --git a/src/index.ts b/src/index.ts new file mode 100644 index 0000000..21dd21d --- /dev/null +++ b/src/index.ts @@ -0,0 +1,78 @@ +import express from "express"; +import { migrate, pool, seedSources } from "./db.js"; +import { protectedResourceMetadata, requireMcpAuth } from "./auth.js"; +import { handleMcp } from "./mcp.js"; +import { SOURCES } from "./sources.js"; +import { syncAll } from "./sync.js"; + +const app = express(); +app.set("trust proxy", true); +app.use(express.json({ limit: "2mb" })); + +app.get("/health", async (_req, res) => { + try { + await pool.query("SELECT 1"); + res.json({ ok: true, service: "media-intelligence", version: "0.1.0" }); + } catch (error) { + res.status(503).json({ ok: false, error: error instanceof Error ? error.message : String(error) }); + } +}); + +app.get("/.well-known/oauth-protected-resource", (req, res) => res.json(protectedResourceMetadata(req))); +app.get("/.well-known/oauth-protected-resource/api/mcp", (req, res) => res.json(protectedResourceMetadata(req))); + +app.post("/api/mcp", requireMcpAuth, handleMcp); +app.get("/api/mcp", requireMcpAuth, (_req, res) => { + res.status(405).setHeader("Allow", "POST").json({ error: "Use POST for MCP JSON-RPC requests" }); +}); + +app.get("/", async (_req, res) => { + const sourceRows = await pool.query("SELECT id,name,last_synced_at,last_error FROM sources ORDER BY id"); + const counts = await pool.query(` + SELECT + (SELECT COUNT(*) FROM signals) AS signals, + (SELECT COUNT(*) FROM opportunities) AS opportunities, + (SELECT COUNT(*) FROM opportunities WHERE status='new') AS new_opportunities + `); + const c = counts.rows[0] ?? {}; + res.type("html").send(` + +Media Intelligence + +

Media Intelligence MVP

+

Human-led NZ public-source monitoring behind a dedicated MCP.

+
+
${c.signals ?? 0}signals
+
${c.opportunities ?? 0}opportunities
+
${c.new_opportunities ?? 0}new
+
+

Sources

+${sourceRows.rows.map((s:any)=>`
${s.name}
${s.id}
${s.last_error ? 'Error: '+s.last_error : (s.last_synced_at ? 'Last sync '+new Date(s.last_synced_at).toLocaleString() : 'Waiting for first sync')}
`).join("")} +

MCP endpoint: /api/mcp

+`); +}); + +async function start() { + await migrate(); + await seedSources(SOURCES); + + const port = Number(process.env.PORT ?? 3000); + app.listen(port, "0.0.0.0", () => { + console.log(JSON.stringify({ event: "server_started", port, version: "0.1.0" })); + }); + + const intervalMinutes = Math.max(Number(process.env.SYNC_INTERVAL_MINUTES ?? 15), 5); + setTimeout(() => syncAll().then((r) => console.log(JSON.stringify({ event: "initial_sync", result: r }))).catch((e) => console.error(e)), 5000); + setInterval(() => syncAll().then((r) => console.log(JSON.stringify({ event: "scheduled_sync", result: r }))).catch((e) => console.error(e)), intervalMinutes * 60_000); +} + +start().catch((error) => { + console.error(error); + process.exit(1); +}); diff --git a/src/mcp.ts b/src/mcp.ts new file mode 100644 index 0000000..2133c7e --- /dev/null +++ b/src/mcp.ts @@ -0,0 +1,176 @@ +import type { Request, Response } from "express"; +import { + getEvidence, + getOpportunity, + listOpportunities, + listSources, + markOpportunity, + searchEvidence, + sourceStatus +} from "./db.js"; +import { syncAll } from "./sync.js"; +import type { SourceId } from "./types.js"; + +const tools = [ + { + name: "sources_list", + description: "List configured NZ public information sources.", + inputSchema: { type: "object", properties: {}, additionalProperties: false } + }, + { + name: "sources_status", + description: "Show sync health and counts for each source.", + inputSchema: { type: "object", properties: {}, additionalProperties: false } + }, + { + name: "sources_sync", + description: "Run ingestion now for one source or all sources.", + inputSchema: { + type: "object", + properties: { source_id: { type: "string", enum: ["geonet", "parliament", "stats-nz", "data-govt-nz"] } }, + additionalProperties: false + } + }, + { + name: "opportunities_list", + description: "List potential media opportunities surfaced from public-source changes.", + inputSchema: { + type: "object", + properties: { + status: { type: "string", enum: ["new", "watch", "investigate", "develop", "ignore", "closed"] }, + since: { type: "string", description: "ISO-8601 timestamp" }, + limit: { type: "integer", minimum: 1, maximum: 100 } + }, + additionalProperties: false + } + }, + { + name: "opportunity_get", + description: "Get one opportunity with its source signal, provenance evidence, and notes.", + inputSchema: { + type: "object", + properties: { id: { type: "string" } }, + required: ["id"], + additionalProperties: false + } + }, + { + name: "opportunity_mark", + description: "Human triage action: mark an opportunity as new, watch, investigate, develop, ignore, or closed.", + inputSchema: { + type: "object", + properties: { + id: { type: "string" }, + status: { type: "string", enum: ["new", "watch", "investigate", "develop", "ignore", "closed"] } + }, + required: ["id", "status"], + additionalProperties: false + } + }, + { + name: "evidence_get", + description: "Retrieve a stored evidence record and its source provenance.", + inputSchema: { + type: "object", + properties: { id: { type: "string" } }, + required: ["id"], + additionalProperties: false + } + }, + { + name: "evidence_search", + description: "Search stored evidence and source text.", + inputSchema: { + type: "object", + properties: { + query: { type: "string" }, + limit: { type: "integer", minimum: 1, maximum: 100 } + }, + required: ["query"], + additionalProperties: false + } + } +]; + +function result(data: unknown) { + return { + content: [{ type: "text", text: JSON.stringify(data, null, 2) }], + structuredContent: data + }; +} + +async function callTool(name: string, args: any) { + if (name === "sources_list") return result(await listSources()); + if (name === "sources_status") return result(await sourceStatus()); + if (name === "sources_sync") return result(await syncAll(args?.source_id as SourceId | undefined)); + if (name === "opportunities_list") return result(await listOpportunities(args ?? {})); + if (name === "opportunity_get") return result(await getOpportunity(args.id)); + if (name === "opportunity_mark") return result(await markOpportunity(args.id, args.status)); + if (name === "evidence_get") return result(await getEvidence(args.id)); + if (name === "evidence_search") return result(await searchEvidence(args.query, args.limit)); + throw new Error(`Unknown tool: ${name}`); +} + +export async function handleMcp(req: Request, res: Response) { + const body = req.body ?? {}; + const id = body.id; + const method = body.method; + + if (!id && method?.startsWith("notifications/")) { + return res.status(202).end(); + } + + try { + let response: any; + if (method === "initialize") { + response = { + protocolVersion: "2025-06-18", + capabilities: { tools: {}, resources: {} }, + serverInfo: { name: "media-intelligence-mcp", version: "0.1.0" } + }; + } else if (method === "ping") { + response = {}; + } else if (method === "tools/list") { + response = { tools }; + } else if (method === "tools/call") { + response = await callTool(body.params?.name, body.params?.arguments ?? {}); + } else if (method === "resources/list") { + response = { + resources: [ + { + uri: "media://sources", + name: "Configured media intelligence sources", + mimeType: "application/json" + } + ] + }; + } else if (method === "resources/templates/list") { + response = { + resourceTemplates: [ + { uriTemplate: "media://opportunities/{id}", name: "Media opportunity", mimeType: "application/json" }, + { uriTemplate: "media://evidence/{id}", name: "Source evidence", mimeType: "application/json" } + ] + }; + } else if (method === "resources/read") { + const uri = String(body.params?.uri ?? ""); + if (uri === "media://sources") response = { contents: [{ uri, mimeType: "application/json", text: JSON.stringify(await listSources(), null, 2) }] }; + else if (uri.startsWith("media://opportunities/")) response = { contents: [{ uri, mimeType: "application/json", text: JSON.stringify(await getOpportunity(uri.split("/").pop()!), null, 2) }] }; + else if (uri.startsWith("media://evidence/")) response = { contents: [{ uri, mimeType: "application/json", text: JSON.stringify(await getEvidence(uri.split("/").pop()!), null, 2) }] }; + else throw new Error("Unknown resource URI"); + } else { + return res.status(404).json({ + jsonrpc: "2.0", + id: id ?? null, + error: { code: -32601, message: `Method not found: ${method}` } + }); + } + + return res.json({ jsonrpc: "2.0", id, result: response }); + } catch (error) { + return res.status(200).json({ + jsonrpc: "2.0", + id: id ?? null, + error: { code: -32000, message: error instanceof Error ? error.message : String(error) } + }); + } +} diff --git a/src/scoring.ts b/src/scoring.ts new file mode 100644 index 0000000..7f954b6 --- /dev/null +++ b/src/scoring.ts @@ -0,0 +1,58 @@ +import type { NormalizedItem, ScoredItem } from "./types.js"; + +const DEFAULT_TERMS = [ + "health", "technology", "agriculture", "primary production", "horticulture", + "infrastructure", "economy", "employment", "trade", "climate", "energy", + "transport", "earthquake", "food", "housing", "business", "environment" +]; + +function watchTerms() { + return (process.env.WATCH_TERMS?.split(",") ?? DEFAULT_TERMS) + .map((v) => v.trim().toLowerCase()) + .filter(Boolean); +} + +export function scoreItem(item: NormalizedItem): ScoredItem { + let score = 0; + const reasons: string[] = []; + const haystack = `${item.title} ${item.summary} ${item.topics.join(" ")}`.toLowerCase(); + + const matches = watchTerms().filter((term) => haystack.includes(term)).slice(0, 5); + if (matches.length) { + score += Math.min(3, matches.length); + reasons.push(`Matches watch terms: ${matches.join(", ")}`); + } + + if (item.sourceId === "geonet") { + const magnitude = Number(item.metrics?.magnitude ?? 0); + const mmi = Number(item.metrics?.mmi ?? 0); + if (magnitude >= 5) { + score += 4; + reasons.push(`Magnitude ${magnitude} earthquake`); + } else if (magnitude >= 4) { + score += 2; + reasons.push(`Magnitude ${magnitude} earthquake`); + } + if (mmi >= 4) { + score += 2; + reasons.push(`Potentially notable shaking (MMI ${mmi})`); + } + } + + if (item.sourceId === "stats-nz") { + score += 1; + reasons.push("New or changed official statistical release"); + } + + if (item.sourceId === "parliament") { + score += 1; + reasons.push("New or changed parliamentary evidence/submission"); + } + + if (item.sourceId === "data-govt-nz") { + score += 1; + reasons.push("New or updated public dataset metadata"); + } + + return { score, reasons }; +} diff --git a/src/sources.ts b/src/sources.ts new file mode 100644 index 0000000..6b995ba --- /dev/null +++ b/src/sources.ts @@ -0,0 +1,169 @@ +import { load } from "cheerio"; +import { XMLParser } from "fast-xml-parser"; +import type { NormalizedItem, SourceDefinition, SourceId } from "./types.js"; + +export const SOURCES: SourceDefinition[] = [ + { + id: "geonet", + name: "GeoNet earthquakes", + type: "geojson", + url: "https://api.geonet.org.nz/quake?MMI=3" + }, + { + id: "parliament", + name: "NZ Parliament evidence and submissions", + type: "atom", + url: "https://www3.parliament.nz/en/channellisting/rss/9343/e37f168d-4521-4437-a6fd-6bb7ff54e800" + }, + { + id: "stats-nz", + name: "Stats NZ information releases", + type: "html", + url: "https://www.stats.govt.nz/information-releases/" + }, + { + id: "data-govt-nz", + name: "data.govt.nz catalogue", + type: "ckan", + url: "https://catalogue.data.govt.nz/api/3/action/package_search?sort=metadata_modified%20desc&rows=25" + } +]; + +const UA = "Arrowstack-Media-Intelligence/0.1 (+https://media.arrowstack.dev)"; + +async function getText(url: string, accept = "*/*") { + const response = await fetch(url, { + headers: { accept, "user-agent": UA }, + signal: AbortSignal.timeout(20000) + }); + if (!response.ok) throw new Error(`${response.status} ${response.statusText} for ${url}`); + return response.text(); +} + +async function getJson(url: string, accept = "application/json") { + const response = await fetch(url, { + headers: { accept, "user-agent": UA }, + signal: AbortSignal.timeout(20000) + }); + if (!response.ok) throw new Error(`${response.status} ${response.statusText} for ${url}`); + return response.json() as Promise; +} + +export async function fetchSource(id: SourceId): Promise { + if (id === "geonet") return fetchGeoNet(); + if (id === "parliament") return fetchParliament(); + if (id === "stats-nz") return fetchStatsNz(); + return fetchDataGovtNz(); +} + +async function fetchGeoNet(): Promise { + const source = SOURCES.find((s) => s.id === "geonet")!; + const data = await getJson(source.url, "application/vnd.geo+json;version=2"); + return (data.features ?? []).slice(0, 30).map((feature: any) => { + const p = feature.properties ?? {}; + const coords = feature.geometry?.coordinates ?? []; + return { + sourceId: "geonet" as const, + externalId: String(p.publicID ?? feature.id), + title: `M${p.magnitude ?? "?"} earthquake — ${p.locality ?? "New Zealand"}`, + summary: `Magnitude ${p.magnitude ?? "unknown"}, depth ${p.depth ?? "unknown"} km, MMI ${p.mmi ?? "unknown"}, quality ${p.quality ?? "unknown"}.`, + url: p.publicID ? `https://www.geonet.org.nz/earthquake/${p.publicID}` : source.url, + publishedAt: p.time ?? null, + topics: ["natural-hazards", "earthquake"], + metrics: { + magnitude: Number(p.magnitude ?? 0), + mmi: Number(p.mmi ?? 0), + depthKm: Number(p.depth ?? 0), + longitude: coords[0] ?? null, + latitude: coords[1] ?? null + }, + raw: feature + }; + }); +} + +function firstLink(value: any): string { + if (Array.isArray(value)) { + const alt = value.find((v) => v?.["@_rel"] === "alternate") ?? value[0]; + return alt?.["@_href"] ?? ""; + } + return value?.["@_href"] ?? ""; +} + +async function fetchParliament(): Promise { + const source = SOURCES.find((s) => s.id === "parliament")!; + const xml = await getText(source.url, "application/atom+xml, application/xml, text/xml"); + const parser = new XMLParser({ ignoreAttributes: false, attributeNamePrefix: "@_" }); + const parsed = parser.parse(xml); + const entries = parsed?.feed?.entry ?? parsed?.rss?.channel?.item ?? []; + const list = Array.isArray(entries) ? entries : [entries]; + return list.slice(0, 30).map((entry: any, index: number) => { + const title = String(entry.title?.["#text"] ?? entry.title ?? "Parliament item").trim(); + const summary = String(entry.summary?.["#text"] ?? entry.summary ?? entry.description ?? "").replace(/<[^>]+>/g, " ").replace(/\s+/g, " ").trim(); + const url = firstLink(entry.link) || String(entry.link ?? source.url); + return { + sourceId: "parliament" as const, + externalId: String(entry.id ?? entry.guid ?? url ?? index), + title, + summary, + url, + publishedAt: String(entry.updated ?? entry.published ?? entry.pubDate ?? "") || null, + topics: ["parliament", "public-policy"], + raw: entry + }; + }); +} + +function parseDate(text: string): string | null { + const match = text.match(/\b(\d{1,2})\s+(January|February|March|April|May|June|July|August|September|October|November|December)\s+(20\d{2})\b/i); + if (!match) return null; + const d = new Date(`${match[1]} ${match[2]} ${match[3]} 12:00:00 +1200`); + return Number.isNaN(d.getTime()) ? null : d.toISOString(); +} + +async function fetchStatsNz(): Promise { + const source = SOURCES.find((s) => s.id === "stats-nz")!; + const html = await getText(source.url, "text/html"); + const $ = load(html); + const seen = new Set(); + const items: NormalizedItem[] = []; + $('a[href*="/information-releases/"]').each((_i, el) => { + if (items.length >= 30) return; + const href = $(el).attr("href"); + const title = $(el).text().replace(/\s+/g, " ").trim(); + if (!href || !title || title.length < 8) return; + const url = new URL(href, source.url).toString().split("?")[0]; + if (seen.has(url)) return; + seen.add(url); + const container = $(el).closest("article, li, div"); + const surrounding = container.text().replace(/\s+/g, " ").trim(); + const summary = surrounding.replace(title, "").trim().slice(0, 900); + items.push({ + sourceId: "stats-nz", + externalId: url, + title, + summary, + url, + publishedAt: parseDate(surrounding), + topics: ["statistics", "new-zealand"], + raw: { title, summary, url } + }); + }); + return items; +} + +async function fetchDataGovtNz(): Promise { + const source = SOURCES.find((s) => s.id === "data-govt-nz")!; + const data = await getJson(source.url); + if (!data.success) throw new Error("data.govt.nz CKAN returned success=false"); + return (data.result?.results ?? []).slice(0, 30).map((pkg: any) => ({ + sourceId: "data-govt-nz" as const, + externalId: String(pkg.id ?? pkg.name), + title: String(pkg.title ?? pkg.name ?? "Dataset"), + summary: String(pkg.notes ?? "").replace(/\s+/g, " ").trim().slice(0, 1200), + url: pkg.url || `https://catalogue.data.govt.nz/dataset/${pkg.name}`, + publishedAt: pkg.metadata_modified ?? pkg.metadata_created ?? null, + topics: ["open-data", ...(pkg.tags ?? []).slice(0, 6).map((tag: any) => String(tag.display_name ?? tag.name).toLowerCase())], + raw: pkg + })); +} diff --git a/src/sync.ts b/src/sync.ts new file mode 100644 index 0000000..3372be3 --- /dev/null +++ b/src/sync.ts @@ -0,0 +1,64 @@ +import { randomUUID } from "node:crypto"; +import { pool, storeItem } from "./db.js"; +import { fetchSource, SOURCES } from "./sources.js"; +import { scoreItem } from "./scoring.js"; +import type { SourceId } from "./types.js"; + +let running = false; + +export async function syncSource(sourceId: SourceId) { + const source = SOURCES.find((s) => s.id === sourceId); + if (!source) throw new Error(`Unknown source: ${sourceId}`); + const runId = randomUUID(); + await pool.query( + "INSERT INTO source_runs(id,source_id,status) VALUES($1,$2,'running')", + [runId, sourceId] + ); + try { + const items = await fetchSource(sourceId); + let created = 0; + for (const item of items) { + const result = await storeItem(item, scoreItem(item)); + if (result.created) created++; + } + await pool.query( + "UPDATE source_runs SET status='ok',finished_at=NOW(),item_count=$2,new_signal_count=$3 WHERE id=$1", + [runId, items.length, created] + ); + await pool.query( + "UPDATE sources SET last_synced_at=NOW(),last_error=NULL WHERE id=$1", + [sourceId] + ); + return { sourceId, items: items.length, newSignals: created }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + await pool.query( + "UPDATE source_runs SET status='error',finished_at=NOW(),error=$2 WHERE id=$1", + [runId, message] + ); + await pool.query( + "UPDATE sources SET last_synced_at=NOW(),last_error=$2 WHERE id=$1", + [sourceId, message] + ); + throw error; + } +} + +export async function syncAll(sourceId?: SourceId) { + if (running) return { skipped: true, reason: "sync already running" }; + running = true; + try { + if (sourceId) return { skipped: false, results: [await syncSource(sourceId)] }; + const results: any[] = []; + for (const source of SOURCES) { + try { + results.push(await syncSource(source.id)); + } catch (error) { + results.push({ sourceId: source.id, error: error instanceof Error ? error.message : String(error) }); + } + } + return { skipped: false, results }; + } finally { + running = false; + } +} diff --git a/src/types.ts b/src/types.ts new file mode 100644 index 0000000..4bddb1d --- /dev/null +++ b/src/types.ts @@ -0,0 +1,25 @@ +export type SourceId = "geonet" | "parliament" | "stats-nz" | "data-govt-nz"; + +export interface SourceDefinition { + id: SourceId; + name: string; + type: "geojson" | "atom" | "html" | "ckan"; + url: string; +} + +export interface NormalizedItem { + sourceId: SourceId; + externalId: string; + title: string; + summary: string; + url: string; + publishedAt: string | null; + topics: string[]; + raw: unknown; + metrics?: Record; +} + +export interface ScoredItem { + score: number; + reasons: string[]; +} diff --git a/tsconfig.json b/tsconfig.json new file mode 100644 index 0000000..b827103 --- /dev/null +++ b/tsconfig.json @@ -0,0 +1,14 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "outDir": "dist", + "rootDir": "src", + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "resolveJsonModule": true + }, + "include": ["src/**/*.ts"] +}