Snapshot Aggregation

Daily compression with merge logic and storage estimation for time-series data.

When to Use This Skill

• Raw event data grows too fast for direct queries

• Need daily snapshots for historical dashboards

• Pipeline runs multiple times per day and needs merge logic

• Want storage estimation for capacity planning

Core Concepts

Raw event data grows fast. Daily snapshots provide:

• Historical dashboards without querying millions of rows

• 90-day retention with minimal storage (~2-3KB/day)

• Multiple pipeline runs per day that merge correctly

• Storage estimation for capacity planning

Key insight: Multiple runs per day must merge correctly, not overwrite.

Implementation

TypeScript

interface DailySnapshot { snapshotDate: string; // YYYY-MM-DD totalArticles: number; totalEvents: number; totalClusters: number; avgRiskScore: number; maxRiskScore: number; categoryTotals: Record<string, number>; topHotspots: Hotspot[]; // Top 10 by risk keyEvents: KeyEvent[]; // Top 20 by risk countryStats: Record<string, CountryStats>; riskTrend: number; // vs previous day pipelineRuns: number; // How many runs merged createdAt: string; updatedAt: string; }

interface Hotspot { country: string; countryCode: string; lat: number; lon: number; riskScore: number; eventCount: number; summary: string; }

interface KeyEvent { id: string; title: string; country: string; riskScore: number; timestamp: string; }

interface CountryStats { events: number; avgRisk: number; maxRisk: number; topCategory: string; }

/**

• Aggregate pipeline results into a daily snapshot */ function aggregateToSnapshot( result: PipelineResult, rawEvents: RawEvent[] = [], date: string = new Date().toISOString().split('T')[0] ): DailySnapshot { const { predictions, stats } = result;

// Calculate category totals const categoryTotals: Record<string, number> = {}; for (const pred of predictions) { if (pred.categoryCounts) { for (const [cat, count] of Object.entries(pred.categoryCounts)) { categoryTotals[cat] = (categoryTotals[cat] || 0) + count; } } }

// Extract top 10 hotspots const topHotspots: Hotspot[] = predictions .slice(0, 10) .map(p => ({ country: getCountryName(p.countryCode), countryCode: p.countryCode, lat: p.lat, lon: p.lon, riskScore: p.riskScore, eventCount: p.eventCount, summary: p.summary, }));

// Extract key events (top 20 by risk) const keyEvents: KeyEvent[] = rawEvents .sort((a, b) => b.riskScore - a.riskScore) .slice(0, 20) .map(e => ({ id: e.id, title: truncate(e.title, 120), country: e.country, riskScore: e.riskScore, timestamp: new Date().toISOString(), }));

// Build country stats const countryStats: Record<string, CountryStats> = {}; for (const pred of predictions) { const country = pred.countryCode; if (!countryStats[country]) { countryStats[country] = { events: 0, avgRisk: 0, maxRisk: 0, topCategory: pred.category, }; } countryStats[country].events += pred.eventCount; countryStats[country].maxRisk = Math.max( countryStats[country].maxRisk, pred.riskScore ); }

// Aggregate stats const totalEvents = predictions.reduce((sum, p) => sum + p.eventCount, 0); const avgRiskScore = predictions.length > 0 ? Math.round(predictions.reduce((sum, p) => sum + p.riskScore, 0) / predictions.length) : 0; const maxRiskScore = predictions.length > 0 ? Math.max(...predictions.map(p => p.riskScore)) : 0;

return { snapshotDate: date, totalArticles: stats.totalFetched, totalEvents, totalClusters: stats.clustersFormed, avgRiskScore, maxRiskScore, categoryTotals, topHotspots, keyEvents, countryStats, riskTrend: 0, pipelineRuns: 1, createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), }; }

/**

• Merge multiple snapshots from the same day */ function mergeSnapshots( existing: DailySnapshot, incoming: DailySnapshot ): DailySnapshot { // Merge category totals (take max - later run has more complete data) const categoryTotals = { ...existing.categoryTotals }; for (const [cat, count] of Object.entries(incoming.categoryTotals)) { categoryTotals[cat] = Math.max(categoryTotals[cat] || 0, count); }

// Merge hotspots (keep top 10 by risk, dedupe by location) const allHotspots = [...existing.topHotspots, ...incoming.topHotspots]; const uniqueHotspots = dedupeByKey( allHotspots, h => ${h.lat.toFixed(1)},${h.lon.toFixed(1)} ); const topHotspots = uniqueHotspots .sort((a, b) => b.riskScore - a.riskScore) .slice(0, 10);

// Merge key events (keep top 20 by risk, dedupe by ID) const allEvents = [...existing.keyEvents, ...incoming.keyEvents]; const uniqueEvents = dedupeByKey(allEvents, e => e.id); const keyEvents = uniqueEvents .sort((a, b) => b.riskScore - a.riskScore) .slice(0, 20);

// Merge country stats (take max values) const countryStats = { ...existing.countryStats }; for (const [country, stats] of Object.entries(incoming.countryStats)) { if (!countryStats[country]) { countryStats[country] = stats; } else { countryStats[country] = { events: Math.max(countryStats[country].events, stats.events), avgRisk: Math.round( (countryStats[country].avgRisk + stats.avgRisk) / 2 ), maxRisk: Math.max(countryStats[country].maxRisk, stats.maxRisk), topCategory: stats.maxRisk > countryStats[country].maxRisk ? stats.topCategory : countryStats[country].topCategory, }; } }

return { ...existing, totalArticles: existing.totalArticles + incoming.totalArticles, totalEvents: Math.max(existing.totalEvents, incoming.totalEvents), totalClusters: Math.max(existing.totalClusters, incoming.totalClusters), avgRiskScore: Math.round( (existing.avgRiskScore + incoming.avgRiskScore) / 2 ), maxRiskScore: Math.max(existing.maxRiskScore, incoming.maxRiskScore), categoryTotals, topHotspots, keyEvents, countryStats, pipelineRuns: existing.pipelineRuns + 1, updatedAt: new Date().toISOString(), }; }

/**

• Estimate storage size of a snapshot */ function estimateSnapshotSize(snapshot: DailySnapshot): number { return JSON.stringify(snapshot).length; }

// Helpers function truncate(str: string, maxLen: number): string { return str.length > maxLen ? str.slice(0, maxLen - 3) + '...' : str; }

function dedupeByKey<T>(arr: T[], keyFn: (item: T) => string): T[] { const seen = new Set<string>(); return arr.filter(item => { const key = keyFn(item); if (seen.has(key)) return false; seen.add(key); return true; }); }

Usage Examples

Save Daily Snapshot

async function saveDailySnapshot(result: PipelineResult) { const today = new Date().toISOString().split('T')[0];

const newSnapshot = aggregateToSnapshot(result, result.rawEvents, today);

// Check for existing snapshot today const { data: existing } = await supabase .from('daily_snapshots') .select('*') .eq('snapshot_date', today) .single();

if (existing) { // Merge with existing const merged = mergeSnapshots(existing, newSnapshot); await supabase .from('daily_snapshots') .update(merged) .eq('snapshot_date', today); } else { // Insert new await supabase .from('daily_snapshots') .insert(newSnapshot); }

console.log(Snapshot size: ${estimateSnapshotSize(newSnapshot)} bytes); }

Database Schema

CREATE TABLE daily_snapshots ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), snapshot_date DATE UNIQUE NOT NULL, total_articles INTEGER DEFAULT 0, total_events INTEGER DEFAULT 0, avg_risk_score INTEGER DEFAULT 0, max_risk_score INTEGER DEFAULT 0, category_totals JSONB DEFAULT '{}', top_hotspots JSONB DEFAULT '[]', key_events JSONB DEFAULT '[]', country_stats JSONB DEFAULT '{}', pipeline_runs INTEGER DEFAULT 1, created_at TIMESTAMPTZ DEFAULT now(), updated_at TIMESTAMPTZ DEFAULT now() );

CREATE INDEX idx_snapshots_date ON daily_snapshots(snapshot_date DESC);

Best Practices

• Structured aggregation - Not just counts, but top-N lists and breakdowns

• Merge logic - Multiple runs per day combine correctly

• Deduplication - Hotspots/events dedupe by key before merging

• Storage estimation - Track size for capacity planning

• Retention policy - Auto-cleanup old snapshots (90 days)

Common Mistakes

• Simple daily counts (lose detail for dashboards)

• Overwriting on multiple runs (lose data)

• No deduplication in merge (duplicate hotspots)

• Unbounded arrays (memory issues)

• No retention policy (storage grows forever)

Related Patterns

• batch-processing - Process events before aggregation

• geographic-clustering - Cluster events for hotspots

• checkpoint-resume - Track which data has been aggregated