Sync Media Intelligence MVP from private canonical repo

.dockerignore

@@ -0,0 +1,4 @@
+node_modules
+dist
+.git
+.env

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"]

README.md

@@ -1,3 +1,48 @@
-# media-intelligence-deploy
+# Media Intelligence
 
-Public deployment mirror for Media Intelligence; canonical repo remains private
+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.

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"
+  }
+}

src/auth.ts

@@ -0,0 +1,45 @@
+import type { NextFunction, Request, Response } from "express";
+import { createRemoteJWKSet, jwtVerify } from "jose";
+
+let jwks: ReturnType<typeof createRemoteJWKSet> | 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" });
+  }
+}

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;
+}

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(`<!doctype html>
+<html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
+<title>Media Intelligence</title>
+<style>
+body{font-family:system-ui,-apple-system,sans-serif;max-width:920px;margin:40px auto;padding:0 20px;color:#1f2937}
+.card{border:1px solid #d1d5db;border-radius:12px;padding:18px;margin:12px 0}
+.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(160px,1fr));gap:12px}
+strong.big{font-size:2rem;display:block}
+.ok{color:#166534}.bad{color:#991b1b}code{background:#f3f4f6;padding:2px 6px;border-radius:5px}
+</style></head><body>
+<h1>Media Intelligence MVP</h1>
+<p>Human-led NZ public-source monitoring behind a dedicated MCP.</p>
+<div class="grid">
+<div class="card"><strong class="big">${c.signals ?? 0}</strong>signals</div>
+<div class="card"><strong class="big">${c.opportunities ?? 0}</strong>opportunities</div>
+<div class="card"><strong class="big">${c.new_opportunities ?? 0}</strong>new</div>
+</div>
+<h2>Sources</h2>
+${sourceRows.rows.map((s:any)=>`<div class="card"><strong>${s.name}</strong><br><code>${s.id}</code><br><span class="${s.last_error?'bad':'ok'}">${s.last_error ? 'Error: '+s.last_error : (s.last_synced_at ? 'Last sync '+new Date(s.last_synced_at).toLocaleString() : 'Waiting for first sync')}</span></div>`).join("")}
+<p>MCP endpoint: <code>/api/mcp</code></p>
+</body></html>`);
+});
+
+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);
+});

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) }
+    });
+  }
+}

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 };
+}

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<T>(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<T>;
+}
+
+export async function fetchSource(id: SourceId): Promise<NormalizedItem[]> {
+  if (id === "geonet") return fetchGeoNet();
+  if (id === "parliament") return fetchParliament();
+  if (id === "stats-nz") return fetchStatsNz();
+  return fetchDataGovtNz();
+}
+
+async function fetchGeoNet(): Promise<NormalizedItem[]> {
+  const source = SOURCES.find((s) => s.id === "geonet")!;
+  const data = await getJson<any>(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<NormalizedItem[]> {
+  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<NormalizedItem[]> {
+  const source = SOURCES.find((s) => s.id === "stats-nz")!;
+  const html = await getText(source.url, "text/html");
+  const $ = load(html);
+  const seen = new Set<string>();
+  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<NormalizedItem[]> {
+  const source = SOURCES.find((s) => s.id === "data-govt-nz")!;
+  const data = await getJson<any>(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
+  }));
+}

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;
+  }
+}

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<string, number | string | null>;
+}
+
+export interface ScoredItem {
+  score: number;
+  reasons: string[];
+}

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"]
+}