diff --git a/src/db.ts b/src/db.ts index 2b36182..1e60030 100644 --- a/src/db.ts +++ b/src/db.ts @@ -4,6 +4,7 @@ import { createHash } from "node:crypto"; import type { NormalizedItem, SourceDefinition } from "./types.js"; import type { ScoredItem } from "./types.js"; import { fetchSourceDocument } from "./documents.js"; +import { publishNewsroomEvent } from "./realtime.js"; const { Pool } = pg; @@ -521,7 +522,9 @@ export async function upsertSourceTrustEntry( })] ); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("source_trust.updated", { entry: realtimeRow }); + return realtimeRow; } export async function evaluateSourceTrust( @@ -903,7 +906,9 @@ export async function markOpportunity(id: string, status: string) { VALUES($1,'opportunity.status_changed','opportunity',$2,$3::jsonb)`, [randomUUID(), id, JSON.stringify({ status })] ); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("opportunity.updated", { opportunity: realtimeRow }); + return realtimeRow; } export async function getEvidence(id: string) { @@ -991,7 +996,9 @@ export async function createStoryThread( [randomUUID(), id, JSON.stringify({ actor, title, question })] ); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("story_thread.created", { story: realtimeRow }); + return realtimeRow; } export async function listStoryThreads(args: { @@ -1336,7 +1343,9 @@ export async function addStoryNode( [randomUUID(), id, JSON.stringify({ threadId, nodeType, actor })] ); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("story_node.added", { node: realtimeRow }); + return realtimeRow; } export async function setStoryNodeReviewState( @@ -1363,7 +1372,9 @@ export async function setStoryNodeReviewState( VALUES($1,'story_node.review_state_changed','story_node',$2,$3::jsonb)`, [randomUUID(), nodeId, JSON.stringify({ threadId, reviewState, actor })] ); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("story_node.updated", { node: realtimeRow }); + return realtimeRow; } export async function addStoryEdge( @@ -1407,7 +1418,9 @@ export async function addStoryEdge( ); await pool.query("UPDATE story_threads SET updated_at=NOW() WHERE id=$1", [threadId]); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("story_edge.updated", { edge: realtimeRow }); + return realtimeRow; } export async function completeCounterEvidenceCheck( @@ -1452,7 +1465,9 @@ export async function completeCounterEvidenceCheck( ); await pool.query("UPDATE story_threads SET updated_at=NOW() WHERE id=$1", [threadId]); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("story_thread.research_check", { check: realtimeRow }); + return realtimeRow; } export async function markStoryThreadStatus( @@ -1489,7 +1504,9 @@ export async function markStoryThreadStatus( [randomUUID(), id, JSON.stringify({ status, actorLabel })] ); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("story_thread.status_changed", { story: realtimeRow }); + return realtimeRow; } @@ -1835,7 +1852,9 @@ export async function postAgentMessage( })] ); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("agent_message.created", { message: realtimeRow }); + return realtimeRow; } export async function listAgentMessages(args: { @@ -1896,7 +1915,9 @@ export async function acknowledgeAgentMessage(id: string, acknowledgedBy: string VALUES($1,'agent_message.acknowledged','agent_message',$2,$3::jsonb)`, [randomUUID(), id, JSON.stringify({ acknowledgedBy })] ); - return result.rows[0]; + const realtimeRow = result.rows[0]; + publishNewsroomEvent("agent_message.acknowledged", { message: realtimeRow }); + return realtimeRow; } export async function postSystemMessageOnce(args: { @@ -1930,5 +1951,7 @@ export async function postSystemMessageOnce(args: { ] ); - return result.rows[0] ?? null; + const message = result.rows[0] ?? null; + if (message) publishNewsroomEvent("agent_message.created", { message }); + return message; } diff --git a/src/index.ts b/src/index.ts index be831e0..1d2871c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -33,6 +33,7 @@ import { import { handleMcp } from "./mcp.js"; import { SOURCES } from "./sources.js"; import { syncAll } from "./sync.js"; +import { serveNewsroomEvents } from "./realtime.js"; import { renderAboutNewsroom, renderDashboard, renderOpportunities, renderOpportunity, renderStories, renderStory, renderTrustPolicy } from "./web.js"; const app = express(); @@ -181,6 +182,8 @@ app.get("/auth/login", beginWebLogin); app.get("/auth/callback", finishWebLogin); app.get("/auth/logout", logoutWeb); +app.get("/api/events", requireWebAuth, serveNewsroomEvents); + app.post("/api/mcp", requireMcpAuth, handleMcp); app.get("/api/mcp", requireMcpAuth, (_req, res) => { res.status(405).setHeader("Allow", "POST").json({ diff --git a/src/realtime.ts b/src/realtime.ts new file mode 100644 index 0000000..b87d022 --- /dev/null +++ b/src/realtime.ts @@ -0,0 +1,82 @@ +import type { Request, Response } from "express"; + +export interface NewsroomRealtimeEvent { + id: string; + type: string; + at: string; + data: unknown; +} + +type Listener = (event: NewsroomRealtimeEvent) => void; + +const listeners = new Set(); +const replayBuffer: NewsroomRealtimeEvent[] = []; +let sequence = 0; +const MAX_REPLAY_EVENTS = 100; +const HEARTBEAT_MS = 20_000; + +function writeEvent(res: Response, event: NewsroomRealtimeEvent) { + res.write(`id: ${event.id}\n`); + res.write("event: newsroom\n"); + res.write(`data: ${JSON.stringify(event)}\n\n`); +} + +export function publishNewsroomEvent(type: string, data: unknown = {}) { + const event: NewsroomRealtimeEvent = { + id: `${Date.now()}-${++sequence}`, + type, + at: new Date().toISOString(), + data + }; + + replayBuffer.push(event); + if (replayBuffer.length > MAX_REPLAY_EVENTS) replayBuffer.shift(); + + for (const listener of listeners) { + try { + listener(event); + } catch { + // A disconnected client is cleaned up by the request close handler. + } + } + return event; +} + +export function serveNewsroomEvents(req: Request, res: Response) { + res.status(200); + res.setHeader("Content-Type", "text/event-stream; charset=utf-8"); + res.setHeader("Cache-Control", "no-cache, no-transform"); + res.setHeader("Connection", "keep-alive"); + res.setHeader("X-Accel-Buffering", "no"); + res.flushHeaders?.(); + res.write("retry: 3000\n\n"); + + const lastEventId = String(req.headers["last-event-id"] ?? "").trim(); + if (lastEventId) { + const index = replayBuffer.findIndex((event) => event.id === lastEventId); + if (index >= 0) { + for (const event of replayBuffer.slice(index + 1)) writeEvent(res, event); + } + } + + writeEvent(res, { + id: `connected-${Date.now()}`, + type: "stream.connected", + at: new Date().toISOString(), + data: { replayCapacity: MAX_REPLAY_EVENTS } + }); + + const listener: Listener = (event) => writeEvent(res, event); + listeners.add(listener); + + const heartbeat = setInterval(() => { + res.write(`: heartbeat ${Date.now()}\n\n`); + }, HEARTBEAT_MS); + + const close = () => { + clearInterval(heartbeat); + listeners.delete(listener); + }; + req.on("close", close); + req.on("aborted", close); +} diff --git a/src/web.ts b/src/web.ts index 4bd61d2..f683cec 100644 --- a/src/web.ts +++ b/src/web.ts @@ -153,7 +153,7 @@ button,input,select{font:inherit} .callout.success{border-left-color:var(--green)} .source-row{display:block} .dot{width:8px;height:8px;border-radius:50%;display:inline-block;margin-right:6px;background:#94a3b8} -.dot.ok{background:#22a06b}.dot.bad{background:#d04f4f}.dot.warn{background:#d78a1f}.attention-grid{display:grid;grid-template-columns:repeat(4,minmax(0,1fr));gap:10px;margin-bottom:20px}.attention-card{display:block;text-decoration:none;background:#fff;border:1px solid var(--line);border-radius:12px;padding:14px 15px;box-shadow:var(--shadow);transition:transform .12s ease,border-color .12s ease}.attention-card:hover{transform:translateY(-1px);border-color:#c8d0db}.attention-top{display:flex;align-items:center;justify-content:space-between;gap:8px}.attention-label{font-size:.78rem;color:var(--muted);font-weight:700}.attention-value{font-size:1.45rem;font-weight:800;letter-spacing:-.03em;margin-top:4px}.attention-copy{font-size:.78rem;color:var(--muted);margin-top:3px}.role-pill{display:inline-flex;align-items:center;border:1px solid rgba(255,255,255,.13);border-radius:999px;padding:2px 7px;font-size:.68rem;color:#cbd5e1;margin-left:5px;text-transform:capitalize}.thread-jumps{position:sticky;top:64px;z-index:35;display:flex;gap:6px;flex-wrap:wrap;margin:15px 0 2px;padding:7px;background:rgba(243,245,247,.92);backdrop-filter:blur(8px);border:1px solid rgba(214,220,228,.9);border-radius:11px}.thread-jumps a{font-size:.77rem;text-decoration:none;color:#536075;padding:6px 9px;border-radius:7px;font-weight:650}.thread-jumps a:hover{background:#fff;color:#172033}.progress-line{height:6px;background:#edf0f4;border-radius:999px;overflow:hidden;margin-top:10px}.progress-line span{display:block;height:100%;background:#16805c;border-radius:999px}.trust-summary{display:grid;grid-template-columns:repeat(4,minmax(0,1fr));gap:10px;margin-bottom:14px}.trust-stat{padding:13px 14px;background:#f8fafc;border:1px solid #e6eaf0;border-radius:10px}.trust-stat strong{display:block;font-size:1.15rem;letter-spacing:-.025em}.trust-stat span{font-size:.76rem;color:var(--muted)} +.dot.ok{background:#22a06b}.dot.bad{background:#d04f4f}.dot.warn{background:#d78a1f}.attention-grid{display:grid;grid-template-columns:repeat(4,minmax(0,1fr));gap:10px;margin-bottom:20px}.attention-card{display:block;text-decoration:none;background:#fff;border:1px solid var(--line);border-radius:12px;padding:14px 15px;box-shadow:var(--shadow);transition:transform .12s ease,border-color .12s ease}.attention-card:hover{transform:translateY(-1px);border-color:#c8d0db}.attention-top{display:flex;align-items:center;justify-content:space-between;gap:8px}.attention-label{font-size:.78rem;color:var(--muted);font-weight:700}.attention-value{font-size:1.45rem;font-weight:800;letter-spacing:-.03em;margin-top:4px}.attention-copy{font-size:.78rem;color:var(--muted);margin-top:3px}.role-pill{display:inline-flex;align-items:center;border:1px solid rgba(255,255,255,.13);border-radius:999px;padding:2px 7px;font-size:.68rem;color:#cbd5e1;margin-left:5px;text-transform:capitalize}.live-pill{display:inline-flex;align-items:center;gap:5px;border:1px solid rgba(255,255,255,.13);border-radius:999px;padding:3px 7px;font-size:.66rem;color:#9ee7c5;text-transform:uppercase;letter-spacing:.05em}.live-pill::before{content:"";width:6px;height:6px;border-radius:50%;background:currentColor}.live-pill.reconnecting{color:#f0c57a}.thread-jumps{position:sticky;top:64px;z-index:35;display:flex;gap:6px;flex-wrap:wrap;margin:15px 0 2px;padding:7px;background:rgba(243,245,247,.92);backdrop-filter:blur(8px);border:1px solid rgba(214,220,228,.9);border-radius:11px}.thread-jumps a{font-size:.77rem;text-decoration:none;color:#536075;padding:6px 9px;border-radius:7px;font-weight:650}.thread-jumps a:hover{background:#fff;color:#172033}.progress-line{height:6px;background:#edf0f4;border-radius:999px;overflow:hidden;margin-top:10px}.progress-line span{display:block;height:100%;background:#16805c;border-radius:999px}.trust-summary{display:grid;grid-template-columns:repeat(4,minmax(0,1fr));gap:10px;margin-bottom:14px}.trust-stat{padding:13px 14px;background:#f8fafc;border:1px solid #e6eaf0;border-radius:10px}.trust-stat strong{display:block;font-size:1.15rem;letter-spacing:-.025em}.trust-stat span{font-size:.76rem;color:var(--muted)} .thread-hero{display:grid;grid-template-columns:minmax(0,1fr) auto;gap:22px;align-items:start} .thread-title{font-size:clamp(1.75rem,3vw,2.55rem);letter-spacing:-.045em;line-height:1.08;margin:5px 0 9px} .thread-question{font-size:1.03rem;color:#536075;max-width:920px} @@ -230,21 +230,93 @@ button,input,select{font:inherit} @media(max-width:700px){.attention-grid,.trust-summary{grid-template-columns:1fr 1fr}.thread-jumps{top:60px;overflow:auto;flex-wrap:nowrap}.thread-jumps a{white-space:nowrap}.trust-form{grid-template-columns:1fr}.topbar-inner{padding:0 14px;gap:13px}.brand span:last-child{display:none}.user-name,.signout{display:none}.page{padding:20px 14px 40px}.page-head{display:block}.page-head>.actions{margin-top:14px}.grid-metrics{grid-template-columns:repeat(2,minmax(0,1fr))}.list-row{grid-template-columns:1fr}.right-meta{text-align:left}.graph-toolbar{align-items:flex-start;flex-direction:column}.graph-scroll{display:none}.mobile-only{display:block}.assertion-grid,.evidence-grid{grid-template-columns:1fr}.relationship{grid-template-columns:1fr}.relationship-mid{text-align:left}} - +
MIMedia Intelligence -
${esc(initials(args.identity))}
${esc(args.identity)}${esc(args.role ?? "viewer")}
Sign out
+
live
${esc(initials(args.identity))}
${esc(args.identity)}${esc(args.role ?? "viewer")}
Sign out
${args.body}
Link copied
`; } @@ -458,7 +530,7 @@ export function renderDashboard(args:{identity:string;role?:string;counts:AnyRow
${compact(c.opportunities)}
active opportunities
${compact(c.new_opportunities)}
new to triage
${compact(c.story_threads)}
open Story Threads
-
${compact(c.unread_messages)}
agent updates
+
${compact(c.unread_messages)}
agent updates

Research needing attention

Threads that still have sourcing or evidence gaps.

All threads
@@ -470,7 +542,7 @@ export function renderDashboard(args:{identity:string;role?:string;counts:AnyRow

Agent Activity

Operational findings and source-monitor updates.

-
${args.messages.length?args.messages.map((m:AnyRow)=>`
${esc(m.title)}
${esc(m.body)}
${badge(m.severity,m.severity==="warning"?"amber":m.severity==="error"?"red":"blue")}${esc(m.actor_name)} via ${esc(m.actor_client_id)}
${canRole(args.role,"editor")?`
`:""}
${esc(nz(m.created_at))}
`).join(""):`
No unread Agent Activity.
`}
`; +
${args.messages.length?args.messages.map((m:AnyRow)=>`
${esc(m.title)}
${esc(m.body)}
${badge(m.severity,m.severity==="warning"?"amber":m.severity==="error"?"red":"blue")}${esc(m.actor_name)} via ${esc(m.actor_client_id)}
${canRole(args.role,"editor")?`
`:""}
${esc(nz(m.created_at))}
`).join(""):`
No unread Agent Activity.
`}
`; return shell({title:"Desk",active:"dashboard",identity:args.identity,role:args.role,body}); }