Làm Phẳng Mô Hình EAV Magento 2 Bằng SQL Đa Tầng & Node.js Stream Pipeline

-- Kỹ thuật Store-Scope Fallback Resolution cho Thuộc tính Tên Sản phẩm
LEFT JOIN catalog_product_entity_varchar AS cpev_name_store
    ON cpev_name_store.entity_id = cpe.entity_id
    AND cpev_name_store.attribute_id = 73 -- attribute_id của 'name'
    AND cpev_name_store.store_id = 1      -- Storefront Tiếng Việt

LEFT JOIN catalog_product_entity_varchar AS cpev_name_default
    ON cpev_name_default.entity_id = cpe.entity_id
    AND cpev_name_default.attribute_id = 73
    AND cpev_name_default.store_id = 0     -- Default Scope

-- Lựa chọn giá trị ưu tiên:
COALESCE(cpev_name_store.value, cpev_name_default.value) AS product_name

SELECT
    cpe.entity_id                                                   AS product_id,
    cpe.sku                                                         AS sku,
    cpe.type_id                                                     AS product_type,
    cpe.created_at                                                  AS created_at,
    cpe.updated_at                                                  AS updated_at,

    -- 1. Tên sản phẩm (Varchar - Fallback Scope)
    COALESCE(cpev_name_store.value, cpev_name_default.value)         AS name,

    -- 2. Trạng thái kích hoạt (Int: 1 = Enabled, 2 = Disabled)
    COALESCE(cpei_status_store.value, cpei_status_default.value)     AS status,

    -- 3. Mức độ hiển thị (Int: 4 = Catalog/Search)
    COALESCE(cpei_vis_store.value, cpei_vis_default.value)           AS visibility,

    -- 4. Giá bán cơ bản (Decimal)
    COALESCE(cped_price_store.value, cped_price_default.value)       AS price,

    -- 5. Giá khuyến mãi đặc biệt (Decimal)
    COALESCE(cped_sprice_store.value, cped_sprice_default.value)     AS special_price,

    -- 6. Khối lượng sản phẩm (Decimal)
    COALESCE(cped_weight_store.value, cped_weight_default.value)     AS weight,

    -- 7. Đường dẫn hình ảnh đại diện (Varchar)
    COALESCE(cpev_img_store.value, cpev_img_default.value)           AS base_image,

    -- 8. Tồn kho thực tế từ bảng MSI hoặc legacy stock
    COALESCE(inv.quantity, 0)                                       AS stock_quantity,
    COALESCE(inv.is_in_stock, 0)                                    AS is_in_stock,

    -- 9. Gom nhóm danh mục thành chuỗi ID cách nhau dấu phẩy
    GROUP_CONCAT(DISTINCT ccp.category_id ORDER BY ccp.category_id ASC) AS category_ids

FROM catalog_product_entity AS cpe

-- Join Tên sản phẩm
LEFT JOIN eav_attribute AS ea_name ON ea_name.entity_type_id = 4 AND ea_name.attribute_code = 'name'
LEFT JOIN catalog_product_entity_varchar AS cpev_name_store 
    ON cpev_name_store.entity_id = cpe.entity_id AND cpev_name_store.attribute_id = ea_name.attribute_id AND cpev_name_store.store_id = 1
LEFT JOIN catalog_product_entity_varchar AS cpev_name_default 
    ON cpev_name_default.entity_id = cpe.entity_id AND cpev_name_default.attribute_id = ea_name.attribute_id AND cpev_name_default.store_id = 0

-- Join Trạng thái sản phẩm
LEFT JOIN eav_attribute AS ea_status ON ea_status.entity_type_id = 4 AND ea_status.attribute_code = 'status'
LEFT JOIN catalog_product_entity_int AS cpei_status_store 
    ON cpei_status_store.entity_id = cpe.entity_id AND cpei_status_store.attribute_id = ea_status.attribute_id AND cpei_status_store.store_id = 1
LEFT JOIN catalog_product_entity_int AS cpei_status_default 
    ON cpei_status_default.entity_id = cpe.entity_id AND cpei_status_default.attribute_id = ea_status.attribute_id AND cpei_status_default.store_id = 0

-- Join Giá sản phẩm
LEFT JOIN eav_attribute AS ea_price ON ea_price.entity_type_id = 4 AND ea_price.attribute_code = 'price'
LEFT JOIN catalog_product_entity_decimal AS cped_price_store 
    ON cped_price_store.entity_id = cpe.entity_id AND cped_price_store.attribute_id = ea_price.attribute_id AND cped_price_store.store_id = 1
LEFT JOIN catalog_product_entity_decimal AS cped_price_default 
    ON cped_price_default.entity_id = cpe.entity_id AND cped_price_default.attribute_id = ea_price.attribute_id AND cped_price_default.store_id = 0

-- Join Ảnh sản phẩm
LEFT JOIN eav_attribute AS ea_img ON ea_img.entity_type_id = 4 AND ea_img.attribute_code = 'image'
LEFT JOIN catalog_product_entity_varchar AS cpev_img_store 
    ON cpev_img_store.entity_id = cpe.entity_id AND cpev_img_store.attribute_id = ea_img.attribute_id AND cpev_img_store.store_id = 1
LEFT JOIN catalog_product_entity_varchar AS cpev_img_default 
    ON cpev_img_default.entity_id = cpe.entity_id AND cpev_img_default.attribute_id = ea_img.attribute_id AND cpev_img_default.store_id = 0

-- Join Tồn kho (Legacy stock table)
LEFT JOIN cataloginventory_stock_item AS inv 
    ON inv.product_id = cpe.entity_id AND inv.stock_id = 1

-- Join Danh mục
LEFT JOIN catalog_category_product AS ccp 
    ON ccp.product_id = cpe.entity_id

GROUP BY cpe.entity_id;

-- ANTI-PATTERN: Sập database khi OFFSET lớn
SELECT * FROM catalog_product_entity LIMIT 1000 OFFSET 500000;

/**
 * Magento 2 EAV Flattening ETL Pipeline
 * Chạy trích xuất hàng triệu sản phẩm với mức tiêu thụ RAM ổn định < 150MB.
 */

const mysql = require('mysql2');
const { Transform, pipeline } = require('stream');
const { Client } = require('pg');
const fs = require('fs');

// Cấu hình kết nối
const mysqlConfig = {
  host: process.env.MAGENTO_DB_HOST || '127.0.0.1',
  user: process.env.MAGENTO_DB_USER || 'root',
  password: process.env.MAGENTO_DB_PASSWORD || 'secret',
  database: process.env.MAGENTO_DB_NAME || 'magento2',
  rowsAsArray: false,
};

const pgClient = new Client({
  connectionString: process.env.TARGET_PG_URL || 'postgresql://postgres:secret@127.0.0.1:5432/ecommerce_new',
});

// File ghi nhận các bản ghi lỗi (Dead-Letter Queue)
const dlqStream = fs.createWriteStream('./failed_products_dlq.jsonl', { flags: 'a' });

async function runETLPipeline() {
  await pgClient.connect();
  console.log('[ETL] Đã kết nối thành công tới Target PostgreSQL Database');

  const connection = mysql.createConnection(mysqlConfig);

  const exportSQL = `
    SELECT 
      cpe.entity_id AS product_id,
      cpe.sku,
      cpe.type_id,
      COALESCE(cpev_name.value, 'Chưa đặt tên') AS name,
      COALESCE(cped_price.value, 0.0000) AS price,
      COALESCE(inv.qty, 0) AS stock_qty
    FROM catalog_product_entity cpe
    LEFT JOIN eav_attribute ea_name ON ea_name.entity_type_id = 4 AND ea_name.attribute_code = 'name'
    LEFT JOIN catalog_product_entity_varchar cpev_name ON cpev_name.entity_id = cpe.entity_id AND cpev_name.attribute_id = ea_name.attribute_id AND cpev_name.store_id = 0
    LEFT JOIN eav_attribute ea_price ON ea_price.entity_type_id = 4 AND ea_price.attribute_code = 'price'
    LEFT JOIN catalog_product_entity_decimal cped_price ON cped_price.entity_id = cpe.entity_id AND cped_price.attribute_id = ea_price.attribute_id AND cped_price.store_id = 0
    LEFT JOIN cataloginventory_stock_item inv ON inv.product_id = cpe.entity_id;
  `;

  // Mở streaming query từ MySQL (Không nạp toàn bộ vào RAM)
  const queryStream = connection.query(exportSQL).stream({ highWaterMark: 250 });

  let totalProcessed = 0;
  let batch = [];
  const BATCH_SIZE = 500;

  // Stream chuyển đổi và làm sạch dữ liệu
  const transformStream = new Transform({
    objectMode: true,
    highWaterMark: 250,
    async transform(row, encoding, callback) {
      try {
        // Kiểm tra tính hợp lệ cơ bản
        if (!row.sku || row.sku.trim() === '') {
          throw new Error('Sản phẩm thiếu SKU');
        }

        const cleanProduct = {
          id: row.product_id,
          sku: row.sku.trim(),
          type: row.type_id,
          name: row.name,
          price: parseFloat(row.price),
          stock: parseInt(row.stock_qty, 10),
          migrated_at: new Date().toISOString(),
        };

        batch.push(cleanProduct);

        // Khi đạt kích thước mẻ, xả xuống PostgreSQL
        if (batch.length >= BATCH_SIZE) {
          const currentBatch = [...batch];
          batch = [];
          await flushBatchToPostgres(currentBatch);
          totalProcessed += currentBatch.length;
          console.log(`[ETL] Đã xử lý và nạp thành công ${totalProcessed} sản phẩm...`);
        }

        callback();
      } catch (err) {
        // Ghi bản ghi lỗi vào DLQ mà không ngắt pipeline
        dlqStream.write(JSON.stringify({ row, error: err.message }) + '\n');
        callback();
      }
    },
    async flush(callback) {
      // Xả nốt các bản ghi còn sót lại trong batch cuối
      if (batch.length > 0) {
        await flushBatchToPostgres(batch);
        totalProcessed += batch.length;
      }
      callback();
    },
  });

  // Điều phối dòng chảy với quản lý áp lực ngược tự động
  pipeline(queryStream, transformStream, (err) => {
    if (err) {
      console.error('[ETL THẤT BẠI] Pipeline gặp sự cố nghiêm trọng:', err);
    } else {
      console.log(`\n🎉 [ETL HOÀN TẤT] Tổng cộng ${totalProcessed} sản phẩm đã được làm phẳng thành công!`);
    }
    connection.end();
    pgClient.end();
    dlqStream.end();
  });
}

// Hàm ghi mẻ vào PostgreSQL với tính năng Idempotent Upsert
async function flushBatchToPostgres(products) {
  const query = `
    INSERT INTO target_products (id, sku, type, name, price, stock, migrated_at)
    VALUES ${products.map((_, i) => `($${i*7+1}, $${i*7+2}, $${i*7+3}, $${i*7+4}, $${i*7+5}, $${i*7+6}, $${i*7+7})`).join(',')}
    ON CONFLICT (sku) DO UPDATE SET
      name = EXCLUDED.name,
      price = EXCLUDED.price,
      stock = EXCLUDED.stock,
      migrated_at = EXCLUDED.migrated_at;
  `;

  const values = products.flatMap(p => [p.id, p.sku, p.type, p.name, p.price, p.stock, p.migrated_at]);
  await pgClient.query(query, values);
}

runETLPipeline().catch(console.error);