/** * 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): 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(); }); });