ETL (Extract, Transform, Load)

// Simpelt ETL eksempel med Node.js

// ============ EXTRACT ============
async function extractFromSources() {
  // Udtræk fra MySQL database
  const dbCustomers = await mysqlClient.query(`
    SELECT customer_id, name, email, created_at
    FROM customers
    WHERE updated_at > '2024-01-01'
  `);
  
  // Udtræk fra REST API
  const apiOrders = await fetch('https://api.shop.dk/orders')
    .then(res => res.json());
  
  // Udtræk fra CSV fil
  const csvProducts = await fs.readFile('products.csv', 'utf-8')
    .then(data => parseCSV(data));
  
  return { customers: dbCustomers, orders: apiOrders, products: csvProducts };
}

// ============ TRANSFORM ============
function transformData({ customers, orders, products }) {
  // 1. Data Cleaning
  const cleanCustomers = customers
    .filter(c => c.email && c.email.includes('@')) // Remove invalid emails
    .map(c => ({
      ...c,
      email: c.email.toLowerCase().trim(), // Normalize
      name: c.name.trim()
    }));
  
  // 2. Data Enrichment - Join orders with customers
  const enrichedOrders = orders.map(order => {
    const customer = cleanCustomers.find(c => c.customer_id === order.customer_id);
    return {
      order_id: order.id,
      customer_name: customer?.name || 'Unknown',
      customer_email: customer?.email,
      order_total: order.total,
      order_date: new Date(order.created_at)
    };
  });
  
  // 3. Data Aggregation
  const customerMetrics = cleanCustomers.map(customer => {
    const customerOrders = enrichedOrders.filter(
      o => o.customer_email === customer.email
    );
    return {
      customer_id: customer.customer_id,
      customer_email: customer.email,
      total_orders: customerOrders.length,
      total_spent: customerOrders.reduce((sum, o) => sum + o.order_total, 0),
      avg_order_value: customerOrders.length > 0 
        ? customerOrders.reduce((sum, o) => sum + o.order_total, 0) / customerOrders.length
        : 0,
      first_order_date: customerOrders.length > 0
        ? new Date(Math.min(...customerOrders.map(o => o.order_date)))
        : null
    };
  });
  
  return { customerMetrics, enrichedOrders, products };
}

// ============ LOAD ============
async function loadToWarehouse(transformedData) {
  const { customerMetrics, enrichedOrders, products } = transformedData;
  
  // Load til data warehouse (PostgreSQL)
  await warehouseDb.transaction(async (trx) => {
    // Truncate staging tables
    await trx.raw('TRUNCATE staging.customer_metrics');
    
    // Bulk insert
    await trx('staging.customer_metrics').insert(customerMetrics);
    await trx('staging.enriched_orders').insert(enrichedOrders);
    
    // Merge til production tables (upsert)
    await trx.raw(`
      INSERT INTO prod.customers 
      SELECT * FROM staging.customer_metrics
      ON CONFLICT (customer_id) 
      DO UPDATE SET
        total_orders = EXCLUDED.total_orders,
        total_spent = EXCLUDED.total_spent,
        updated_at = NOW()
    `);
  });
  
  console.log(`ETL completed: Loaded ${customerMetrics.length} customers`);
}

// ============ ORCHESTRATION ============
async function runETLPipeline() {
  try {
    console.log('Starting ETL pipeline...');
    const rawData = await extractFromSources();
    const transformedData = transformData(rawData);
    await loadToWarehouse(transformedData);
    console.log('ETL pipeline completed successfully');
  } catch (error) {
    console.error('ETL failed:', error);
    // Send alert, log to monitoring
  }
}

// Schedule: Run every day at 2 AM
cron.schedule('0 2 * * *', runETLPipeline);