0pt/tests/analytics/materialization-drift.test.ts

/**
 * tests/analytics/materialization-drift.test.ts
 * ==============================================
 * BI-P105 — MATERIALIZATION / PRE-AGGREGATION.
 *
 * The DoD asks for measured improvement and documented freshness semantics.
 * The failure mode it does not name, and the one that actually costs money, is
 * a pre-aggregate that returns a confident wrong number. So every test here is
 * written as an attack on the router:
 *
 *   - a drifted aggregate must NOT serve, and the counterfactual (a naive
 *     router that only checks staleness) must serve it — otherwise the guard
 *     is not known to be doing anything;
 *   - a MISSING bucket must count as drift even though every value present is
 *     correct;
 *   - a comparison over an empty window must be `insufficient_data`, never
 *     `match`;
 *   - an aggregate that has never been verified must not serve, because on
 *     launch day it is exactly as unproven as a drifted one.
 */
import { describe, expect, it } from 'vitest';
import {
  applyDriftReport,
  detectDrift,
  type AggregateBucket,
} from '@/src/analytics/materialization/drift-detector';
import {
  assertSpec,
  describeRouting,
  materializationStaleness,
  selectQuerySource,
  type MaterializationSpec,
  type MaterializationStatus,
} from '@/src/analytics/materialization/materialization';
import type { AnalyticsQueryRequest } from '@/src/analytics/contract/query-request';

const NOW = new Date('2026-09-07T12:00:00.000Z');
const TOLERANCE = { absolute: 0.5, relative: 0.001 };

const SPEC: MaterializationSpec = {
  materialization_id: 'mv_sales_daily',
  tenant_id: 't_alpha',
  project_id: 'p_main',
  dataset: 'ds_sales',
  strategy: 'clickhouse_materialized_view',
  target_ref: 'opt_sales_daily',
  metrics: ['revenue', 'orders'],
  dimensions: ['order_date', 'region'],
  time_dimension: 'order_date',
  grain: 'day',
  max_staleness_seconds: 3600,
  drift_check_interval_seconds: 86_400,
  tolerance: TOLERANCE,
};

/** Fresh, recently verified, matching. The only state that may serve. */
function healthyStatus(): MaterializationStatus {
  return {
    materialization_id: SPEC.materialization_id,
    state: 'serving',
    data_watermark: '2026-09-07T11:45:00.000Z',
    last_refresh_at: '2026-09-07T11:46:00.000Z',
    last_drift_check_at: '2026-09-07T06:00:00.000Z',
    last_drift_verdict: 'match',
  };
}

const REQUEST: AnalyticsQueryRequest = {
  dataset: 'ds_sales',
  metrics: ['revenue'],
  dimensions: ['order_date'],
  filters: [{ dimension: 'region', operator: 'eq', values: ['SE'] }],
  timeRange: { from: '2026-09-01T00:00:00.000Z', to: '2026-09-07T00:00:00.000Z', grain: 'day' },
};

/**
 * The router someone writes when they have thought about staleness but not
 * about correctness. It exists here as the counterfactual: it must SERVE the
 * drifted aggregate that `selectQuerySource` refuses.
 */
function naiveRouter(status: MaterializationStatus, now: Date): 'materialization' | 'base_table' {
  const staleness = materializationStaleness(status, now) ?? 0;
  return staleness <= SPEC.max_staleness_seconds ? 'materialization' : 'base_table';
}

function buckets(spec: Record<string, [number, number]>): AggregateBucket[] {
  return Object.entries(spec).map(([key, [revenue, orders]]) => ({ key, values: { revenue, orders } }));
}

const SOURCE = buckets({
  '2026-09-01|SE': [120_000, 410],
  '2026-09-02|SE': [131_500, 442],
  '2026-09-03|SE': [98_250, 305],
  '2026-09-04|SE': [140_000, 480],
});

describe('BI-P105 drift detection', () => {
  it('an identical aggregate matches', () => {
    const report = detectDrift({
      fromSource: SOURCE,
      fromAggregate: SOURCE,
      tolerance: TOLERANCE,
      metrics: ['revenue', 'orders'],
      now: NOW,
    });
    expect(report.verdict).toBe('match');
    expect(report.buckets_compared).toBe(4);
    expect(report.totals.revenue.delta).toBe(0);
  });

  it('a value beyond tolerance is drift, and the report says what a caller would have been told', () => {
    const drifted = buckets({
      '2026-09-01|SE': [120_000, 410],
      '2026-09-02|SE': [131_500, 442],
      '2026-09-03|SE': [98_250, 305],
      // A source row was deleted (GDPR erasure) but the aggregate still counts it.
      '2026-09-04|SE': [152_400, 495],
    });
    const report = detectDrift({
      fromSource: SOURCE,
      fromAggregate: drifted,
      tolerance: TOLERANCE,
      metrics: ['revenue', 'orders'],
      now: NOW,
    });
    expect(report.verdict).toBe('drifted');
    expect(report.worst?.key).toBe('2026-09-04|SE');
    expect(report.worst?.absolute_delta).toBe(12_400);
    expect(report.totals.revenue.aggregate - report.totals.revenue.source).toBe(12_400);
  });

  it('a MISSING bucket is drift even though every value present is correct', () => {
    const lateArriving = SOURCE.filter((b) => b.key !== '2026-09-04|SE');
    const report = detectDrift({
      fromSource: SOURCE,
      fromAggregate: lateArriving,
      tolerance: TOLERANCE,
      metrics: ['revenue', 'orders'],
      now: NOW,
    });
    expect(report.discrepancies).toHaveLength(0); // every compared value agrees
    expect(report.verdict).toBe('drifted'); // and it is still wrong
    expect(report.missing_in_aggregate).toEqual(['2026-09-04|SE']);
    expect(report.totals.revenue.delta).toBe(-140_000);
  });

  it('an EXTRA bucket is drift — the source rows are gone, the aggregate is not', () => {
    const stale = [...SOURCE, { key: '2026-08-31|SE', values: { revenue: 90_000, orders: 300 } }];
    const report = detectDrift({
      fromSource: SOURCE,
      fromAggregate: stale,
      tolerance: TOLERANCE,
      metrics: ['revenue', 'orders'],
      now: NOW,
    });
    expect(report.verdict).toBe('drifted');
    expect(report.extra_in_aggregate).toEqual(['2026-08-31|SE']);
  });

  it('an EMPTY comparison is insufficient_data, not match', () => {
    const report = detectDrift({
      fromSource: [],
      fromAggregate: [],
      tolerance: TOLERANCE,
      metrics: ['revenue'],
      now: NOW,
    });
    expect(report.verdict).toBe('insufficient_data');
    expect(report.buckets_compared).toBe(0);
    // And it must not be allowed to serve on the strength of that.
    const status = applyDriftReport(healthyStatus(), report);
    expect(status.state).toBe('drifted');
  });

  it('both tolerances must hold — a large bucket cannot hide behind a relative bound', () => {
    const source = buckets({ 'b|SE': [40_000_000, 1] });
    const off = buckets({ 'b|SE': [40_020_000, 1] }); // 0.05%: inside relative, far outside absolute
    const report = detectDrift({
      fromSource: source,
      fromAggregate: off,
      tolerance: { absolute: 1000, relative: 0.001 },
      metrics: ['revenue'],
      now: NOW,
    });
    expect(report.verdict).toBe('drifted');
    expect(report.worst?.relative_delta).toBeCloseTo(0.0005, 6);
  });

  it('a bucket that should be zero and is not is drift, not a division error', () => {
    const report = detectDrift({
      fromSource: buckets({ 'b|SE': [0, 0] }),
      fromAggregate: buckets({ 'b|SE': [12, 1] }),
      tolerance: TOLERANCE,
      metrics: ['revenue'],
      now: NOW,
    });
    expect(report.verdict).toBe('drifted');
    expect(report.worst?.relative_delta).toBe(Number.POSITIVE_INFINITY);
  });
});

describe('BI-P105 routing — a pre-aggregate serves only while it is provably current and undrifted', () => {
  it('serves when everything is proved', () => {
    const decision = selectQuerySource({ request: REQUEST, spec: SPEC, status: healthyStatus(), now: NOW });
    expect(decision.source).toBe('materialization');
    expect(describeRouting(decision)).toMatchObject({
      served_from: 'materialization',
      materialization_id: 'mv_sales_daily',
      staleness_seconds: 900,
    });
  });

  it('a DRIFTED aggregate is refused — and the naive router serves it', () => {
    const drifted = applyDriftReport(healthyStatus(), {
      verdict: 'drifted',
      checked_at: '2026-09-07T11:00:00.000Z',
      buckets_compared: 4,
      missing_in_aggregate: [],
      extra_in_aggregate: [],
      discrepancies: [],
      tolerance: TOLERANCE,
      totals: {},
    });

    const guarded = selectQuerySource({ request: REQUEST, spec: SPEC, status: drifted, now: NOW });
    expect(guarded.source).toBe('base_table');
    expect(guarded).toMatchObject({ refusal: 'state_not_serving' });

    // The counterfactual: the same state, routed by staleness alone.
    expect(naiveRouter(drifted, NOW)).toBe('materialization');
  });

  it('a NEVER-VERIFIED aggregate is refused — and the naive router serves it', () => {
    const brandNew: MaterializationStatus = {
      materialization_id: SPEC.materialization_id,
      state: 'serving',
      data_watermark: '2026-09-07T11:59:00.000Z',
      last_refresh_at: '2026-09-07T11:59:00.000Z',
    };
    const guarded = selectQuerySource({ request: REQUEST, spec: SPEC, status: brandNew, now: NOW });
    expect(guarded).toMatchObject({ source: 'base_table', refusal: 'drift_never_verified' });
    expect(naiveRouter(brandNew, NOW)).toBe('materialization');
  });

  it('a verification older than the required interval is refused', () => {
    const status = { ...healthyStatus(), last_drift_check_at: '2026-09-05T06:00:00.000Z' };
    expect(selectQuerySource({ request: REQUEST, spec: SPEC, status, now: NOW })).toMatchObject({
      refusal: 'drift_verification_stale',
    });
  });

  it('an UNKNOWN watermark is refused rather than treated as current', () => {
    const status = { ...healthyStatus(), data_watermark: undefined };
    expect(materializationStaleness(status, NOW)).toBeUndefined();
    expect(selectQuerySource({ request: REQUEST, spec: SPEC, status, now: NOW })).toMatchObject({
      refusal: 'watermark_unknown',
    });
  });

  it('an aggregate older than its freshness contract is refused', () => {
    const status = { ...healthyStatus(), data_watermark: '2026-09-07T09:00:00.000Z' };
    expect(selectQuerySource({ request: REQUEST, spec: SPEC, status, now: NOW })).toMatchObject({
      refusal: 'staleness_exceeded',
    });
  });

  it('a question that reaches past the watermark is refused, not silently shortened', () => {
    const request = { ...REQUEST, timeRange: { ...REQUEST.timeRange, to: '2026-09-07T12:00:00.000Z' } };
    expect(selectQuerySource({ request, spec: SPEC, status: healthyStatus(), now: NOW })).toMatchObject({
      refusal: 'window_beyond_watermark',
    });
  });

  it('a filter on a dimension the aggregate summed across is refused', () => {
    const request = {
      ...REQUEST,
      filters: [{ dimension: 'sales_channel', operator: 'eq' as const, values: ['web'] }],
    };
    const decision = selectQuerySource({ request, spec: SPEC, status: healthyStatus(), now: NOW });
    expect(decision).toMatchObject({ source: 'base_table', refusal: 'coverage_miss' });
    expect((decision as { detail: string }).detail).toMatch(/already summed across it/);
  });

  it('a metric or dimension the aggregate does not carry is refused', () => {
    expect(
      selectQuerySource({
        request: { ...REQUEST, metrics: ['margin'] },
        spec: SPEC,
        status: healthyStatus(),
        now: NOW,
      }),
    ).toMatchObject({ refusal: 'coverage_miss' });
    expect(
      selectQuerySource({
        request: { ...REQUEST, dimensions: ['product'] },
        spec: SPEC,
        status: healthyStatus(),
        now: NOW,
      }),
    ).toMatchObject({ refusal: 'coverage_miss' });
  });

  it('a daily aggregate cannot answer an hourly question, but can answer a monthly one', () => {
    const hourly = { ...REQUEST, timeRange: { ...REQUEST.timeRange, grain: 'hour' as const } };
    expect(
      selectQuerySource({ request: hourly, spec: SPEC, status: healthyStatus(), now: NOW }),
    ).toMatchObject({ refusal: 'coverage_miss' });

    const monthly = { ...REQUEST, timeRange: { ...REQUEST.timeRange, grain: 'month' as const } };
    expect(
      selectQuerySource({ request: monthly, spec: SPEC, status: healthyStatus(), now: NOW }).source,
    ).toBe('materialization');
  });

  it('a WEEKLY aggregate cannot answer a monthly question — weeks do not compose into months', () => {
    const weekly = { ...SPEC, grain: 'week' as const };
    const monthly = { ...REQUEST, timeRange: { ...REQUEST.timeRange, grain: 'month' as const } };
    expect(
      selectQuerySource({ request: monthly, spec: weekly, status: healthyStatus(), now: NOW }),
    ).toMatchObject({ refusal: 'coverage_miss' });
  });

  it('a spec with no drift check interval is refused at declaration time', () => {
    expect(() => assertSpec({ ...SPEC, drift_check_interval_seconds: 0 })).toThrow(/never re-verified/);
    expect(() => assertSpec(SPEC)).not.toThrow();
  });
});