Duriin-API/intelligence/graphWorker.js

// builds cross-company relationship graph from company_facts rows where type = 'relationship'

const VALID_TYPES = ["supplier", "customer", "competitor", "partner", "investor", "dependency"];

const KEYWORD_MAP = [
  ["manufactur", "supplier"],
  ["suppli", "supplier"],
  ["distribut", "supplier"],
  ["compet", "competitor"],
  ["rival", "competitor"],
  ["invest", "investor"],
  ["fund", "investor"],
  ["backer", "investor"],
  ["customer", "customer"],
  ["client", "customer"],
  ["depend", "dependency"],
  ["partner", "partner"],
  ["collaborat", "partner"],
  ["joint", "partner"],
];

const RECIPROCAL = {
  supplier: "customer",
  customer: "supplier",
  competitor: "competitor",
  partner: "partner",
  investor: "dependency",
  dependency: "investor",
};


function parseRelationshipType(rawType) {
  const t = (rawType || "").toLowerCase().trim();

  if (VALID_TYPES.includes(t)) return t;

  for (const [kw, mapped] of KEYWORD_MAP) {
    if (t.includes(kw)) return mapped;
  }

  return "partner";
}


function parseClaim(claim) {
  // format from consolidationWorker: "entity is a type"
  const parts = claim.split(" is a ");
  if (parts.length < 2) return null;

  const toEntity = parts[0].trim();
  const rawType = parts.slice(1).join(" is a ").trim();

  if (!toEntity || !rawType) return null;

  return {
    toEntity,
    relationshipType: parseRelationshipType(rawType),
  };
}


function resolveCompany(toEntity, companies) {
  const needle = toEntity.toLowerCase();

  for (const co of companies) {
    if (co.name.toLowerCase() === needle) return co;
    if (co.ticker && co.ticker.toLowerCase() === needle) return co;

    let aliases = [];
    try { aliases = JSON.parse(co.aliases || "[]"); } catch (_) {}

    for (const alias of aliases) {
      if (alias.toLowerCase() === needle) return co;
    }
  }

  return null;
}


function confidenceFromCount(count) {
  if (count >= 8) return "very_high";
  if (count >= 4) return "high";
  if (count >= 2) return "medium";
  return "low";
}


function mergeEventIds(existingJson, newJson) {
  let existing = [];
  let incoming = [];

  try { existing = JSON.parse(existingJson || "[]"); } catch (_) {}
  try { incoming = JSON.parse(newJson || "[]"); } catch (_) {}

  const merged = [...new Set([...existing, ...incoming])];
  return JSON.stringify(merged);
}


async function runGraphWorker(archiveDb, intelligenceDb, config) {
  const loopDelay = config.workers?.graphWorkerLoopDelayMs ?? 90000;

  const getRelationshipFacts = intelligenceDb.prepare(`
    SELECT * FROM company_facts WHERE type = 'relationship'
  `);

  const recordEvents = intelligenceDb.prepare(
    `INSERT INTO worker_events (worker) VALUES ('graph')`
  );
  const pruneEvents = intelligenceDb.prepare(
    `DELETE FROM worker_events WHERE worker = 'graph' AND completed_at < datetime('now', '-1 hour')`
  );

  const getCompanies = intelligenceDb.prepare("SELECT * FROM tracked_companies");

  const getCompanyById = intelligenceDb.prepare("SELECT * FROM tracked_companies WHERE id = ?");

  const getExistingRel = intelligenceDb.prepare(`
    SELECT * FROM company_relationships
    WHERE from_company_id = ? AND relationship_type = ? AND to_entity = ?
  `);

  const insertRel = intelligenceDb.prepare(`
    INSERT INTO company_relationships
      (from_company_id, relationship_type, to_entity, to_company_id, confidence, confirmation_count, first_seen_at, last_seen_at, supporting_event_ids)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
  `);

  const updateRel = intelligenceDb.prepare(`
    UPDATE company_relationships
    SET confirmation_count = ?,
        confidence = ?,
        last_seen_at = ?,
        supporting_event_ids = ?
    WHERE from_company_id = ? AND relationship_type = ? AND to_entity = ?
  `);

  const getExistingReciprocal = intelligenceDb.prepare(`
    SELECT id FROM company_relationships
    WHERE from_company_id = ? AND relationship_type = ? AND to_entity = ?
  `);

  const insertReciprocal = intelligenceDb.prepare(`
    INSERT OR IGNORE INTO company_relationships
      (from_company_id, relationship_type, to_entity, to_company_id, confidence, confirmation_count, first_seen_at, last_seen_at, supporting_event_ids)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
  `);


  while (true) {
    try {
      const facts = getRelationshipFacts.all();
      const companies = getCompanies.all();

      let upserted = 0;
      let reciprocals = 0;

      const processAll = intelligenceDb.transaction(() => {
        for (const fact of facts) {
          const parsed = parseClaim(fact.claim);
          if (!parsed) continue;

          const { toEntity, relationshipType } = parsed;

          const resolved = resolveCompany(toEntity, companies);
          const toCompanyId = resolved ? resolved.id : null;

          const existing = getExistingRel.get(fact.company_id, relationshipType, toEntity);
          const now = new Date().toISOString();

          let finalCount;
          let finalFirst;
          let finalLast;
          let finalEventIds;

          if (existing) {
            // derive count from the fact itself, not accumulate — avoids compounding on re-runs
            finalCount = fact.confirmation_count;
            finalFirst = existing.first_seen_at;
            finalLast = fact.last_seen_at || now;
            finalEventIds = mergeEventIds(existing.supporting_event_ids, fact.supporting_event_ids);

            updateRel.run(
              finalCount,
              confidenceFromCount(finalCount),
              finalLast,
              finalEventIds,
              fact.company_id,
              relationshipType,
              toEntity
            );
          } else {
            finalCount = fact.confirmation_count || 1;
            finalFirst = fact.first_seen_at || now;
            finalLast = fact.last_seen_at || now;
            finalEventIds = fact.supporting_event_ids || "[]";

            insertRel.run(
              fact.company_id,
              relationshipType,
              toEntity,
              toCompanyId,
              confidenceFromCount(finalCount),
              finalCount,
              finalFirst,
              finalLast,
              finalEventIds
            );
          }

          upserted++;


          // reciprocal edge — only where both companies are tracked
          if (toCompanyId) {
            const fromCompany = getCompanyById.get(fact.company_id);
            if (!fromCompany) continue;

            const recipType = RECIPROCAL[relationshipType] || "partner";
            const recipToEntity = fromCompany.name;

            const recipExists = getExistingReciprocal.get(toCompanyId, recipType, recipToEntity);

            if (!recipExists) {
              insertReciprocal.run(
                toCompanyId,
                recipType,
                recipToEntity,
                fact.company_id,
                confidenceFromCount(finalCount),
                finalCount,
                finalFirst,
                finalLast,
                finalEventIds
              );

              reciprocals++;
            }
          }
        }
      });

      processAll();

      // record one event per edge upserted so the rate tracks actual work
      if (upserted > 0) {
        const insertMany = intelligenceDb.transaction((n) => {
          for (let i = 0; i < n; i++) recordEvents.run();
        });
        insertMany(upserted);
        pruneEvents.run();
      }

      if (upserted > 0 || reciprocals > 0) {
        console.log(`[graph] cycle complete — ${upserted} edges upserted, ${reciprocals} reciprocals added`);
      } else {
        console.log("[graph] cycle complete — no new data");
      }

    } catch (err) {
      console.error("[graph] cycle error:", err.message);
    }

    await sleep(loopDelay);
  }
}


function sleep(ms) {
  return new Promise(r => setTimeout(r, ms));
}

module.exports = { runGraphWorker };