import express from "express";
import {
acknowledgeAgentMessage,
listAgentMessages,
migrate,
pool,
seedSources
} from "./db.js";
import {
beginWebLogin,
finishWebLogin,
logoutWeb,
protectedResourceMetadata,
requireMcpAuth,
requireWebAuth
} 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.use(express.urlencoded({ extended: false }));
function escapeHtml(value: unknown) {
return String(value ?? "")
.replace(/&/g, "&")
.replace(//g, ">")
.replace(/"/g, """)
.replace(/'/g, "'");
}
function nzTime(value: unknown) {
if (!value) return "";
try {
return new Intl.DateTimeFormat("en-NZ", {
timeZone: "Pacific/Auckland",
dateStyle: "medium",
timeStyle: "short"
}).format(new Date(String(value)));
} catch {
return String(value);
}
}
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("/ready", async (_req, res) => {
try {
const result = await pool.query(
`SELECT s.id, s.name, s.last_synced_at, s.last_error,
COUNT(sig.id)::int AS signal_count
FROM sources s
LEFT JOIN signals sig ON sig.source_id=s.id
GROUP BY s.id,s.name,s.last_synced_at,s.last_error
ORDER BY s.id`
);
const missing = result.rows
.filter((row: any) => Number(row.signal_count) < 1)
.map((row: any) => row.id);
const errored = result.rows
.filter((row: any) => row.last_error)
.map((row: any) => row.id);
if (missing.length || errored.length) {
return res.status(503).json({
ok: false,
missing_sources: missing,
errored_sources: errored,
sources: result.rows
});
}
return res.json({ ok: true, sources: result.rows });
} catch (error) {
return 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.get("/auth/login", beginWebLogin);
app.get("/auth/callback", finishWebLogin);
app.get("/auth/logout", logoutWeb);
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.post("/messages/:id/ack", requireWebAuth, async (req, res) => {
const user = (req as any).user ?? {};
const identity = String(
user.preferred_username ?? user.email ?? user.name ?? user.sub ?? "web-user"
);
await acknowledgeAgentMessage(String(req.params.id), `web:${identity}`);
return res.redirect("/");
});
app.get("/", requireWebAuth, async (req, res) => {
const sourceRows = await pool.query(
`SELECT s.id,s.name,s.last_synced_at,s.last_error,
COUNT(sig.id)::int AS signal_count
FROM sources s
LEFT JOIN signals sig ON sig.source_id=s.id
GROUP BY s.id,s.name,s.last_synced_at,s.last_error
ORDER BY s.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,
(SELECT COUNT(*) FROM agent_messages WHERE acknowledged_at IS NULL) AS unread_messages
`);
const messages = await listAgentMessages({ acknowledged: false, limit: 10 });
const c = counts.rows[0] ?? {};
const user = (req as any).user ?? {};
const identity = user.name || user.preferred_username || user.email || "Signed-in user";
res.type("html").send(`
Media Intelligence
${c.signals ?? 0}signals
${c.opportunities ?? 0}opportunities
${c.new_opportunities ?? 0}new
${c.unread_messages ?? 0}agent messages
Agent Activity
${messages.length
? messages.map((m:any)=>`
${escapeHtml(m.title)}
${escapeHtml(m.kind)} · ${escapeHtml(m.severity)}
${escapeHtml(nzTime(m.created_at))}
${escapeHtml(m.body)}
From ${escapeHtml(m.actor_name)} via ${escapeHtml(m.actor_client_id)}${m.related_type ? ` · ${escapeHtml(m.related_type)} ${escapeHtml(m.related_id ?? "")}` : ""}
`).join("")
: 'No unread agent messages.
'}
Sources
${sourceRows.rows.map((s:any)=>`${s.name}
${s.id}
${s.signal_count} stored signals
${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 — OAuth protected separately from this browser session.
`);
});
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);
});