1079 lines
35 KiB
TypeScript
1079 lines
35 KiB
TypeScript
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,
|
|
baseline_complete BOOLEAN NOT NULL DEFAULT FALSE,
|
|
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,
|
|
change_kind TEXT NOT NULL DEFAULT 'new',
|
|
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 story_threads (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL,
|
|
question TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL DEFAULT 'researching',
|
|
created_by_sub TEXT NOT NULL,
|
|
created_by_name TEXT NOT NULL,
|
|
created_by_client_id TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_story_threads_status_updated
|
|
ON story_threads(status, updated_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS story_nodes (
|
|
id TEXT PRIMARY KEY,
|
|
thread_id TEXT NOT NULL REFERENCES story_threads(id) ON DELETE CASCADE,
|
|
node_type TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
summary TEXT NOT NULL DEFAULT '',
|
|
source_id TEXT REFERENCES sources(id) ON DELETE SET NULL,
|
|
signal_id TEXT REFERENCES signals(id) ON DELETE SET NULL,
|
|
evidence_id TEXT REFERENCES evidence(id) ON DELETE SET NULL,
|
|
opportunity_id TEXT REFERENCES opportunities(id) ON DELETE SET NULL,
|
|
message_id TEXT,
|
|
confidence NUMERIC(4,3),
|
|
created_by_sub TEXT NOT NULL,
|
|
created_by_name TEXT NOT NULL,
|
|
created_by_client_id TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_story_nodes_thread
|
|
ON story_nodes(thread_id, created_at ASC);
|
|
|
|
CREATE TABLE IF NOT EXISTS story_edges (
|
|
id TEXT PRIMARY KEY,
|
|
thread_id TEXT NOT NULL REFERENCES story_threads(id) ON DELETE CASCADE,
|
|
from_node_id TEXT NOT NULL REFERENCES story_nodes(id) ON DELETE CASCADE,
|
|
to_node_id TEXT NOT NULL REFERENCES story_nodes(id) ON DELETE CASCADE,
|
|
relationship TEXT NOT NULL,
|
|
reason TEXT NOT NULL DEFAULT '',
|
|
created_by_sub TEXT NOT NULL,
|
|
created_by_name TEXT NOT NULL,
|
|
created_by_client_id TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
UNIQUE(thread_id, from_node_id, to_node_id, relationship)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_story_edges_thread
|
|
ON story_edges(thread_id, created_at ASC);
|
|
|
|
CREATE TABLE IF NOT EXISTS story_research_checks (
|
|
id TEXT PRIMARY KEY,
|
|
thread_id TEXT NOT NULL REFERENCES story_threads(id) ON DELETE CASCADE,
|
|
check_type TEXT NOT NULL DEFAULT 'counter_evidence',
|
|
result TEXT NOT NULL,
|
|
sources_checked JSONB NOT NULL DEFAULT '[]'::jsonb,
|
|
summary TEXT NOT NULL DEFAULT '',
|
|
completed_by_sub TEXT NOT NULL,
|
|
completed_by_name TEXT NOT NULL,
|
|
completed_by_client_id TEXT NOT NULL,
|
|
completed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
UNIQUE(thread_id, check_type)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS agent_messages (
|
|
id TEXT PRIMARY KEY,
|
|
actor_sub TEXT NOT NULL,
|
|
actor_name TEXT NOT NULL,
|
|
actor_client_id TEXT NOT NULL,
|
|
severity TEXT NOT NULL DEFAULT 'info',
|
|
kind TEXT NOT NULL DEFAULT 'update',
|
|
title TEXT NOT NULL,
|
|
body TEXT NOT NULL,
|
|
related_type TEXT,
|
|
related_id TEXT,
|
|
dedupe_key TEXT UNIQUE,
|
|
acknowledged_at TIMESTAMPTZ,
|
|
acknowledged_by TEXT,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_agent_messages_created
|
|
ON agent_messages(created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_agent_messages_unread
|
|
ON agent_messages(acknowledged_at, created_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS app_meta (
|
|
key TEXT PRIMARY KEY,
|
|
value JSONB NOT NULL DEFAULT '{}'::jsonb,
|
|
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()
|
|
);
|
|
`);
|
|
|
|
await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS baseline_complete BOOLEAN NOT NULL DEFAULT FALSE");
|
|
await pool.query("ALTER TABLE signals ADD COLUMN IF NOT EXISTS change_kind TEXT NOT NULL DEFAULT 'new'");
|
|
await pool.query("ALTER TABLE agent_messages ADD COLUMN IF NOT EXISTS dedupe_key TEXT");
|
|
await pool.query("CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_messages_dedupe_key ON agent_messages(dedupe_key)");
|
|
|
|
const marker = await pool.query(
|
|
`INSERT INTO app_meta(key,value)
|
|
VALUES('noise_baseline_v1','{"applied":true}'::jsonb)
|
|
ON CONFLICT(key) DO NOTHING
|
|
RETURNING key`
|
|
);
|
|
|
|
if (marker.rowCount) {
|
|
await pool.query(
|
|
`UPDATE opportunities
|
|
SET status='closed',
|
|
reasons = reasons || '["Archived during initial baseline cleanup"]'::jsonb,
|
|
updated_at=NOW()
|
|
WHERE status='new'`
|
|
);
|
|
}
|
|
}
|
|
|
|
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]
|
|
);
|
|
}
|
|
|
|
await pool.query(
|
|
`UPDATE sources s
|
|
SET baseline_complete=TRUE
|
|
WHERE baseline_complete=FALSE
|
|
AND EXISTS (SELECT 1 FROM signals sig WHERE sig.source_id=s.id)`
|
|
);
|
|
}
|
|
|
|
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,
|
|
options: { baseline?: boolean } = {}
|
|
): Promise<{
|
|
created: boolean;
|
|
signalId?: string;
|
|
opportunityId?: string;
|
|
changeKind?: "baseline" | "new" | "updated";
|
|
}> {
|
|
const hash = contentHash(item);
|
|
const previous = await pool.query(
|
|
`SELECT id,content_hash
|
|
FROM signals
|
|
WHERE source_id=$1 AND external_id=$2
|
|
ORDER BY detected_at DESC
|
|
LIMIT 1`,
|
|
[item.sourceId, item.externalId]
|
|
);
|
|
|
|
if (previous.rowCount && previous.rows[0].content_hash === hash) {
|
|
return { created: false };
|
|
}
|
|
|
|
const changeKind: "baseline" | "new" | "updated" =
|
|
options.baseline ? "baseline" : previous.rowCount ? "updated" : "new";
|
|
|
|
const signalId = randomUUID();
|
|
const opportunityId = randomUUID();
|
|
const evidenceId = randomUUID();
|
|
|
|
const minScore = Number(process.env.OPPORTUNITY_MIN_SCORE ?? 2);
|
|
const maxAgeDays = Math.max(Number(process.env.MAX_OPPORTUNITY_AGE_DAYS ?? 14), 1);
|
|
|
|
let recentEnough = true;
|
|
if (item.publishedAt) {
|
|
const published = new Date(item.publishedAt).getTime();
|
|
if (!Number.isNaN(published)) {
|
|
recentEnough = Date.now() - published <= maxAgeDays * 86_400_000;
|
|
}
|
|
}
|
|
|
|
const shouldCreateOpportunity =
|
|
!options.baseline &&
|
|
recentEnough &&
|
|
scored.score >= minScore;
|
|
|
|
await pool.query("BEGIN");
|
|
try {
|
|
await pool.query(
|
|
`INSERT INTO signals(
|
|
id,source_id,external_id,title,summary,url,published_at,content_hash,change_kind,topics,raw_json
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10::jsonb,$11::jsonb)`,
|
|
[
|
|
signalId,
|
|
item.sourceId,
|
|
item.externalId,
|
|
item.title,
|
|
item.summary,
|
|
item.url,
|
|
item.publishedAt,
|
|
hash,
|
|
changeKind,
|
|
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 (shouldCreateOpportunity) {
|
|
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,
|
|
changeKind,
|
|
opportunityCreated: shouldCreateOpportunity,
|
|
recentEnough
|
|
})
|
|
]
|
|
);
|
|
|
|
await pool.query("COMMIT");
|
|
|
|
return {
|
|
created: true,
|
|
signalId,
|
|
opportunityId: shouldCreateOpportunity ? opportunityId : undefined,
|
|
changeKind
|
|
};
|
|
} 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
|
|
AND o.status <> 'closed') 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}`);
|
|
} else {
|
|
where.push("o.status <> 'closed'");
|
|
}
|
|
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;
|
|
}
|
|
|
|
|
|
export interface StoryActor {
|
|
sub: string;
|
|
name: string;
|
|
clientId: string;
|
|
}
|
|
|
|
const STORY_NODE_TYPES = new Set([
|
|
"signal", "evidence", "claim", "counterclaim", "trend", "forecast",
|
|
"actual", "question", "context", "organisation", "person", "place", "metric"
|
|
]);
|
|
|
|
const STORY_RELATIONSHIPS = new Set([
|
|
"supports", "contradicts", "qualifies", "explains", "correlates_with",
|
|
"precedes", "updates", "supersedes", "same_subject",
|
|
"regional_breakdown_of", "forecast_for", "actual_for", "investigates"
|
|
]);
|
|
|
|
const STORY_STATUSES = new Set(["researching", "review", "develop", "closed"]);
|
|
|
|
export async function createStoryThread(
|
|
args: { title: string; question?: string },
|
|
actor: StoryActor
|
|
) {
|
|
const title = String(args.title ?? "").trim();
|
|
const question = String(args.question ?? "").trim();
|
|
if (!title) throw new Error("Story thread title is required");
|
|
if (title.length > 220) throw new Error("Story thread title is too long");
|
|
if (question.length > 1200) throw new Error("Story thread question is too long");
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO story_threads(
|
|
id,title,question,status,created_by_sub,created_by_name,created_by_client_id
|
|
) VALUES($1,$2,$3,'researching',$4,$5,$6)
|
|
RETURNING *`,
|
|
[id, title, question, actor.sub, actor.name, actor.clientId]
|
|
);
|
|
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'story_thread.created','story_thread',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({ actor, title, question })]
|
|
);
|
|
|
|
return result.rows[0];
|
|
}
|
|
|
|
export async function listStoryThreads(args: {
|
|
status?: string;
|
|
limit?: number;
|
|
} = {}) {
|
|
const limit = Math.min(Math.max(Number(args.limit ?? 30), 1), 100);
|
|
|
|
if (args.status) {
|
|
if (!STORY_STATUSES.has(args.status)) throw new Error("Invalid story status");
|
|
const result = await pool.query(
|
|
`SELECT t.*,
|
|
(SELECT COUNT(*)::int FROM story_nodes n WHERE n.thread_id=t.id) AS node_count,
|
|
(SELECT COUNT(*)::int FROM story_edges e WHERE e.thread_id=t.id) AS edge_count,
|
|
(SELECT result FROM story_research_checks c
|
|
WHERE c.thread_id=t.id AND c.check_type='counter_evidence'
|
|
LIMIT 1) AS counter_evidence_result
|
|
FROM story_threads t
|
|
WHERE t.status=$1
|
|
ORDER BY t.updated_at DESC
|
|
LIMIT $2`,
|
|
[args.status, limit]
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
const result = await pool.query(
|
|
`SELECT t.*,
|
|
(SELECT COUNT(*)::int FROM story_nodes n WHERE n.thread_id=t.id) AS node_count,
|
|
(SELECT COUNT(*)::int FROM story_edges e WHERE e.thread_id=t.id) AS edge_count,
|
|
(SELECT result FROM story_research_checks c
|
|
WHERE c.thread_id=t.id AND c.check_type='counter_evidence'
|
|
LIMIT 1) AS counter_evidence_result
|
|
FROM story_threads t
|
|
ORDER BY t.updated_at DESC
|
|
LIMIT $1`,
|
|
[limit]
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
export async function getStoryThread(id: string) {
|
|
const threadResult = await pool.query(
|
|
`SELECT t.*,
|
|
(SELECT result FROM story_research_checks c
|
|
WHERE c.thread_id=t.id AND c.check_type='counter_evidence'
|
|
LIMIT 1) AS counter_evidence_result
|
|
FROM story_threads t
|
|
WHERE t.id=$1`,
|
|
[id]
|
|
);
|
|
if (!threadResult.rowCount) return null;
|
|
|
|
const [nodes, edges, checks] = await Promise.all([
|
|
pool.query(
|
|
`SELECT n.*,
|
|
COALESCE(n.source_id,s.source_id) AS provenance_source_id,
|
|
e.source_url AS evidence_url,
|
|
s.url AS signal_url,
|
|
m.title AS message_title
|
|
FROM story_nodes n
|
|
LEFT JOIN evidence e ON e.id=n.evidence_id
|
|
LEFT JOIN signals s ON s.id=COALESCE(n.signal_id,e.signal_id)
|
|
LEFT JOIN agent_messages m ON m.id=n.message_id
|
|
WHERE n.thread_id=$1
|
|
ORDER BY n.created_at ASC`,
|
|
[id]
|
|
),
|
|
pool.query(
|
|
"SELECT * FROM story_edges WHERE thread_id=$1 ORDER BY created_at ASC",
|
|
[id]
|
|
),
|
|
pool.query(
|
|
"SELECT * FROM story_research_checks WHERE thread_id=$1 ORDER BY completed_at DESC",
|
|
[id]
|
|
)
|
|
]);
|
|
|
|
return {
|
|
...threadResult.rows[0],
|
|
nodes: nodes.rows,
|
|
edges: edges.rows,
|
|
research_checks: checks.rows
|
|
};
|
|
}
|
|
|
|
export async function addStoryNode(
|
|
threadId: string,
|
|
args: {
|
|
nodeType: string;
|
|
title: string;
|
|
summary?: string;
|
|
sourceId?: string | null;
|
|
signalId?: string | null;
|
|
evidenceId?: string | null;
|
|
opportunityId?: string | null;
|
|
messageId?: string | null;
|
|
confidence?: number | null;
|
|
},
|
|
actor: StoryActor
|
|
) {
|
|
const nodeType = String(args.nodeType ?? "");
|
|
if (!STORY_NODE_TYPES.has(nodeType)) throw new Error("Invalid story node type");
|
|
|
|
const title = String(args.title ?? "").trim();
|
|
const summary = String(args.summary ?? "").trim();
|
|
if (!title) throw new Error("Story node title is required");
|
|
if (title.length > 240) throw new Error("Story node title is too long");
|
|
if (summary.length > 8000) throw new Error("Story node summary is too long");
|
|
|
|
const confidence =
|
|
args.confidence == null ? null : Number(args.confidence);
|
|
if (confidence != null && (!Number.isFinite(confidence) || confidence < 0 || confidence > 1)) {
|
|
throw new Error("confidence must be between 0 and 1");
|
|
}
|
|
|
|
const thread = await pool.query("SELECT id FROM story_threads WHERE id=$1", [threadId]);
|
|
if (!thread.rowCount) throw new Error("Story thread not found");
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO story_nodes(
|
|
id,thread_id,node_type,title,summary,source_id,signal_id,evidence_id,
|
|
opportunity_id,message_id,confidence,
|
|
created_by_sub,created_by_name,created_by_client_id
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)
|
|
RETURNING *`,
|
|
[
|
|
id, threadId, nodeType, title, summary,
|
|
args.sourceId ?? null,
|
|
args.signalId ?? null,
|
|
args.evidenceId ?? null,
|
|
args.opportunityId ?? null,
|
|
args.messageId ?? null,
|
|
confidence,
|
|
actor.sub, actor.name, actor.clientId
|
|
]
|
|
);
|
|
|
|
await pool.query("UPDATE story_threads SET updated_at=NOW() WHERE id=$1", [threadId]);
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'story_node.added','story_node',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({ threadId, nodeType, actor })]
|
|
);
|
|
|
|
return result.rows[0];
|
|
}
|
|
|
|
export async function addStoryEdge(
|
|
threadId: string,
|
|
args: {
|
|
fromNodeId: string;
|
|
toNodeId: string;
|
|
relationship: string;
|
|
reason?: string;
|
|
},
|
|
actor: StoryActor
|
|
) {
|
|
const relationship = String(args.relationship ?? "");
|
|
if (!STORY_RELATIONSHIPS.has(relationship)) throw new Error("Invalid story relationship");
|
|
if (args.fromNodeId === args.toNodeId) throw new Error("Story edge cannot point to itself");
|
|
|
|
const nodeCheck = await pool.query(
|
|
`SELECT id FROM story_nodes
|
|
WHERE thread_id=$1 AND id = ANY($2::text[])`,
|
|
[threadId, [args.fromNodeId, args.toNodeId]]
|
|
);
|
|
if (nodeCheck.rowCount !== 2) {
|
|
throw new Error("Both story nodes must belong to this thread");
|
|
}
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO story_edges(
|
|
id,thread_id,from_node_id,to_node_id,relationship,reason,
|
|
created_by_sub,created_by_name,created_by_client_id
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)
|
|
ON CONFLICT(thread_id,from_node_id,to_node_id,relationship)
|
|
DO UPDATE SET reason=EXCLUDED.reason
|
|
RETURNING *`,
|
|
[
|
|
id, threadId, args.fromNodeId, args.toNodeId, relationship,
|
|
String(args.reason ?? "").slice(0, 2000),
|
|
actor.sub, actor.name, actor.clientId
|
|
]
|
|
);
|
|
|
|
await pool.query("UPDATE story_threads SET updated_at=NOW() WHERE id=$1", [threadId]);
|
|
return result.rows[0];
|
|
}
|
|
|
|
export async function completeCounterEvidenceCheck(
|
|
threadId: string,
|
|
args: { result: string; sourcesChecked: string[]; summary: string },
|
|
actor: StoryActor
|
|
) {
|
|
const resultValue = String(args.result ?? "");
|
|
if (!new Set(["found", "none_found"]).has(resultValue)) {
|
|
throw new Error("Counter-evidence result must be found or none_found");
|
|
}
|
|
|
|
const sources = Array.from(
|
|
new Set((args.sourcesChecked ?? []).map((v) => String(v).trim()).filter(Boolean))
|
|
).slice(0, 50);
|
|
if (!sources.length) throw new Error("At least one checked source is required");
|
|
|
|
const summary = String(args.summary ?? "").trim();
|
|
if (!summary) throw new Error("Counter-evidence summary is required");
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO story_research_checks(
|
|
id,thread_id,check_type,result,sources_checked,summary,
|
|
completed_by_sub,completed_by_name,completed_by_client_id
|
|
)
|
|
VALUES($1,$2,'counter_evidence',$3,$4::jsonb,$5,$6,$7,$8)
|
|
ON CONFLICT(thread_id,check_type)
|
|
DO UPDATE SET
|
|
result=EXCLUDED.result,
|
|
sources_checked=EXCLUDED.sources_checked,
|
|
summary=EXCLUDED.summary,
|
|
completed_by_sub=EXCLUDED.completed_by_sub,
|
|
completed_by_name=EXCLUDED.completed_by_name,
|
|
completed_by_client_id=EXCLUDED.completed_by_client_id,
|
|
completed_at=NOW()
|
|
RETURNING *`,
|
|
[
|
|
id, threadId, resultValue, JSON.stringify(sources), summary,
|
|
actor.sub, actor.name, actor.clientId
|
|
]
|
|
);
|
|
|
|
await pool.query("UPDATE story_threads SET updated_at=NOW() WHERE id=$1", [threadId]);
|
|
return result.rows[0];
|
|
}
|
|
|
|
export async function markStoryThreadStatus(
|
|
id: string,
|
|
status: string,
|
|
actorLabel: string
|
|
) {
|
|
if (!STORY_STATUSES.has(status)) throw new Error("Invalid story status");
|
|
|
|
if (status === "review" || status === "develop") {
|
|
const check = await pool.query(
|
|
`SELECT result FROM story_research_checks
|
|
WHERE thread_id=$1 AND check_type='counter_evidence'`,
|
|
[id]
|
|
);
|
|
if (!check.rowCount || !new Set(["found", "none_found"]).has(check.rows[0].result)) {
|
|
throw new Error("Counter-evidence research must be completed before review/develop");
|
|
}
|
|
}
|
|
|
|
const result = await pool.query(
|
|
"UPDATE story_threads 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,'story_thread.status_changed','story_thread',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({ status, actorLabel })]
|
|
);
|
|
|
|
return result.rows[0];
|
|
}
|
|
|
|
export async function seedInitialStoryThread() {
|
|
const existing = await pool.query(
|
|
"SELECT id FROM story_threads WHERE title=$1 LIMIT 1",
|
|
["NZ labour market — August 2026"]
|
|
);
|
|
if (existing.rowCount) return existing.rows[0];
|
|
|
|
const evidence = await pool.query(
|
|
`SELECT e.id AS evidence_id,e.signal_id,s.source_id,s.title,s.summary
|
|
FROM evidence e
|
|
JOIN signals s ON s.id=e.signal_id
|
|
WHERE s.url ILIKE '%employment-indicators-august-2026%'
|
|
ORDER BY e.retrieved_at DESC
|
|
LIMIT 1`
|
|
);
|
|
if (!evidence.rowCount) return null;
|
|
|
|
const message = await pool.query(
|
|
`SELECT id,title,body
|
|
FROM agent_messages
|
|
WHERE title='Counter-evidence pass: August employment lead'
|
|
ORDER BY created_at DESC
|
|
LIMIT 1`
|
|
);
|
|
|
|
const actor: StoryActor = {
|
|
sub: "system:story-bootstrap",
|
|
name: "Story Bootstrap",
|
|
clientId: "media-intelligence-worker"
|
|
};
|
|
|
|
await pool.query("BEGIN");
|
|
try {
|
|
const thread = await createStoryThread({
|
|
title: "NZ labour market — August 2026",
|
|
question: "How should rising filled jobs be interpreted alongside unemployment, underutilisation and economic forecasts?"
|
|
}, actor);
|
|
|
|
const primary = await addStoryNode(thread.id, {
|
|
nodeType: "evidence",
|
|
title: "Stats NZ: Employment indicators August 2026",
|
|
summary: "Primary stored employment signal for this research thread.",
|
|
sourceId: evidence.rows[0].source_id,
|
|
signalId: evidence.rows[0].signal_id,
|
|
evidenceId: evidence.rows[0].evidence_id
|
|
}, actor);
|
|
|
|
const claim = await addStoryNode(thread.id, {
|
|
nodeType: "claim",
|
|
title: "Filled jobs are increasing",
|
|
summary: "Candidate interpretation from the August filled-jobs signal; this is not a final editorial conclusion."
|
|
}, actor);
|
|
|
|
const counter = await addStoryNode(thread.id, {
|
|
nodeType: "counterclaim",
|
|
title: "Filled jobs alone do not establish labour-market strengthening",
|
|
summary: "Counter-hypothesis: unemployment, underutilisation, participation, population and forward-looking indicators may qualify the headline signal."
|
|
}, actor);
|
|
|
|
const question = await addStoryNode(thread.id, {
|
|
nodeType: "question",
|
|
title: "What evidence contradicts the apparent improvement?",
|
|
summary: "Required 360° research branch before this thread can move to editorial review."
|
|
}, actor);
|
|
|
|
await addStoryEdge(thread.id, {
|
|
fromNodeId: primary.id,
|
|
toNodeId: claim.id,
|
|
relationship: "supports",
|
|
reason: "The Stats NZ filled-jobs signal is the primary evidence behind the candidate claim."
|
|
}, actor);
|
|
|
|
await addStoryEdge(thread.id, {
|
|
fromNodeId: counter.id,
|
|
toNodeId: claim.id,
|
|
relationship: "qualifies",
|
|
reason: "Filled-jobs growth needs to be interpreted alongside wider labour-market measures."
|
|
}, actor);
|
|
|
|
await addStoryEdge(thread.id, {
|
|
fromNodeId: question.id,
|
|
toNodeId: counter.id,
|
|
relationship: "investigates",
|
|
reason: "The thread explicitly requires a counter-evidence search."
|
|
}, actor);
|
|
|
|
if (message.rowCount) {
|
|
const context = await addStoryNode(thread.id, {
|
|
nodeType: "context",
|
|
title: "Counter-evidence research pass",
|
|
summary: message.rows[0].body,
|
|
messageId: message.rows[0].id
|
|
}, actor);
|
|
|
|
await addStoryEdge(thread.id, {
|
|
fromNodeId: context.id,
|
|
toNodeId: counter.id,
|
|
relationship: "supports",
|
|
reason: "Prior research identified wider labour-market measures that qualify the filled-jobs interpretation."
|
|
}, actor);
|
|
|
|
await completeCounterEvidenceCheck(thread.id, {
|
|
result: "found",
|
|
sourcesChecked: ["Stats NZ", "Treasury", "BNZ"],
|
|
summary: "Material qualifying evidence was identified. Broader RBNZ and MBIE cross-checking remains desirable when those upstream sources are reachable."
|
|
}, actor);
|
|
}
|
|
|
|
await pool.query("COMMIT");
|
|
|
|
await postSystemMessageOnce({
|
|
dedupeKey: `story-thread-bootstrap:${thread.id}`,
|
|
title: "Story graph initialized",
|
|
body: `Created the first 360° research thread: ${thread.title}. The graph includes the primary Stats NZ evidence, candidate claim, counterclaim, explicit counter-evidence question and persisted research check.`,
|
|
severity: "notice",
|
|
kind: "update",
|
|
relatedType: "general",
|
|
relatedId: thread.id
|
|
});
|
|
|
|
return thread;
|
|
} catch (error) {
|
|
await pool.query("ROLLBACK");
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
|
|
export interface AgentMessageActor {
|
|
sub: string;
|
|
name: string;
|
|
clientId: string;
|
|
}
|
|
|
|
export async function postAgentMessage(
|
|
args: {
|
|
title: string;
|
|
body: string;
|
|
severity?: string;
|
|
kind?: string;
|
|
relatedType?: string | null;
|
|
relatedId?: string | null;
|
|
},
|
|
actor: AgentMessageActor
|
|
) {
|
|
const title = String(args.title ?? "").trim();
|
|
const body = String(args.body ?? "").trim();
|
|
if (!title || !body) throw new Error("Message title and body are required");
|
|
if (title.length > 180) throw new Error("Message title is too long");
|
|
if (body.length > 8000) throw new Error("Message body is too long");
|
|
|
|
const severity = String(args.severity ?? "info");
|
|
const kind = String(args.kind ?? "update");
|
|
const allowedSeverities = new Set(["info", "notice", "warning", "error"]);
|
|
const allowedKinds = new Set(["update", "finding", "request", "warning", "error"]);
|
|
const allowedRelated = new Set(["opportunity", "source", "evidence", "general"]);
|
|
|
|
if (!allowedSeverities.has(severity)) throw new Error("Invalid message severity");
|
|
if (!allowedKinds.has(kind)) throw new Error("Invalid message kind");
|
|
|
|
const relatedType = args.relatedType ? String(args.relatedType) : null;
|
|
if (relatedType && !allowedRelated.has(relatedType)) throw new Error("Invalid related_type");
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO agent_messages(
|
|
id,actor_sub,actor_name,actor_client_id,severity,kind,title,body,related_type,related_id
|
|
) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
|
|
RETURNING *`,
|
|
[
|
|
id,
|
|
actor.sub,
|
|
actor.name,
|
|
actor.clientId,
|
|
severity,
|
|
kind,
|
|
title,
|
|
body,
|
|
relatedType,
|
|
args.relatedId ? String(args.relatedId) : null
|
|
]
|
|
);
|
|
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'agent_message.posted','agent_message',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({
|
|
actorSub: actor.sub,
|
|
actorName: actor.name,
|
|
actorClientId: actor.clientId,
|
|
severity,
|
|
kind,
|
|
relatedType,
|
|
relatedId: args.relatedId ?? null
|
|
})]
|
|
);
|
|
|
|
return result.rows[0];
|
|
}
|
|
|
|
export async function listAgentMessages(args: {
|
|
acknowledged?: boolean;
|
|
severity?: string;
|
|
since?: string;
|
|
limit?: number;
|
|
} = {}) {
|
|
const values: unknown[] = [];
|
|
const where: string[] = [];
|
|
|
|
if (typeof args.acknowledged === "boolean") {
|
|
where.push(args.acknowledged ? "acknowledged_at IS NOT NULL" : "acknowledged_at IS NULL");
|
|
}
|
|
if (args.severity) {
|
|
values.push(args.severity);
|
|
where.push(`severity=$${values.length}`);
|
|
}
|
|
if (args.since) {
|
|
values.push(args.since);
|
|
where.push(`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 * FROM agent_messages
|
|
${where.length ? "WHERE " + where.join(" AND ") : ""}
|
|
ORDER BY created_at DESC
|
|
LIMIT $${values.length}`,
|
|
values
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
export async function getAgentMessage(id: string) {
|
|
const result = await pool.query(
|
|
"SELECT * FROM agent_messages WHERE id=$1",
|
|
[id]
|
|
);
|
|
return result.rows[0] ?? null;
|
|
}
|
|
|
|
export async function acknowledgeAgentMessage(id: string, acknowledgedBy: string) {
|
|
const result = await pool.query(
|
|
`UPDATE agent_messages
|
|
SET acknowledged_at=COALESCE(acknowledged_at,NOW()),
|
|
acknowledged_by=COALESCE(acknowledged_by,$2)
|
|
WHERE id=$1
|
|
RETURNING *`,
|
|
[id, acknowledgedBy]
|
|
);
|
|
if (!result.rowCount) return null;
|
|
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'agent_message.acknowledged','agent_message',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({ acknowledgedBy })]
|
|
);
|
|
return result.rows[0];
|
|
}
|
|
|
|
export async function postSystemMessageOnce(args: {
|
|
dedupeKey: string;
|
|
title: string;
|
|
body: string;
|
|
severity?: string;
|
|
kind?: string;
|
|
relatedType?: string | null;
|
|
relatedId?: string | null;
|
|
}) {
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO agent_messages(
|
|
id,actor_sub,actor_name,actor_client_id,severity,kind,title,body,
|
|
related_type,related_id,dedupe_key
|
|
)
|
|
VALUES($1,'system:source-monitor','Source Monitor','media-intelligence-worker',
|
|
$2,$3,$4,$5,$6,$7,$8)
|
|
ON CONFLICT(dedupe_key) DO NOTHING
|
|
RETURNING *`,
|
|
[
|
|
id,
|
|
args.severity ?? "warning",
|
|
args.kind ?? "warning",
|
|
args.title,
|
|
args.body,
|
|
args.relatedType ?? null,
|
|
args.relatedId ?? null,
|
|
args.dedupeKey
|
|
]
|
|
);
|
|
|
|
return result.rows[0] ?? null;
|
|
}
|