From 2888ec1f34b5fb9c5b8ace3294d826f296e14d08 Mon Sep 17 00:00:00 2001 From: Admin Date: Tue, 29 Sep 2026 06:39:43 +0000 Subject: [PATCH] Sync ingestion baseline and noise suppression --- src/db.ts | 132 ++++++++++++++++++++++++++++++++++++++++++++++----- src/index.ts | 8 ++-- src/sync.ts | 28 +++++++++-- 3 files changed, 147 insertions(+), 21 deletions(-) diff --git a/src/db.ts b/src/db.ts index 17822e9..d4f1d7b 100644 --- a/src/db.ts +++ b/src/db.ts @@ -26,6 +26,7 @@ export async function migrate() { 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() ); @@ -50,6 +51,7 @@ export async function migrate() { 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) @@ -107,6 +109,12 @@ export async function migrate() { 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, @@ -116,6 +124,26 @@ export async function migrate() { 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'"); + + 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[]) { @@ -127,6 +155,13 @@ export async function seedSources(sources: SourceDefinition[]) { [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) { @@ -141,35 +176,88 @@ export function contentHash(item: NormalizedItem) { .digest("hex"); } -export async function storeItem(item: NormalizedItem, scored: ScoredItem): Promise<{ created: boolean; signalId?: string; opportunityId?: string }> { +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 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] + 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 (existing.rowCount) return { created: false }; + + 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,topics,raw_json) - VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9::jsonb,$10::jsonb)`, + `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, JSON.stringify(item.topics), JSON.stringify(item.raw) + 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)] + [ + evidenceId, + signalId, + item.url, + hash, + `${item.title}\n${item.summary}`.slice(0, 2000) + ] ); - if (scored.score >= Number(process.env.OPPORTUNITY_MIN_SCORE ?? 1)) { + if (shouldCreateOpportunity) { await pool.query( `INSERT INTO opportunities(id,signal_id,status,score,reasons) VALUES($1,$2,'new',$3,$4::jsonb)`, @@ -180,10 +268,28 @@ export async function storeItem(item: NormalizedItem, scored: ScoredItem): Promi 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 })] + [ + 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: scored.score >= Number(process.env.OPPORTUNITY_MIN_SCORE ?? 1) ? opportunityId : undefined }; + + return { + created: true, + signalId, + opportunityId: shouldCreateOpportunity ? opportunityId : undefined, + changeKind + }; } catch (error) { await pool.query("ROLLBACK"); throw error; diff --git a/src/index.ts b/src/index.ts index 6a0bc85..fe9ee1e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -61,7 +61,7 @@ app.get("/ready", async (_req, res) => { 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 + GROUP BY s.id,s.name,s.last_synced_at,s.last_error,s.baseline_complete ORDER BY s.id` ); const missing = result.rows @@ -114,7 +114,7 @@ app.post("/messages/:id/ack", requireWebAuth, async (req, res) => { app.get("/", requireWebAuth, async (req, res) => { const sourceRows = await pool.query( - `SELECT s.id,s.name,s.last_synced_at,s.last_error, + `SELECT s.id,s.name,s.last_synced_at,s.last_error,s.baseline_complete, COUNT(sig.id)::int AS signal_count FROM sources s LEFT JOIN signals sig ON sig.source_id=s.id @@ -124,7 +124,7 @@ app.get("/", requireWebAuth, async (req, res) => { const counts = await pool.query(` SELECT (SELECT COUNT(*) FROM signals) AS signals, - (SELECT COUNT(*) FROM opportunities) AS opportunities, + (SELECT COUNT(*) FROM opportunities WHERE status <> 'closed') 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 `); @@ -175,7 +175,7 @@ ${messages.length `).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("")} +${sourceRows.rows.map((s:any)=>`
${s.name}
${s.id} · ${s.baseline_complete ? 'baseline ready' : 'awaiting baseline'}
${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.

`); }); diff --git a/src/sync.ts b/src/sync.ts index 3372be3..9dc8c0f 100644 --- a/src/sync.ts +++ b/src/sync.ts @@ -15,21 +15,41 @@ export async function syncSource(sourceId: SourceId) { [runId, sourceId] ); try { + const sourceState = await pool.query( + "SELECT baseline_complete FROM sources WHERE id=$1", + [sourceId] + ); + const baseline = !Boolean(sourceState.rows[0]?.baseline_complete); + const items = await fetchSource(sourceId); let created = 0; + let opportunitiesCreated = 0; + for (const item of items) { - const result = await storeItem(item, scoreItem(item)); + const result = await storeItem(item, scoreItem(item), { baseline }); if (result.created) created++; + if (result.opportunityId) opportunitiesCreated++; } 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] + `UPDATE sources + SET last_synced_at=NOW(), + last_error=NULL, + baseline_complete=CASE WHEN $2 THEN TRUE ELSE baseline_complete END + WHERE id=$1`, + [sourceId, baseline && items.length > 0] ); - return { sourceId, items: items.length, newSignals: created }; + + return { + sourceId, + mode: baseline ? "baseline" : "live", + items: items.length, + newSignals: created, + opportunitiesCreated + }; } catch (error) { const message = error instanceof Error ? error.message : String(error); await pool.query(