runtime/observabilityEngine.js

function correlateSignals(inputs) {
  const lockEvents = inputs?.lockEvents || [];
  const planEvents = inputs?.queryPlanEvents || [];
  const replicationEvents = inputs?.replicationEvents || [];

  const incidents = [];
  const now = Date.now();

  const highWaiters = lockEvents.filter((event) => event.waitMs >= 800);
  if (highWaiters.length) {
    incidents.push({
      type: "lock_pressure",
      severity: "medium",
      evidence: highWaiters.map((e) => e.session),
      summary: "blocking chain or high wait duration detected",
      ts: now,
    });
  }

  const slowRegressions = planEvents.filter((p) => p.p95Ms && p.p95Ms > 800);
  if (slowRegressions.length) {
    incidents.push({
      type: "query_regression",
      severity: "high",
      evidence: slowRegressions.map((e) => e.queryId),
      summary: "runtime p95 regressions detected",
      ts: now,
    });
  }

  if (replicationEvents.some((r) => r.lagSeconds > 5)) {
    incidents.push({
      type: "replication_anomaly",
      severity: "medium",
      evidence: replicationEvents.map((r) => r.topology),
      summary: "replication lag risk above healthy threshold",
      ts: now,
    });
  }

  const causalGraph = incidents.map((incident, index) => ({
    id: `node-${index + 1}`,
    type: incident.type,
    severity: incident.severity,
    ts: incident.ts || now,
    evidence: incident.evidence,
  }));
  const edges = [];
  for (let i = 1; i < causalGraph.length; i++) {
    edges.push({
      from: causalGraph[i - 1].id,
      to: causalGraph[i].id,
      relation: "observed_sequence",
    });
  }
  const mttdMs = incidents.length ? 420000 : 0;
  const mttrMs = incidents.length ? 1540000 : 0;
  return {
    severity: incidents.some((x) => x.severity === "high") ? "high" : incidents.length ? "medium" : "low",
    incidentCount: incidents.length,
    incidents,
    causalGraph: {
      nodes: causalGraph,
      edges,
    },
    mttdMinutes: Math.round(mttdMs / 60000),
    mttrMinutesEstimate: Math.round(mttrMs / 60000),
    recommendations: incidents.length
      ? ["Prioritize lock chains", "validate plan stability in last 60m", "check replication lag telemetry"]
      : ["No immediate cross-signal incident"],
  };
}

module.exports = {
  correlateSignals,
};