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";
import { fetchSourceDocument } from "./documents.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,
      publisher TEXT NOT NULL DEFAULT '',
      source_class TEXT NOT NULL DEFAULT 'catalogue',
      independence_group TEXT NOT NULL DEFAULT '',
      authority_tier INTEGER NOT NULL DEFAULT 3,
      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 source_snapshots (
      id TEXT PRIMARY KEY,
      source_id TEXT NOT NULL REFERENCES sources(id),
      signal_id TEXT NOT NULL REFERENCES signals(id) ON DELETE CASCADE,
      source_url TEXT NOT NULL,
      retrieved_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
      published_at TIMESTAMPTZ,
      content_hash TEXT NOT NULL,
      content_type TEXT NOT NULL DEFAULT 'application/vnd.media-intelligence.normalized+json',
      byte_length INTEGER NOT NULL DEFAULT 0,
      snapshot_json JSONB NOT NULL,
      snapshot_kind TEXT NOT NULL DEFAULT 'normalized',
      canonical_url TEXT,
      document_text TEXT,
      http_status INTEGER,
      UNIQUE(signal_id, content_hash)
    );

    CREATE INDEX IF NOT EXISTS idx_source_snapshots_signal
      ON source_snapshots(signal_id, retrieved_at DESC);

    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 '',
      snapshot_id TEXT REFERENCES source_snapshots(id) ON DELETE SET NULL,
      evidence_kind TEXT NOT NULL DEFAULT 'normalized_excerpt',
      locator TEXT
    );

    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),
      review_state TEXT NOT NULL DEFAULT 'active',
      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()
    );

    CREATE TABLE IF NOT EXISTS source_trust_policy (
      id TEXT PRIMARY KEY,
      domain TEXT NOT NULL UNIQUE,
      name TEXT NOT NULL,
      trust_weight NUMERIC(4,3) NOT NULL CHECK (trust_weight >= 0 AND trust_weight <= 1),
      independence_group TEXT NOT NULL,
      source_class TEXT NOT NULL DEFAULT 'independent',
      notes TEXT NOT NULL DEFAULT '',
      enabled BOOLEAN NOT NULL DEFAULT TRUE,
      source_id TEXT REFERENCES sources(id) ON DELETE SET NULL,
      created_by TEXT NOT NULL DEFAULT 'system',
      created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
      updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
    );

    CREATE INDEX IF NOT EXISTS idx_source_trust_policy_enabled_weight
      ON source_trust_policy(enabled, trust_weight DESC);
  `);

  await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS baseline_complete BOOLEAN NOT NULL DEFAULT FALSE");
  await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS publisher TEXT NOT NULL DEFAULT ''");
  await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS source_class TEXT NOT NULL DEFAULT 'catalogue'");
  await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS independence_group TEXT NOT NULL DEFAULT ''");
  await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS authority_tier INTEGER NOT NULL DEFAULT 3");
  await pool.query("ALTER TABLE evidence ADD COLUMN IF NOT EXISTS snapshot_id TEXT REFERENCES source_snapshots(id) ON DELETE SET NULL");
  await pool.query("ALTER TABLE evidence ADD COLUMN IF NOT EXISTS evidence_kind TEXT NOT NULL DEFAULT 'normalized_excerpt'");
  await pool.query("ALTER TABLE evidence ADD COLUMN IF NOT EXISTS locator TEXT");
  await pool.query("ALTER TABLE source_snapshots ADD COLUMN IF NOT EXISTS snapshot_kind TEXT NOT NULL DEFAULT 'normalized'");
  await pool.query("ALTER TABLE source_snapshots ADD COLUMN IF NOT EXISTS canonical_url TEXT");
  await pool.query("ALTER TABLE source_snapshots ADD COLUMN IF NOT EXISTS document_text TEXT");
  await pool.query("ALTER TABLE source_snapshots ADD COLUMN IF NOT EXISTS http_status INTEGER");
  await pool.query("ALTER TABLE story_nodes ADD COLUMN IF NOT EXISTS review_state TEXT NOT NULL DEFAULT 'active'");

  const payloadMarker = await pool.query(
    `INSERT INTO app_meta(key,value)
     VALUES('evidence_payload_quality_v1','{"applied":true}'::jsonb)
     ON CONFLICT(key) DO NOTHING
     RETURNING key`
  );
  if (payloadMarker.rowCount) {
    await pool.query(`
      UPDATE source_snapshots ss
      SET snapshot_kind='source_payload'
      FROM sources src
      WHERE ss.source_id=src.id
        AND src.source_type IN ('rss','geojson')
        AND src.authority_tier=1
        AND ss.snapshot_kind='normalized'
    `);

    await pool.query(`
      UPDATE evidence e
      SET evidence_kind='source_payload',
          locator=COALESCE(locator,'official feed/API item')
      FROM source_snapshots ss, sources src
      WHERE e.snapshot_id=ss.id
        AND ss.source_id=src.id
        AND ss.snapshot_kind='source_payload'
        AND src.authority_tier=1
        AND e.evidence_kind='normalized_excerpt'
    `);
  }

  await pool.query(`
    INSERT INTO source_snapshots(
      id,source_id,signal_id,source_url,retrieved_at,published_at,content_hash,
      content_type,byte_length,snapshot_json
    )
    SELECT
      'snapshot-' || e.id,
      s.source_id,
      s.id,
      e.source_url,
      e.retrieved_at,
      s.published_at,
      e.content_hash,
      'application/vnd.media-intelligence.normalized+json',
      LENGTH(s.raw_json::text),
      jsonb_build_object(
        'title',s.title,
        'summary',s.summary,
        'url',s.url,
        'publishedAt',s.published_at,
        'topics',s.topics,
        'raw',s.raw_json
      )
    FROM evidence e
    JOIN signals s ON s.id=e.signal_id
    WHERE e.snapshot_id IS NULL
    ON CONFLICT(id) DO NOTHING
  `);
  await pool.query(`
    UPDATE evidence
    SET snapshot_id='snapshot-' || id
    WHERE snapshot_id IS NULL
      AND EXISTS (SELECT 1 FROM source_snapshots ss WHERE ss.id='snapshot-' || evidence.id)
  `);
  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,publisher,source_class,independence_group,authority_tier
       )
       VALUES($1,$2,$3,$4,$5,$6,$7,$8)
       ON CONFLICT(id) DO UPDATE SET
         name=EXCLUDED.name,
         source_type=EXCLUDED.source_type,
         url=EXCLUDED.url,
         publisher=EXCLUDED.publisher,
         source_class=EXCLUDED.source_class,
         independence_group=EXCLUDED.independence_group,
         authority_tier=EXCLUDED.authority_tier`,
      [
        source.id, source.name, source.type, source.url,
        source.publisher, source.sourceClass, source.independenceGroup, source.authorityTier
      ]
    );
  }

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


function normaliseTrustDomain(value: string) {
  const raw = String(value ?? "").trim().toLowerCase();
  if (!raw) throw new Error("Source domain is required");
  let hostname = raw;
  try {
    const url = new URL(raw.includes("://") ? raw : `https://${raw}`);
    hostname = url.hostname.toLowerCase();
  } catch {
    throw new Error("Invalid source domain or URL");
  }
  hostname = hostname.replace(/^www\./, "").replace(/\.$/, "");
  if (!hostname || !hostname.includes(".")) throw new Error("Invalid source domain");
  return hostname;
}

function defaultTrustWeight(sourceClass: string, authorityTier: number) {
  const byClass: Record<string, number> = {
    official_primary: 1.0,
    official_data: 0.98,
    official_registry: 0.97,
    official_notice: 0.95,
    official_analysis: 0.93,
    catalogue: 0.75
  };
  if (sourceClass in byClass) return byClass[sourceClass];
  if (authorityTier === 1) return 0.92;
  if (authorityTier === 2) return 0.78;
  return 0.65;
}

export async function seedSourceTrustPolicy(sources: SourceDefinition[]) {
  for (const source of sources) {
    const domain = normaliseTrustDomain(source.url);
    const weight = defaultTrustWeight(source.sourceClass, source.authorityTier);
    await pool.query(
      `INSERT INTO source_trust_policy(
         id,domain,name,trust_weight,independence_group,source_class,notes,
         enabled,source_id,created_by
       )
       VALUES($1,$2,$3,$4,$5,$6,$7,TRUE,$8,'system-seed')
       ON CONFLICT(domain) DO UPDATE SET
         source_id=COALESCE(source_trust_policy.source_id,EXCLUDED.source_id),
         name=CASE WHEN source_trust_policy.created_by='system-seed' THEN EXCLUDED.name ELSE source_trust_policy.name END,
         independence_group=CASE WHEN source_trust_policy.created_by='system-seed' THEN EXCLUDED.independence_group ELSE source_trust_policy.independence_group END,
         source_class=CASE WHEN source_trust_policy.created_by='system-seed' THEN EXCLUDED.source_class ELSE source_trust_policy.source_class END,
         updated_at=NOW()`,
      [
        randomUUID(),
        domain,
        source.publisher || source.name,
        weight,
        source.independenceGroup || domain,
        source.sourceClass,
        "Seeded from configured ingestion source. Weight is an editorial prior, not a truth guarantee.",
        source.id
      ]
    );
  }
}

export async function listSourceTrustPolicy(args: { enabledOnly?: boolean } = {}) {
  const result = await pool.query(
    `SELECT *
     FROM source_trust_policy
     ${args.enabledOnly ? "WHERE enabled=TRUE" : ""}
     ORDER BY trust_weight DESC, name ASC`
  );
  return result.rows;
}

export async function upsertSourceTrustEntry(
  args: {
    domain: string;
    name: string;
    trustWeight: number;
    independenceGroup?: string;
    sourceClass?: string;
    notes?: string;
    enabled?: boolean;
  },
  actorLabel: string
) {
  const domain = normaliseTrustDomain(args.domain);
  const name = String(args.name ?? "").trim();
  const weight = Number(args.trustWeight);
  const independenceGroup = String(args.independenceGroup ?? domain).trim().toLowerCase();
  const sourceClass = String(args.sourceClass ?? "independent").trim().toLowerCase();
  const notes = String(args.notes ?? "").trim();

  if (!name) throw new Error("Source name is required");
  if (!Number.isFinite(weight) || weight < 0 || weight > 1) {
    throw new Error("Trust weight must be between 0 and 1");
  }
  if (!independenceGroup) throw new Error("Independence group is required");
  if (name.length > 180 || notes.length > 4000) throw new Error("Source trust entry is too long");

  const existing = await pool.query(
    "SELECT id FROM source_trust_policy WHERE domain=$1",
    [domain]
  );
  const id = existing.rows[0]?.id ?? randomUUID();

  const result = await pool.query(
    `INSERT INTO source_trust_policy(
       id,domain,name,trust_weight,independence_group,source_class,notes,
       enabled,created_by
     )
     VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)
     ON CONFLICT(domain) DO UPDATE SET
       name=EXCLUDED.name,
       trust_weight=EXCLUDED.trust_weight,
       independence_group=EXCLUDED.independence_group,
       source_class=EXCLUDED.source_class,
       notes=EXCLUDED.notes,
       enabled=EXCLUDED.enabled,
       created_by=EXCLUDED.created_by,
       updated_at=NOW()
     RETURNING *`,
    [
      id,domain,name,weight,independenceGroup,sourceClass,notes,
      args.enabled !== false,actorLabel
    ]
  );

  await pool.query(
    `INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
     VALUES($1,'source_trust.updated','source_trust',$2,$3::jsonb)`,
    [randomUUID(), result.rows[0].id, JSON.stringify({
      domain, trustWeight: weight, independenceGroup, actor: actorLabel
    })]
  );

  return result.rows[0];
}

export async function evaluateSourceTrust(
  inputs: Array<{ url?: string; domain?: string; sourceId?: string }>
) {
  if (!Array.isArray(inputs) || !inputs.length) {
    throw new Error("At least one source is required");
  }

  const policy = await listSourceTrustPolicy({ enabledOnly: true });
  const bySourceId = new Map(
    policy.filter((p: any) => p.source_id).map((p: any) => [String(p.source_id), p])
  );

  const matched = inputs.slice(0, 50).map((input) => {
    let entry: any = null;
    let domain: string | null = null;

    if (input.sourceId && bySourceId.has(String(input.sourceId))) {
      entry = bySourceId.get(String(input.sourceId));
      domain = String(entry.domain);
    } else {
      const value = input.url || input.domain;
      if (value) {
        try { domain = normaliseTrustDomain(value); } catch { domain = null; }
      }
      if (domain) {
        entry = policy
          .filter((p: any) => domain === p.domain || domain!.endsWith(`.${p.domain}`))
          .sort((a: any, b: any) => String(b.domain).length - String(a.domain).length)[0] ?? null;
      }
    }

    return {
      input,
      matched: Boolean(entry),
      domain,
      policy_id: entry?.id ?? null,
      name: entry?.name ?? null,
      weight: entry ? Number(entry.trust_weight) : null,
      independence_group: entry?.independence_group ?? null,
      source_class: entry?.source_class ?? null
    };
  });

  const groups = new Map<string, any>();
  for (const item of matched.filter((x) => x.matched)) {
    const group = String(item.independence_group || item.domain);
    const current = groups.get(group);
    if (!current || Number(item.weight) > Number(current.weight)) groups.set(group, item);
  }

  const independent = Array.from(groups.values()).sort(
    (a, b) => Number(b.weight) - Number(a.weight)
  );
  const strongest = independent.length ? Number(independent[0].weight) : 0;
  const corroborationBonus = independent
    .slice(1)
    .reduce((sum, item) => sum + Number(item.weight) * 0.05, 0);
  const effectiveScore = Math.min(1, strongest + Math.min(0.15, corroborationBonus));
  const independentCount = independent.length;

  let confidenceBand = "unrated";
  if (independentCount >= 2 && effectiveScore >= 0.9) confidenceBand = "strong";
  else if (independentCount >= 2 && effectiveScore >= 0.75) confidenceBand = "moderate";
  else if (independentCount === 1 && strongest >= 0.85) confidenceBand = "strong_single_source";
  else if (independentCount >= 1) confidenceBand = "limited";

  return {
    policy_version: "source-trust-v1",
    methodology: {
      description: "Use the strongest configured source from each independence group. Repeated sources from the same publisher/group do not add corroboration. Additional independent groups contribute a small capped corroboration bonus.",
      strongest_source_weight: strongest,
      corroboration_bonus: Number(Math.min(0.15, corroborationBonus).toFixed(3)),
      maximum_independence_bonus: 0.15
    },
    effective_score: Number(effectiveScore.toFixed(3)),
    confidence_band: confidenceBand,
    independent_source_count: independentCount,
    meets_independence_minimum: independentCount >= 2,
    policy_coverage: Number((matched.filter((x) => x.matched).length / matched.length).toFixed(3)),
    independent_groups: independent.map((x) => ({
      group: x.independence_group || x.domain,
      representative: x.name,
      weight: x.weight,
      domain: x.domain
    })),
    sources: matched,
    guidance: independentCount >= 2
      ? "Independent corroboration present. Evaluate whether the sources genuinely support the same claim and check for shared upstream evidence."
      : "Independent corroboration is not yet present. Seek at least one source from a different independence group before treating the claim as robust."
  };
}


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 snapshotId = 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;

  const sourceMeta = await pool.query(
    "SELECT source_type,authority_tier FROM sources WHERE id=$1",
    [item.sourceId]
  );
  const directPayload =
    Number(sourceMeta.rows[0]?.authority_tier) === 1 &&
    new Set(["rss","geojson"]).has(String(sourceMeta.rows[0]?.source_type ?? ""));
  const snapshotKind = directPayload ? "source_payload" : "normalized";
  const evidenceKind = directPayload ? "source_payload" : "normalized_excerpt";
  const locator = directPayload ? "official feed/API item" : "title+summary";

  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)
      ]
    );

    const snapshotPayload = {
      title: item.title,
      summary: item.summary,
      url: item.url,
      publishedAt: item.publishedAt,
      topics: item.topics,
      metrics: item.metrics ?? null,
      raw: item.raw
    };
    const snapshotText = JSON.stringify(snapshotPayload);

    await pool.query(
      `INSERT INTO source_snapshots(
         id,source_id,signal_id,source_url,published_at,content_hash,
         content_type,byte_length,snapshot_json,snapshot_kind
       )
       VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9::jsonb,$10)`,
      [
        snapshotId,
        item.sourceId,
        signalId,
        item.url,
        item.publishedAt,
        hash,
        directPayload
          ? "application/vnd.media-intelligence.source-payload+json"
          : "application/vnd.media-intelligence.normalized+json",
        Buffer.byteLength(snapshotText),
        snapshotText,
        snapshotKind
      ]
    );

    await pool.query(
      `INSERT INTO evidence(
         id,signal_id,source_url,content_hash,excerpt,snapshot_id,evidence_kind,locator
       )
       VALUES($1,$2,$3,$4,$5,$6,$7,$8)`,
      [
        evidenceId,
        signalId,
        item.url,
        hash,
        `${item.title}\n${item.summary}`.slice(0, 2000),
        snapshotId,
        evidenceKind,
        locator
      ]
    );

    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 limit = Math.min(Math.max(Number(args.limit ?? 25), 1), 100);
  const base = `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`;

  if (args.status && args.since) {
    const result = await pool.query(
      `${base}
       WHERE o.status=$1 AND o.created_at >= $2::timestamptz
       ORDER BY o.created_at DESC
       LIMIT $3`,
      [args.status, args.since, limit]
    );
    return result.rows;
  }

  if (args.status) {
    const result = await pool.query(
      `${base}
       WHERE o.status=$1
       ORDER BY o.created_at DESC
       LIMIT $2`,
      [args.status, limit]
    );
    return result.rows;
  }

  if (args.since) {
    const result = await pool.query(
      `${base}
       WHERE o.status <> 'closed' AND o.created_at >= $1::timestamptz
       ORDER BY o.created_at DESC
       LIMIT $2`,
      [args.since, limit]
    );
    return result.rows;
  }

  const result = await pool.query(
    `${base}
     WHERE o.status <> 'closed'
     ORDER BY o.created_at DESC
     LIMIT $1`,
    [limit]
  );
  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,
            ss.retrieved_at AS snapshot_retrieved_at,
            ss.published_at AS snapshot_published_at,
            ss.content_type AS snapshot_content_type,
            ss.byte_length AS snapshot_byte_length,
            src.publisher,src.source_class,src.independence_group,src.authority_tier
     FROM evidence e
     JOIN signals s ON s.id=e.signal_id
     LEFT JOIN source_snapshots ss ON ss.id=e.snapshot_id
     LEFT JOIN sources src ON src.id=s.source_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,
            src.publisher,src.source_class,src.independence_group,src.authority_tier,
            ss.retrieved_at AS snapshot_retrieved_at
     FROM evidence e
     JOIN signals s ON s.id=e.signal_id
     LEFT JOIN sources src ON src.id=s.source_id
     LEFT JOIN source_snapshots ss ON ss.id=e.snapshot_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;
}

function buildStorySourceAudit(story: any) {
  const allNodes = Array.isArray(story.nodes) ? story.nodes : [];
  const nodes = allNodes.filter((n: any) => n.review_state !== "irrelevant");
  const nodeIds = new Set(nodes.map((n: any) => n.id));
  const edges = (Array.isArray(story.edges) ? story.edges : [])
    .filter((e: any) => nodeIds.has(e.from_node_id) && nodeIds.has(e.to_node_id));

  const evidenceBacked = nodes.filter((n: any) => {
    if (!n.evidence_id || !n.snapshot_id || !n.provenance_source_id) return false;

    const documentBacked =
      n.snapshot_kind === "source_document" &&
      n.evidence_kind === "source_document" &&
      Number(n.snapshot_text_length ?? 0) >= 500;

    const payloadBacked =
      n.snapshot_kind === "source_payload" &&
      n.evidence_kind === "source_payload" &&
      Number(n.provenance_authority_tier) === 1 &&
      Number(n.snapshot_payload_length ?? 0) >= 350 &&
      Number(n.evidence_excerpt_length ?? 0) >= 160;

    return documentBacked || payloadBacked;
  });

  const distinctSourceIds = Array.from(new Set(
    evidenceBacked.map((n: any) => String(n.provenance_source_id)).filter(Boolean)
  ));
  const distinctGroups = Array.from(new Set(
    evidenceBacked.map((n: any) => String(n.provenance_independence_group)).filter(Boolean)
  ));
  const publishers = Array.from(new Set(
    evidenceBacked.map((n: any) => String(n.provenance_publisher)).filter(Boolean)
  ));
  const tierOneCount = evidenceBacked.filter((n: any) =>
    Number(n.provenance_authority_tier) === 1
  ).length;

  const directEvidenceTypes = new Set(["evidence","actual","signal"]);
  const sourceTaggedButUnbacked = nodes
    .filter((n: any) => directEvidenceTypes.has(n.node_type) && n.source_id && !n.evidence_id)
    .map((n: any) => ({ id: n.id, title: n.title, node_type: n.node_type, source_id: n.source_id }));

  const assertionTypes = new Set(["claim","counterclaim","forecast","trend","metric"]);
  const evidenceNodeIds = new Set(evidenceBacked.map((n: any) => n.id));
  const supportingRelationships = new Set([
    "supports","contradicts","qualifies","forecast_for","actual_for","regional_breakdown_of"
  ]);

  const unsupportedClaims = nodes
    .filter((n: any) => assertionTypes.has(n.node_type))
    .filter((claim: any) => !edges.some((edge: any) =>
      edge.to_node_id === claim.id &&
      evidenceNodeIds.has(edge.from_node_id) &&
      supportingRelationships.has(edge.relationship)
    ))
    .map((n: any) => ({ id: n.id, title: n.title, node_type: n.node_type }));

  const staleCutoff = Date.now() - 45 * 86_400_000;
  const staleEvidence = evidenceBacked
    .filter((n: any) => {
      const value = n.snapshot_published_at ?? n.snapshot_retrieved_at;
      const ts = value ? new Date(value).getTime() : NaN;
      return Number.isFinite(ts) && ts < staleCutoff;
    })
    .map((n: any) => ({ id: n.id, title: n.title, source_id: n.provenance_source_id }));

  const checks = {
    at_least_two_independent_sources: distinctGroups.length >= 2,
    at_least_two_evidence_backed_nodes: evidenceBacked.length >= 2,
    at_least_one_tier_one_source: tierOneCount >= 1,
    all_assertions_have_direct_evidence: unsupportedClaims.length === 0,
    no_source_tag_only_direct_evidence_nodes: sourceTaggedButUnbacked.length === 0
  };

  return {
    robust: Object.values(checks).every(Boolean),
    checks,
    counts: {
      evidence_backed_nodes: evidenceBacked.length,
      distinct_sources: distinctSourceIds.length,
      distinct_independence_groups: distinctGroups.length,
      publishers: publishers.length,
      tier_one_evidence_nodes: tierOneCount,
      unsupported_claims: unsupportedClaims.length,
      source_tag_only_nodes: sourceTaggedButUnbacked.length,
      stale_evidence: staleEvidence.length
    },
    sources: evidenceBacked.map((n: any) => ({
      node_id: n.id,
      node_type: n.node_type,
      title: n.title,
      source_id: n.provenance_source_id,
      publisher: n.provenance_publisher,
      source_class: n.provenance_source_class,
      independence_group: n.provenance_independence_group,
      authority_tier: n.provenance_authority_tier,
      evidence_id: n.evidence_id,
      snapshot_id: n.snapshot_id,
      snapshot_retrieved_at: n.snapshot_retrieved_at,
      snapshot_published_at: n.snapshot_published_at,
      evidence_kind: n.evidence_kind,
      snapshot_kind: n.snapshot_kind,
      snapshot_text_length: n.snapshot_text_length,
      snapshot_payload_length: n.snapshot_payload_length,
      evidence_excerpt_length: n.evidence_excerpt_length,
      url: n.evidence_url ?? n.signal_url
    })),
    unsupported_claims: unsupportedClaims,
    source_tag_only_nodes: sourceTaggedButUnbacked,
    stale_evidence: staleEvidence
  };
}

function buildStoryResearchChecklist(story: any, sourceAudit: any) {
  const nodes = Array.isArray(story.nodes) ? story.nodes : [];
  const activeNodes = nodes.filter((n: any) => n.review_state !== "irrelevant");
  const counterDone = (story.research_checks ?? []).some(
    (c: any) => c.check_type === "counter_evidence" && ["found","none_found"].includes(c.result)
  );
  const unresolvedQuestions = activeNodes.filter(
    (n: any) => n.node_type === "question" && n.review_state !== "resolved"
  );

  const items = [
    {
      key: "primary_evidence",
      label: "Primary/substantive evidence captured",
      done: Number(sourceAudit?.counts?.evidence_backed_nodes ?? 0) >= 1
    },
    {
      key: "counter_evidence",
      label: "Counter-evidence pass completed",
      done: counterDone
    },
    {
      key: "independent_corroboration",
      label: "At least two independent source groups",
      done: Number(sourceAudit?.counts?.distinct_independence_groups ?? 0) >= 2
    },
    {
      key: "assertions_supported",
      label: "All active assertions connected to substantive evidence",
      done: Number(sourceAudit?.counts?.unsupported_claims ?? 0) === 0
    },
    {
      key: "questions_resolved",
      label: "Open research questions resolved or explicitly parked",
      done: unresolvedQuestions.length === 0
    }
  ];

  return {
    ready_for_editorial: Boolean(sourceAudit?.robust && counterDone),
    completed: items.filter((item) => item.done).length,
    total: items.length,
    unresolved_questions: unresolvedQuestions.map((n: any) => ({
      id: n.id,
      title: n.title,
      summary: n.summary
    })),
    items
  };
}

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, requests] = 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,
              src.publisher AS provenance_publisher,
              src.source_class AS provenance_source_class,
              src.independence_group AS provenance_independence_group,
              src.authority_tier AS provenance_authority_tier,
              e.evidence_kind AS evidence_kind,
              e.locator AS evidence_locator,
              e.excerpt AS evidence_excerpt,
              ss.id AS snapshot_id,
              ss.snapshot_kind AS snapshot_kind,
              ss.retrieved_at AS snapshot_retrieved_at,
              ss.published_at AS snapshot_published_at,
              LENGTH(COALESCE(ss.document_text,''))::int AS snapshot_text_length,
              LENGTH(COALESCE(ss.snapshot_json::text,''))::int AS snapshot_payload_length,
              LENGTH(COALESCE(e.excerpt,''))::int AS evidence_excerpt_length
       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
       LEFT JOIN sources src ON src.id=COALESCE(n.source_id,s.source_id)
       LEFT JOIN source_snapshots ss ON ss.id=e.snapshot_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]
    ),
    pool.query(
      "SELECT * FROM agent_messages WHERE related_type='story' AND related_id=$1 ORDER BY created_at DESC",
      [id]
    )
  ]);

  const story = {
    ...threadResult.rows[0],
    nodes: nodes.rows,
    edges: edges.rows,
    research_checks: checks.rows,
    research_requests: requests.rows
  };

  const sourceAudit = buildStorySourceAudit(story);
  return {
    ...story,
    source_audit: sourceAudit,
    research_checklist: buildStoryResearchChecklist(story, sourceAudit)
  };
}

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");

  if (args.evidenceId) {
    await enrichEvidenceDocument(args.evidenceId);
  }

  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 setStoryNodeReviewState(
  threadId: string,
  nodeId: string,
  reviewState: string,
  actor: StoryActor
) {
  const allowed = new Set(["active","resolved","irrelevant"]);
  if (!allowed.has(reviewState)) throw new Error("Invalid story node review state");

  const result = await pool.query(
    `UPDATE story_nodes
     SET review_state=$3
     WHERE id=$2 AND thread_id=$1
     RETURNING *`,
    [threadId,nodeId,reviewState]
  );
  if (!result.rowCount) throw new Error("Story node not found");

  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.review_state_changed','story_node',$2,$3::jsonb)`,
    [randomUUID(), nodeId, JSON.stringify({ threadId, reviewState, 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 sourceAudit = await auditStorySources(id);
    if (!sourceAudit.robust) {
      throw new Error("Story sourcing is not robust enough for review/develop; run story_source_audit for gaps");
    }
  }

  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 auditStorySources(threadId: string) {
  const story = await getStoryThread(threadId);
  if (!story) throw new Error("Story thread not found");
  return { thread_id: threadId, ...(story as any).source_audit };
}

export async function enrichEvidenceDocument(evidenceId: string) {
  const current = await pool.query(
    `SELECT e.*,s.source_id,s.url,s.published_at
     FROM evidence e
     JOIN signals s ON s.id=e.signal_id
     WHERE e.id=$1`,
    [evidenceId]
  );
  if (!current.rowCount) throw new Error("Evidence not found");

  const row = current.rows[0];
  if (
    (row.evidence_kind === "source_document" || row.evidence_kind === "source_payload") &&
    row.snapshot_id
  ) {
    return getEvidence(evidenceId);
  }

  const document = await fetchSourceDocument(row.url);
  const snapshotId = randomUUID();

  await pool.query("BEGIN");
  try {
    await pool.query(
      `INSERT INTO source_snapshots(
         id,source_id,signal_id,source_url,retrieved_at,published_at,content_hash,
         content_type,byte_length,snapshot_json,snapshot_kind,canonical_url,
         document_text,http_status
       )
       VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10::jsonb,'source_document',$11,$12,$13)
       ON CONFLICT(signal_id,content_hash) DO NOTHING`,
      [
        snapshotId,
        row.source_id,
        row.signal_id,
        row.url,
        document.retrievedAt,
        row.published_at,
        document.contentHash,
        document.contentType,
        document.byteLength,
        JSON.stringify({
          requestedUrl: document.requestedUrl,
          canonicalUrl: document.canonicalUrl,
          title: document.title,
          retrievedAt: document.retrievedAt,
          contentHash: document.contentHash
        }),
        document.canonicalUrl,
        document.text,
        document.httpStatus
      ]
    );

    const snapshot = await pool.query(
      `SELECT id FROM source_snapshots
       WHERE signal_id=$1 AND content_hash=$2
       ORDER BY retrieved_at DESC
       LIMIT 1`,
      [row.signal_id, document.contentHash]
    );

    const resolvedSnapshotId = snapshot.rows[0]?.id;
    if (!resolvedSnapshotId) throw new Error("Failed to persist source document snapshot");

    await pool.query(
      `UPDATE evidence
       SET snapshot_id=$2,
           content_hash=$3,
           excerpt=$4,
           evidence_kind='source_document',
           locator='main document body'
       WHERE id=$1`,
      [
        evidenceId,
        resolvedSnapshotId,
        document.contentHash,
        document.text.slice(0, 8000)
      ]
    );

    await pool.query(
      `INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
       VALUES($1,'evidence.source_document_frozen','evidence',$2,$3::jsonb)`,
      [randomUUID(), evidenceId, JSON.stringify({
        sourceId: row.source_id,
        url: document.canonicalUrl,
        snapshotId: resolvedSnapshotId,
        contentHash: document.contentHash
      })]
    );

    await pool.query("COMMIT");
    return getEvidence(evidenceId);
  } catch (error) {
    await pool.query("ROLLBACK");
    throw error;
  }
}

export async function enrichStoryEvidenceDocuments(limit = 20) {
  const result = await pool.query(
    `SELECT DISTINCT n.evidence_id
     FROM story_nodes n
     JOIN evidence e ON e.id=n.evidence_id
     WHERE n.evidence_id IS NOT NULL
       AND e.evidence_kind <> 'source_document'
     ORDER BY n.evidence_id
     LIMIT $1`,
    [Math.min(Math.max(limit, 1), 100)]
  );

  const enriched: any[] = [];
  for (const row of result.rows) {
    try {
      const evidence = await enrichEvidenceDocument(row.evidence_id);
      enriched.push({ evidence_id: row.evidence_id, ok: true, snapshot_id: evidence?.snapshot_id ?? null });
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error);
      const errorHash = createHash("sha256").update(message).digest("hex").slice(0, 16);
      await postSystemMessageOnce({
        dedupeKey: `evidence-enrich-error:${row.evidence_id}:${errorHash}`,
        title: "Source document enrichment failed",
        body: `Evidence ${row.evidence_id} could not be upgraded from discovery metadata to a frozen source document: ${message}`,
        severity: "warning",
        kind: "warning",
        relatedType: "evidence",
        relatedId: row.evidence_id
      });
      enriched.push({
        evidence_id: row.evidence_id,
        ok: false,
        error: message
      });
    }
  }
  return enriched;
}

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", "story", "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;
}