1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

// Comprehensive metric instrumentation example
 
import { Metrics, Counter, Gauge, Histogram } from '@opentelemetry/metrics';
 
class DataIntegrityMetrics {
    // Consistency metrics
    private readonly mismatchCount: Gauge;
    private readonly mismatchRate: Gauge;
    
    // Sync metrics
    private readonly syncOperations: Counter;
    private readonly syncDuration: Histogram;
    private readonly syncLag: Gauge;
    private readonly syncErrors: Counter;
    
    // Staleness metrics
    private readonly staleRecords: Gauge;
    private readonly maxStaleness: Gauge;
    
    constructor(private metrics: Metrics) {
        this.mismatchCount = metrics.createGauge('denorm.mismatch_count', {
            description: 'Number of records with mismatched denormalized values',
            unit: 'records',
        });
        
        this.mismatchRate = metrics.createGauge('denorm.mismatch_rate', {
            description: 'Percentage of records with mismatches',
            unit: 'percent',
        });
        
        this.syncOperations = metrics.createCounter('denorm.sync_operations_total', {
            description: 'Total number of sync operations performed',
        });
        
        this.syncDuration = metrics.createHistogram('denorm.sync_duration_seconds', {
            description: 'Duration of sync operations',
            unit: 'seconds',
            boundaries: [0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10],
        });
        
        this.syncLag = metrics.createGauge('denorm.sync_lag_seconds', {
            description: 'Time since oldest unsynced change',
            unit: 'seconds',
        });
        
        this.syncErrors = metrics.createCounter('denorm.sync_errors_total', {
            description: 'Total number of sync errors',
        });
        
        this.staleRecords = metrics.createGauge('denorm.stale_records', {
            description: 'Number of records with stale denormalized data',
            unit: 'records',
        });
        
        this.maxStaleness = metrics.createGauge('denorm.staleness_max_seconds', {
            description: 'Age of the oldest stale record',
            unit: 'seconds',
        });
    }
    
    // Called by consistency check queries
    recordConsistencyCheck(table: string, total: number, mismatches: number) {
        const labels = { table };
        
        this.mismatchCount.record(mismatches, labels);
        this.mismatchRate.record(total > 0 ? (mismatches / total) * 100 : 0, labels);
    }
    
    // Called by sync operations
    recordSyncOperation(table: string, duration: number, success: boolean) {
        const labels = { table, success: String(success) };
        
        this.syncOperations.add(1, labels);
        this.syncDuration.record(duration, labels);
        
        if (!success) {
            this.syncErrors.add(1, { table });
        }
    }
    
    // Called periodically to measure lag
    recordSyncLag(table: string, lagSeconds: number) {
        this.syncLag.record(lagSeconds, { table });
    }
    
    // Called by staleness analysis
    recordStaleness(table: string, staleCount: number, maxAgeSeconds: number) {
        this.staleRecords.record(staleCount, { table });
        this.maxStaleness.record(maxAgeSeconds, { table });
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

// Statistical anomaly detection for sync operations
 
class SyncAnomalyDetector {
    constructor(
        private metricsStore: MetricsStore,
        private alertService: AlertService
    ) {}
    
    async analyzeRecentActivity(tableName: string): Promise<AnomalyReport> {
        const anomalies: Anomaly[] = [];
        
        // Get recent metrics
        const currentHour = await this.metricsStore.getHourlyMetrics(tableName, 1);
        const historical = await this.metricsStore.getHourlyMetrics(tableName, 168); // 1 week
        
        // Volume anomaly detection using Z-score
        const volumeAnomaly = this.detectVolumeAnomaly(
            currentHour.syncCount,
            historical.map(h => h.syncCount)
        );
        if (volumeAnomaly) {
            anomalies.push(volumeAnomaly);
        }
        
        // Duration anomaly detection
        const durationAnomaly = this.detectDurationAnomaly(
            currentHour.avgDuration,
            historical.map(h => h.avgDuration)
        );
        if (durationAnomaly) {
            anomalies.push(durationAnomaly);
        }
        
        // Error rate anomaly
        const errorAnomaly = this.detectErrorRateAnomaly(
            currentHour.errorRate,
            historical.map(h => h.errorRate)
        );
        if (errorAnomaly) {
            anomalies.push(errorAnomaly);
        }
        
        return {
            tableName,
            analyzedAt: new Date(),
            anomalies,
            severity: this.calculateOverallSeverity(anomalies)
        };
    }
    
    private detectVolumeAnomaly(current: number, historical: number[]): Anomaly | null {
        const mean = historical.reduce((a, b) => a + b, 0) / historical.length;
        const stdDev = Math.sqrt(
            historical.reduce((sum, val) => sum + Math.pow(val - mean, 2), 0) / historical.length
        );
        
        const zScore = (current - mean) / stdDev;
        
        if (Math.abs(zScore) > 3) {  // 3 standard deviations
            return {
                type: 'volume',
                severity: zScore > 0 ? 'high_volume' : 'low_volume',
                currentValue: current,
                expectedRange: { min: mean - 2 * stdDev, max: mean + 2 * stdDev },
                zScore,
                message: zScore > 0
                    ? `Sync volume ${current} is unusually high (expected ~${mean.toFixed(0)})`
                    : `Sync volume ${current} is unusually low (expected ~${mean.toFixed(0)})`
            };
        }
        
        return null;
    }
    
    private detectDurationAnomaly(current: number, historical: number[]): Anomaly | null {
        const p95 = this.percentile(historical, 95);
        
        if (current > p95 * 2) {
            return {
                type: 'duration',
                severity: 'slow_sync',
                currentValue: current,
                threshold: p95,
                message: `Sync duration ${current}ms is 2x above p95 (${p95}ms)`
            };
        }
        
        return null;
    }
    
    private detectErrorRateAnomaly(current: number, historical: number[]): Anomaly | null {
        const maxHistorical = Math.max(...historical);
        const avgHistorical = historical.reduce((a, b) => a + b, 0) / historical.length;
        
        if (current > maxHistorical * 1.5 || (current > 0.01 && current > avgHistorical * 5)) {
            return {
                type: 'error_rate',
                severity: 'elevated_errors',
                currentValue: current,
                historicalMax: maxHistorical,
                message: `Error rate ${(current * 100).toFixed(2)}% is unusually high`
            };
        }
        
        return null;
    }
    
    private percentile(values: number[], p: number): number {
        const sorted = [...values].sort((a, b) => a - b);
        const index = Math.ceil((p / 100) * sorted.length) - 1;
        return sorted[index];
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

// Composite health score calculation
 
interface HealthComponent {
    name: string;
    weight: number;
    score: number;  // 0-100
    status: 'healthy' | 'degraded' | 'critical';
}
 
class DataIntegrityHealthScore {
    calculateOverallScore(components: HealthComponent[]): number {
        const totalWeight = components.reduce((sum, c) => sum + c.weight, 0);
        const weightedSum = components.reduce((sum, c) => sum + c.score * c.weight, 0);
        return Math.round(weightedSum / totalWeight);
    }
    
    async computeComponents(): Promise<HealthComponent[]> {
        return [
            await this.computeConsistencyScore(),
            await this.computeSyncPerformanceScore(),
            await this.computeErrorRateScore(),
            await this.computeStalenessScore(),
        ];
    }
    
    private async computeConsistencyScore(): Promise<HealthComponent> {
        const mismatchRate = await this.metrics.get('denorm.mismatch_rate');
        
        // Score: 100 at 0% mismatch, 0 at 1% or more
        const score = Math.max(0, 100 - (mismatchRate * 100));
        
        return {
            name: 'Data Consistency',
            weight: 40,  // Highest weight - this is the core metric
            score,
            status: score >= 99 ? 'healthy' : score >= 95 ? 'degraded' : 'critical'
        };
    }
    
    private async computeSyncPerformanceScore(): Promise<HealthComponent> {
        const p99Duration = await this.metrics.get('denorm.sync_duration_p99');
        const targetP99 = 1000;  // 1 second target
        
        // Score: 100 if at or below target, decreasing for slower
        const score = p99Duration <= targetP99 
            ? 100 
            : Math.max(0, 100 - ((p99Duration - targetP99) / targetP99) * 50);
        
        return {
            name: 'Sync Performance',
            weight: 20,
            score,
            status: score >= 80 ? 'healthy' : score >= 50 ? 'degraded' : 'critical'
        };
    }
    
    private async computeErrorRateScore(): Promise<HealthComponent> {
        const errorRate = await this.metrics.get('denorm.sync_error_rate');
        
        // Score: 100 at 0% errors, drops quickly with any errors
        const score = errorRate === 0 ? 100 : Math.max(0, 100 - (errorRate * 1000));
        
        return {
            name: 'Sync Reliability',
            weight: 25,
            score,
            status: errorRate === 0 ? 'healthy' : errorRate < 0.001 ? 'degraded' : 'critical'
        };
    }
    
    private async computeStalenessScore(): Promise<HealthComponent> {
        const maxStaleness = await this.metrics.get('denorm.staleness_max_seconds');
        const targetStaleness = 3600;  // 1 hour target
        
        // Score: 100 if fresh, decreasing with age
        const score = maxStaleness <= targetStaleness
            ? 100
            : Math.max(0, 100 - ((maxStaleness - targetStaleness) / targetStaleness) * 25);
        
        return {
            name: 'Data Freshness',
            weight: 15,
            score,
            status: score >= 90 ? 'healthy' : score >= 50 ? 'degraded' : 'critical'
        };
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

# Alert definitions for data integrity monitoring
 
alerts:
  - name: CriticalDataMismatch
    description: High rate of data inconsistencies detected
    severity: critical
    condition: |
      avg(denorm_mismatch_rate{table=~".+"}) > 0.01
    for: 5m
    labels:
      team: data-platform
      runbook: https://runbooks.example.com/data-integrity/critical-mismatch
    annotations:
      summary: "Critical: {{ $value | humanizePercentage }} data mismatch rate"
      description: |
        Mismatch rate has exceeded 1% for 5 minutes.
        This indicates widespread data inconsistency.
        Immediate investigation required.
        
        Affected tables: {{ $labels.table }}
        Current rate: {{ $value | humanizePercentage }}
        
        Investigation steps:
        1. Check recent deployments for trigger/sync changes
        2. Run detailed consistency check
        3. Review sync job logs for errors
        4. Check for database performance issues
 
  - name: SyncLagHigh
    description: Denormalized data sync is falling behind
    severity: warning
    condition: |
      max(denorm_sync_lag_seconds) > 3600
    for: 30m
    labels:
      team: data-platform
    annotations:
      summary: "Sync lag is {{ $value | humanizeDuration }}"
      description: |
        Denormalized data is falling behind source.
        Users may see stale or inconsistent data.
        
        Steps:
        1. Check sync job status
        2. Review queue depth if using event-driven sync
        3. Check for database locks or resource constraints
 
  - name: SyncJobFailed
    description: Batch synchronization job did not complete
    severity: warning
    condition: |
      absent(denorm_sync_job_last_success_timestamp) 
      or (time() - denorm_sync_job_last_success_timestamp) > 86400
    for: 0m
    labels:
      team: data-platform
    annotations:
      summary: "Sync job has not succeeded in 24 hours"
      description: |
        The batch sync job has not completed successfully.
        Data inconsistencies may accumulate.
        
        Check job logs and restart if necessary.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

-- Comprehensive audit log schema for forensic analysis
 
CREATE TABLE data_change_audit (
    audit_id            BIGSERIAL PRIMARY KEY,
    occurred_at         TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    
    -- What changed
    table_name          VARCHAR(100) NOT NULL,
    record_id           VARCHAR(100) NOT NULL,
    operation           VARCHAR(10) NOT NULL,  -- INSERT, UPDATE, DELETE
    
    -- Change details
    old_values          JSONB,
    new_values          JSONB,
    changed_columns     TEXT[],
    
    -- Context
    change_source       VARCHAR(50) NOT NULL,  -- 'trigger', 'sync_job', 'api', 'manual'
    transaction_id      BIGINT,
    session_user        VARCHAR(100),
    application_name    VARCHAR(100),
    
    -- For sync operations
    sync_job_id         UUID,
    source_table        VARCHAR(100),
    source_record_id    VARCHAR(100),
    
    -- Indexing for forensic queries
    INDEX idx_audit_table_record (table_name, record_id),
    INDEX idx_audit_time (occurred_at),
    INDEX idx_audit_sync_job (sync_job_id)
);
 
-- Enable automatic audit triggering
CREATE OR REPLACE FUNCTION audit_trigger_function()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO data_change_audit (
        table_name, record_id, operation,
        old_values, new_values, changed_columns,
        change_source, transaction_id, session_user, application_name
    )
    VALUES (
        TG_TABLE_NAME,
        COALESCE(NEW.id::TEXT, OLD.id::TEXT),
        TG_OP,
        CASE WHEN TG_OP != 'INSERT' THEN to_jsonb(OLD) END,
        CASE WHEN TG_OP != 'DELETE' THEN to_jsonb(NEW) END,
        CASE WHEN TG_OP = 'UPDATE' THEN
            ARRAY(SELECT key FROM jsonb_each(to_jsonb(OLD)) 
                  WHERE to_jsonb(OLD)->>key IS DISTINCT FROM to_jsonb(NEW)->>key)
        END,
        current_setting('app.change_source', TRUE),
        txid_current(),
        session_user,
        current_setting('application_name')
    );
    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;
 
-- Apply to denormalized tables
CREATE TRIGGER orders_audit
    AFTER INSERT OR UPDATE OR DELETE ON orders
    FOR EACH ROW EXECUTE FUNCTION audit_trigger_function();
 
-- Forensic query examples
 
-- Find when a specific record diverged
SELECT 
    occurred_at,
    operation,
    changed_columns,
    old_values->>'customer_name' AS old_name,
    new_values->>'customer_name' AS new_name,
    change_source
FROM data_change_audit
WHERE table_name = 'orders' 
  AND record_id = '12345'
  AND 'customer_name' = ANY(changed_columns)
ORDER BY occurred_at DESC;
 
-- Find all unsourced changes (not from sync)
SELECT *
FROM data_change_audit
WHERE table_name = 'orders'
  AND change_source NOT IN ('trigger', 'sync_job')
  AND 'customer_name' = ANY(changed_columns)
ORDER BY occurred_at DESC
LIMIT 100;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

// Incident categorization for continuous improvement
 
interface DataIntegrityIncident {
    id: string;
    detectedAt: Date;
    resolvedAt: Date | null;
    affectedTable: string;
    affectedRecords: number;
    impact: 'low' | 'medium' | 'high' | 'critical';
    
    // Root cause analysis
    category: IncidentCategory;
    rootCause: string;
    
    // Resolution
    resolution: 'auto_corrected' | 'manual_fix' | 'code_fix' | 'unresolved';
    preventionActions: string[];
}
 
enum IncidentCategory {
    TRIGGER_BUG = 'trigger_bug',
    MISSING_SYNC_PATH = 'missing_sync_path',
    CONCURRENCY_RACE = 'concurrency_race',
    SYNC_JOB_FAILURE = 'sync_job_failure',
    MANUAL_DATA_CHANGE = 'manual_data_change',
    SCHEMA_MIGRATION = 'schema_migration',
    EXTERNAL_SYSTEM = 'external_system',
    UNKNOWN = 'unknown'
}
 
class IncidentAnalyzer {
    async generateImprovementReport(timeRange: DateRange): Promise<ImprovementReport> {
        const incidents = await this.getIncidents(timeRange);
        
        // Group by category
        const byCategory = this.groupByCategory(incidents);
        
        // Calculate impact per category
        const categoryImpact = Object.entries(byCategory).map(([category, incidents]) => ({
            category,
            incidentCount: incidents.length,
            totalAffectedRecords: incidents.reduce((sum, i) => sum + i.affectedRecords, 0),
            criticalCount: incidents.filter(i => i.impact === 'critical').length,
            avgTimeToResolve: this.avgResolutionTime(incidents),
            trend: this.calculateTrend(category, timeRange)
        }));
        
        // Sort by impact
        categoryImpact.sort((a, b) => 
            (b.criticalCount * 100 + b.incidentCount) - (a.criticalCount * 100 + a.incidentCount)
        );
        
        return {
            timeRange,
            totalIncidents: incidents.length,
            categoryBreakdown: categoryImpact,
            topPriorities: categoryImpact.slice(0, 3).map(c => ({
                category: c.category,
                recommendedActions: this.getRecommendedActions(c.category),
                expectedImpact: this.estimateFixImpact(c)
            })),
            trendsImproving: categoryImpact.filter(c => c.trend < 0).map(c => c.category),
            trendsDegrading: categoryImpact.filter(c => c.trend > 0).map(c => c.category)
        };
    }
    
    private getRecommendedActions(category: IncidentCategory): string[] {
        switch (category) {
            case IncidentCategory.TRIGGER_BUG:
                return [
                    'Review trigger test coverage',
                    'Add integration tests for edge cases',
                    'Implement trigger change review process'
                ];
            case IncidentCategory.MISSING_SYNC_PATH:
                return [
                    'Audit all update paths for denormalized data',
                    'Document denormalization contracts',
                    'Add automated coverage checking'
                ];
            case IncidentCategory.CONCURRENCY_RACE:
                return [
                    'Review transaction isolation levels',
                    'Add pessimistic locking where needed',
                    'Implement idempotent sync operations'
                ];
            // ... other categories
            default:
                return ['Investigate specific incidents for patterns'];
        }
    }
}