0pt/src/analytics/materialization/drift-detector.ts

/**
 * src/analytics/materialization/drift-detector.ts
 * ================================================
 * BI-P105 — is the pre-aggregate still telling the truth?
 *
 * The comparison is bucket-by-bucket between the aggregate and a recomputation
 * from the source over the same window. Three things make it a real detector
 * rather than a ceremony:
 *
 * 1. A MISSING BUCKET IS DRIFT. The nastiest pre-aggregate failure is not a
 *    wrong value, it is an absent row: late-arriving data, a materialized view
 *    that was created after the backfill, a refresh that failed silently. Every
 *    value it DOES hold is correct, so a value-only comparison passes while the
 *    total is wrong. Set difference is checked first, before any arithmetic.
 *
 * 2. ZERO BUCKETS IS `insufficient_data`, NEVER `match`. A comparison over an
 *    empty window finds no discrepancies — the same false green BI-P025 found
 *    in a quality check over an empty table. The router treats
 *    `insufficient_data` as "do not serve", not as "fine".
 *
 * 3. RELATIVE AND ABSOLUTE TOLERANCE BOTH HAVE TO HOLD. A 0.1% tolerance on a
 *    revenue bucket of 40M is 40 000 kr of invisible error; an absolute-only
 *    tolerance flags every rounding difference on a large bucket. Requiring
 *    both means a discrepancy has to be small in kronor AND small in
 *    proportion.
 */

import type { DriftTolerance, DriftVerdict, MaterializationStatus } from './materialization';

/** One aggregated row, keyed by its dimension tuple. */
export interface AggregateBucket {
  /** Stable key for the dimension tuple, e.g. "2026-09-01|SE". */
  key: string;
  /** metric name -> value. */
  values: Record<string, number>;
}

export interface DriftDiscrepancy {
  key: string;
  metric: string;
  source_value: number;
  aggregate_value: number;
  absolute_delta: number;
  /** |delta| / |source|, or Infinity when the source value is 0 and the aggregate is not. */
  relative_delta: number;
}

export interface DriftReport {
  verdict: DriftVerdict;
  checked_at: string;
  buckets_compared: number;
  /** In the source recomputation, absent from the aggregate. Always drift. */
  missing_in_aggregate: string[];
  /** In the aggregate, absent from the source. Always drift (deleted rows). */
  extra_in_aggregate: string[];
  discrepancies: DriftDiscrepancy[];
  worst?: DriftDiscrepancy;
  tolerance: DriftTolerance;
  /**
   * What a caller would have been told if it had trusted the aggregate: the
   * total the aggregate reports vs the total the source computes, per metric.
   * This is the number that makes a drift report actionable — "0.4% off" means
   * nothing until it is "1 284 500 kr reported, 1 271 900 kr true".
   */
  totals: Record<string, { source: number; aggregate: number; delta: number }>;
}

const MAX_REPORTED_DISCREPANCIES = 50;

function relative(sourceValue: number, delta: number): number {
  if (sourceValue === 0) return delta === 0 ? 0 : Number.POSITIVE_INFINITY;
  return Math.abs(delta / sourceValue);
}

/**
 * Compare a recomputation from source against the pre-aggregate.
 *
 * `fromSource` is the truth. Both sides must cover the same window; the caller
 * is responsible for that, and `missing_in_aggregate` will catch it loudly if
 * they do not.
 */
export function detectDrift(args: {
  fromSource: AggregateBucket[];
  fromAggregate: AggregateBucket[];
  tolerance: DriftTolerance;
  metrics: string[];
  now?: Date;
}): DriftReport {
  const { fromSource, fromAggregate, tolerance, metrics } = args;
  const checked_at = (args.now ?? new Date()).toISOString();
  const sourceByKey = new Map(fromSource.map((b) => [b.key, b]));
  const aggByKey = new Map(fromAggregate.map((b) => [b.key, b]));

  const missing_in_aggregate = [...sourceByKey.keys()].filter((k) => !aggByKey.has(k)).sort();
  const extra_in_aggregate = [...aggByKey.keys()].filter((k) => !sourceByKey.has(k)).sort();

  const discrepancies: DriftDiscrepancy[] = [];
  let buckets_compared = 0;
  const totals: DriftReport['totals'] = {};
  for (const m of metrics) totals[m] = { source: 0, aggregate: 0, delta: 0 };

  for (const [key, sourceBucket] of sourceByKey) {
    for (const m of metrics) totals[m].source += sourceBucket.values[m] ?? 0;
    const aggBucket = aggByKey.get(key);
    if (!aggBucket) continue;
    buckets_compared += 1;
    for (const m of metrics) {
      const sv = sourceBucket.values[m] ?? 0;
      const av = aggBucket.values[m] ?? 0;
      const delta = av - sv;
      const rel = relative(sv, delta);
      if (Math.abs(delta) > tolerance.absolute || rel > tolerance.relative) {
        discrepancies.push({
          key,
          metric: m,
          source_value: sv,
          aggregate_value: av,
          absolute_delta: delta,
          relative_delta: rel,
        });
      }
    }
  }
  for (const bucket of aggByKey.values()) {
    for (const m of metrics) totals[m].aggregate += bucket.values[m] ?? 0;
  }
  for (const m of metrics) totals[m].delta = totals[m].aggregate - totals[m].source;

  discrepancies.sort((a, b) => Math.abs(b.absolute_delta) - Math.abs(a.absolute_delta));

  // A comparison that compared nothing has not proved anything. This is the
  // BI-P025 false green, restated for pre-aggregates.
  const verdict: DriftVerdict =
    buckets_compared === 0 && missing_in_aggregate.length === 0 && extra_in_aggregate.length === 0
      ? 'insufficient_data'
      : discrepancies.length > 0 || missing_in_aggregate.length > 0 || extra_in_aggregate.length > 0
        ? 'drifted'
        : 'match';

  const report: DriftReport = {
    verdict,
    checked_at,
    buckets_compared,
    missing_in_aggregate,
    extra_in_aggregate,
    discrepancies: discrepancies.slice(0, MAX_REPORTED_DISCREPANCIES),
    tolerance,
    totals,
  };
  if (discrepancies.length > 0) report.worst = discrepancies[0];
  return report;
}

/**
 * Fold a drift report into the runtime status. A drifted aggregate is taken
 * OUT of service here rather than merely annotated: leaving it `serving` with
 * a bad verdict relies on every future reader remembering to check the
 * verdict, and one that forgets serves the wrong number.
 *
 * `insufficient_data` also stops service. It means the check could not prove
 * anything, and an unproven aggregate is not a current one.
 */
export function applyDriftReport(status: MaterializationStatus, report: DriftReport): MaterializationStatus {
  return {
    ...status,
    state: report.verdict === 'match' ? status.state : 'drifted',
    last_drift_check_at: report.checked_at,
    last_drift_verdict: report.verdict,
  };
}