← back to Watches
database/etl-pipeline.js
650 lines
/**
* ETL Pipeline for Omega Watches Database
*
* Features:
* - Extracts data from JSON files
* - Transforms and validates data
* - Loads into PostgreSQL with error handling
* - Tracks ETL job metrics
* - Supports incremental updates
* - Connection pooling
* - Batch processing for performance
*/
import pg from 'pg';
import fs from 'fs';
import path from 'path';
import { fileURLToPath } from 'url';
import crypto from 'crypto';
const { Pool } = pg;
const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);
// ============================================================================
// CONNECTION POOL CONFIGURATION
// ============================================================================
const pool = new Pool({
host: process.env.DB_HOST || 'localhost',
port: process.env.DB_PORT || 5432,
database: process.env.DB_NAME || 'omega_watches',
user: process.env.DB_USER || 'postgres',
password: process.env.DB_PASSWORD || 'postgres',
// Pool settings optimized for ETL
max: 30, // Higher for bulk operations
min: 5,
idleTimeoutMillis: 30000,
connectionTimeoutMillis: 5000,
statement_timeout: 60000, // 60s for complex ETL queries
query_timeout: 60000,
application_name: 'omega_etl_pipeline'
});
// ============================================================================
// MATERIAL & ENUM MAPPINGS
// ============================================================================
const MATERIAL_MAPPING = {
'Stainless steel': 'stainless_steel',
'Stainless steel with white outer case': 'stainless_steel',
'Stainless steel and gold (two-tone)': 'two_tone',
'18K gold': 'gold_18k',
'18K yellow gold': 'gold_18k',
'18K white gold': 'canopus_gold',
'18K gold or stainless steel': 'gold_18k',
'Stainless steel or gold': 'stainless_steel',
'Black ceramic (ZrO2)': 'ceramic',
'Ceramic': 'ceramic',
'Titanium': 'titanium',
'Platinum': 'platinum',
'Bronze': 'bronze',
'Omega Canopus Gold (18K white gold)': 'canopus_gold',
'Omega Sedna Gold': 'sedna_gold',
'Omega Moonshine Gold': 'moonshine_gold'
};
const MOVEMENT_TYPE_MAPPING = {
'automatic': 'automatic',
'manual': 'manual',
'quartz': 'quartz',
'co-axial': 'co_axial'
};
// ============================================================================
// HELPER FUNCTIONS
// ============================================================================
function slugify(text) {
return text
.toLowerCase()
.replace(/[^\w\s-]/g, '')
.replace(/\s+/g, '-')
.replace(/-+/g, '-')
.trim();
}
function extractNumeric(value, pattern = /(\d+\.?\d*)/)) {
if (!value) return null;
const match = value.toString().match(pattern);
return match ? parseFloat(match[1]) : null;
}
function mapCaseMaterial(material) {
if (!material) return null;
return MATERIAL_MAPPING[material] || 'stainless_steel';
}
function mapMovementType(movement) {
if (!movement) return null;
const lower = movement.toLowerCase();
if (lower.includes('automatic') || lower.includes('co-axial')) return 'automatic';
if (lower.includes('manual')) return 'manual';
if (lower.includes('quartz')) return 'quartz';
return 'automatic'; // Default assumption
}
function calculateDataQuality(watch) {
let score = 0;
let maxScore = 0;
// Required fields (40 points)
maxScore += 40;
if (watch.model) score += 10;
if (watch.series) score += 10;
if (watch.reference) score += 10;
if (watch.yearIntroduced) score += 10;
// Price history (30 points)
maxScore += 30;
if (watch.priceHistory && watch.priceHistory.length > 0) {
score += 10;
if (watch.priceHistory.length >= 5) score += 10;
if (watch.priceHistory.length >= 10) score += 10;
}
// Specifications (20 points)
maxScore += 20;
if (watch.specifications) {
score += 5;
if (watch.specifications.movement) score += 5;
if (watch.specifications.caseSize) score += 5;
if (watch.specifications.caseMaterial) score += 5;
}
// Additional metadata (10 points)
maxScore += 10;
if (watch.description) score += 5;
if (watch.imageUrl) score += 5;
return Math.round((score / maxScore) * 100);
}
// ============================================================================
// ETL JOB TRACKING
// ============================================================================
class ETLJob {
constructor(jobName, jobType) {
this.id = null;
this.jobName = jobName;
this.jobType = jobType;
this.startTime = null;
this.stats = {
processed: 0,
inserted: 0,
updated: 0,
failed: 0
};
}
async start(client) {
this.startTime = Date.now();
const result = await client.query(`
INSERT INTO omega.etl_jobs (job_name, job_type, status, started_at)
VALUES ($1, $2, 'running', NOW())
RETURNING id
`, [this.jobName, this.jobType]);
this.id = result.rows[0].id;
console.log(`\n🚀 ETL Job Started: ${this.jobName} (${this.id})`);
}
async updateProgress(client, status = 'running') {
if (!this.id) return;
await client.query(`
UPDATE omega.etl_jobs
SET status = $1,
records_processed = $2,
records_inserted = $3,
records_updated = $4,
records_failed = $5
WHERE id = $6
`, [
status,
this.stats.processed,
this.stats.inserted,
this.stats.updated,
this.stats.failed,
this.id
]);
}
async complete(client, error = null) {
if (!this.id) return;
const executionTime = Date.now() - this.startTime;
const status = error ? 'failed' : 'completed';
await client.query(`
UPDATE omega.etl_jobs
SET status = $1,
completed_at = NOW(),
records_processed = $2,
records_inserted = $3,
records_updated = $4,
records_failed = $5,
execution_time_ms = $6,
error_message = $7
WHERE id = $8
`, [
status,
this.stats.processed,
this.stats.inserted,
this.stats.updated,
this.stats.failed,
executionTime,
error ? error.message : null,
this.id
]);
console.log(`\n✅ ETL Job ${status.toUpperCase()}: ${this.jobName}`);
console.log(` Execution Time: ${(executionTime / 1000).toFixed(2)}s`);
console.log(` Processed: ${this.stats.processed}`);
console.log(` Inserted: ${this.stats.inserted}`);
console.log(` Updated: ${this.stats.updated}`);
console.log(` Failed: ${this.stats.failed}`);
}
}
// ============================================================================
// DATA TRANSFORMERS
// ============================================================================
class WatchTransformer {
static transform(watch) {
return {
slug: watch.id,
model: watch.model,
series: watch.series,
reference: watch.reference,
year_introduced: watch.yearIntroduced,
description: watch.description || null,
detailed_description: watch.detailedDescription || null,
image_url: watch.imageUrl || null,
is_limited_edition: watch.isLimitedEdition || false,
limited_edition_count: watch.limitedEditionCount || null,
data_quality_score: calculateDataQuality(watch)
};
}
}
class SpecificationTransformer {
static transform(spec) {
if (!spec) return null;
return {
case_size: spec.caseSize || null,
case_diameter: extractNumeric(spec.caseSize),
case_material: mapCaseMaterial(spec.caseMaterial),
case_thickness: spec.thickness || null,
case_thickness_mm: extractNumeric(spec.thickness),
lug_width: spec.lugWidth || null,
lug_width_mm: extractNumeric(spec.lugWidth),
movement_calibre: spec.movement || null,
movement_type: mapMovementType(spec.movement),
power_reserve: spec.powerReserve || null,
power_reserve_hours: extractNumeric(spec.powerReserve),
water_resistance: spec.waterResistance || null,
water_resistance_meters: extractNumeric(spec.waterResistance),
crystal_type: spec.crystal || null,
dial_color: spec.dialColor || null,
bezel_type: spec.bezel || null,
weight: spec.weight || null,
weight_grams: extractNumeric(spec.weight)
};
}
}
class PriceHistoryTransformer {
static transform(pricePoint) {
return {
year: pricePoint.year,
price: pricePoint.price,
condition: pricePoint.condition || 'new',
source: pricePoint.source || null,
currency: pricePoint.currency || 'USD',
is_estimate: pricePoint.source ?
pricePoint.source.toLowerCase().includes('estimate') : false,
confidence_level: pricePoint.confidence || 3
};
}
}
// ============================================================================
// ETL PIPELINE
// ============================================================================
class ETLPipeline {
constructor() {
this.featureCache = new Map();
}
async loadJSON(filePath) {
console.log(`📂 Loading JSON from: ${filePath}`);
const rawData = fs.readFileSync(filePath, 'utf8');
const data = JSON.parse(rawData);
console.log(` Found ${data.watches.length} watches`);
return data;
}
async ensureFeature(client, featureName, category = 'general') {
if (this.featureCache.has(featureName)) {
return this.featureCache.get(featureName);
}
const result = await client.query(`
INSERT INTO omega.features (name, category)
VALUES ($1, $2)
ON CONFLICT (name) DO UPDATE SET name = EXCLUDED.name
RETURNING id
`, [featureName, category]);
const featureId = result.rows[0].id;
this.featureCache.set(featureName, featureId);
return featureId;
}
async upsertWatch(client, watch) {
const transformed = WatchTransformer.transform(watch);
// Check if watch exists
const existing = await client.query(
'SELECT id FROM omega.watches WHERE slug = $1',
[transformed.slug]
);
let watchId;
if (existing.rows.length > 0) {
// Update existing watch
watchId = existing.rows[0].id;
await client.query(`
UPDATE omega.watches
SET model = $2, series = $3, reference = $4, year_introduced = $5,
description = $6, detailed_description = $7, image_url = $8,
is_limited_edition = $9, limited_edition_count = $10,
data_quality_score = $11, updated_at = NOW()
WHERE id = $1
`, [
watchId, transformed.model, transformed.series, transformed.reference,
transformed.year_introduced, transformed.description,
transformed.detailed_description, transformed.image_url,
transformed.is_limited_edition, transformed.limited_edition_count,
transformed.data_quality_score
]);
return { id: watchId, action: 'updated' };
} else {
// Insert new watch
const result = await client.query(`
INSERT INTO omega.watches (
slug, model, series, reference, year_introduced,
description, detailed_description, image_url,
is_limited_edition, limited_edition_count, data_quality_score
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
RETURNING id
`, [
transformed.slug, transformed.model, transformed.series,
transformed.reference, transformed.year_introduced,
transformed.description, transformed.detailed_description,
transformed.image_url, transformed.is_limited_edition,
transformed.limited_edition_count, transformed.data_quality_score
]);
return { id: result.rows[0].id, action: 'inserted' };
}
}
async upsertSpecifications(client, watchId, specifications) {
if (!specifications) return;
const transformed = SpecificationTransformer.transform(specifications);
if (!transformed) return;
// Check if specifications exist
const existing = await client.query(
'SELECT id FROM omega.specifications WHERE watch_id = $1',
[watchId]
);
if (existing.rows.length > 0) {
// Update existing
await client.query(`
UPDATE omega.specifications
SET case_size = $2, case_diameter = $3, case_material = $4,
case_thickness = $5, case_thickness_mm = $6, lug_width = $7,
lug_width_mm = $8, movement_calibre = $9, movement_type = $10,
power_reserve = $11, power_reserve_hours = $12,
water_resistance = $13, water_resistance_meters = $14,
crystal_type = $15, dial_color = $16, bezel_type = $17,
weight = $18, weight_grams = $19, updated_at = NOW()
WHERE watch_id = $1
`, [
watchId, transformed.case_size, transformed.case_diameter,
transformed.case_material, transformed.case_thickness,
transformed.case_thickness_mm, transformed.lug_width,
transformed.lug_width_mm, transformed.movement_calibre,
transformed.movement_type, transformed.power_reserve,
transformed.power_reserve_hours, transformed.water_resistance,
transformed.water_resistance_meters, transformed.crystal_type,
transformed.dial_color, transformed.bezel_type, transformed.weight,
transformed.weight_grams
]);
} else {
// Insert new
await client.query(`
INSERT INTO omega.specifications (
watch_id, case_size, case_diameter, case_material,
case_thickness, case_thickness_mm, lug_width, lug_width_mm,
movement_calibre, movement_type, power_reserve,
power_reserve_hours, water_resistance, water_resistance_meters,
crystal_type, dial_color, bezel_type, weight, weight_grams
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)
`, [
watchId, transformed.case_size, transformed.case_diameter,
transformed.case_material, transformed.case_thickness,
transformed.case_thickness_mm, transformed.lug_width,
transformed.lug_width_mm, transformed.movement_calibre,
transformed.movement_type, transformed.power_reserve,
transformed.power_reserve_hours, transformed.water_resistance,
transformed.water_resistance_meters, transformed.crystal_type,
transformed.dial_color, transformed.bezel_type, transformed.weight,
transformed.weight_grams
]);
}
}
async upsertFeatures(client, watchId, features) {
if (!features || !Array.isArray(features)) return;
// Remove existing features
await client.query(
'DELETE FROM omega.watch_features WHERE watch_id = $1',
[watchId]
);
// Insert new features
for (let i = 0; i < features.length; i++) {
const featureId = await this.ensureFeature(client, features[i]);
await client.query(`
INSERT INTO omega.watch_features (watch_id, feature_id, display_order)
VALUES ($1, $2, $3)
ON CONFLICT DO NOTHING
`, [watchId, featureId, i]);
}
}
async upsertPriceHistory(client, watchId, priceHistory) {
if (!priceHistory || !Array.isArray(priceHistory)) return;
// Delete existing price history (idempotent reload)
await client.query(
'DELETE FROM omega.price_history WHERE watch_id = $1',
[watchId]
);
// Insert all price points
for (const pricePoint of priceHistory) {
const transformed = PriceHistoryTransformer.transform(pricePoint);
await client.query(`
INSERT INTO omega.price_history (
watch_id, year, price, condition, source,
currency, is_estimate, confidence_level
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
`, [
watchId, transformed.year, transformed.price,
transformed.condition, transformed.source, transformed.currency,
transformed.is_estimate, transformed.confidence_level
]);
}
}
async processWatch(client, watch, job) {
try {
job.stats.processed++;
// 1. Upsert watch
const { id: watchId, action } = await this.upsertWatch(client, watch);
if (action === 'inserted') {
job.stats.inserted++;
} else {
job.stats.updated++;
}
// 2. Upsert specifications
await this.upsertSpecifications(client, watchId, watch.specifications);
// 3. Upsert features
if (watch.specifications && watch.specifications.features) {
await this.upsertFeatures(client, watchId, watch.specifications.features);
}
// 4. Upsert price history
await this.upsertPriceHistory(client, watchId, watch.priceHistory);
if (job.stats.processed % 10 === 0) {
console.log(` Processed ${job.stats.processed} watches...`);
}
} catch (error) {
job.stats.failed++;
console.error(` ❌ Failed to process ${watch.model}:`, error.message);
}
}
async run(jsonPath) {
const client = await pool.connect();
const job = new ETLJob('JSON to PostgreSQL Migration', 'full_load');
try {
await job.start(client);
// Load source data
const data = await this.loadJSON(jsonPath);
// Begin transaction
await client.query('BEGIN');
console.log('\n📊 Processing watches...\n');
// Process each watch
for (const watch of data.watches) {
await this.processWatch(client, watch, job);
}
// Refresh materialized views
console.log('\n🔄 Refreshing materialized views...');
await client.query('SELECT omega.refresh_all_materialized_views()');
// Analyze tables
console.log('📈 Analyzing tables for query optimization...');
await client.query('ANALYZE omega.watches');
await client.query('ANALYZE omega.price_history');
await client.query('ANALYZE omega.specifications');
// Commit transaction
await client.query('COMMIT');
await job.complete(client);
// Print statistics
await this.printStatistics(client);
} catch (error) {
await client.query('ROLLBACK');
await job.complete(client, error);
throw error;
} finally {
client.release();
}
}
async printStatistics(client) {
console.log('\n📊 DATABASE STATISTICS\n');
// Watch counts by series
const series = await client.query(`
SELECT series, COUNT(*) as count
FROM omega.watches
WHERE is_active = true
GROUP BY series
ORDER BY count DESC
`);
console.log('Watches by Series:');
for (const row of series.rows) {
console.log(` ${row.series}: ${row.count}`);
}
// Price statistics
const prices = await client.query(`
SELECT
COUNT(DISTINCT watch_id) as watches_with_prices,
COUNT(*) as total_price_points,
MIN(year) as earliest_year,
MAX(year) as latest_year,
MIN(price) as min_price,
MAX(price) as max_price
FROM omega.price_history
`);
const ps = prices.rows[0];
console.log('\nPrice History:');
console.log(` Watches with prices: ${ps.watches_with_prices}`);
console.log(` Total price points: ${ps.total_price_points}`);
console.log(` Year range: ${ps.earliest_year} - ${ps.latest_year}`);
console.log(` Price range: $${ps.min_price} - $${parseInt(ps.max_price).toLocaleString()}`);
// Top performers
const performers = await client.query(`
SELECT model, series, appreciation_percentage, cagr_percentage
FROM omega.watch_statistics
WHERE appreciation_percentage IS NOT NULL
ORDER BY appreciation_percentage DESC
LIMIT 5
`);
console.log('\nTop 5 Performers:');
for (const row of performers.rows) {
console.log(` ${row.model}: ${row.appreciation_percentage}% (CAGR: ${row.cagr_percentage}%)`);
}
}
}
// ============================================================================
// MAIN EXECUTION
// ============================================================================
async function main() {
console.log(`
╔════════════════════════════════════════════════════════════════╗
║ OMEGA WATCHES ETL PIPELINE - PRODUCTION v2.0 ║
╚════════════════════════════════════════════════════════════════╝
`);
const jsonPath = path.join(__dirname, '..', 'data', 'watches.json');
const pipeline = new ETLPipeline();
try {
await pipeline.run(jsonPath);
console.log('\n✅ ETL PIPELINE COMPLETED SUCCESSFULLY\n');
process.exit(0);
} catch (error) {
console.error('\n❌ ETL PIPELINE FAILED:', error);
process.exit(1);
} finally {
await pool.end();
}
}
// Run if executed directly
if (import.meta.url === `file://${process.argv[1]}`) {
main();
}
export { ETLPipeline, WatchTransformer, SpecificationTransformer };