Apiflow-gateway/sync/utils.js

'use strict';

/**
 * Upsert en masse dans PG destination (chunks de 500 lignes).
 * Les colonnes sont déduites dynamiquement des lignes source.
 */
async function batchUpsert(pg, table, rows, pk) {
  if (!rows.length) return 0;
  const cols  = Object.keys(rows[0]);
  const CHUNK = 500;
  let total   = 0;

  for (let i = 0; i < rows.length; i += CHUNK) {
    const chunk  = rows.slice(i, i + CHUNK);
    const values = [];
    const placeholders = chunk.map((row, ri) => {
      const ph = cols.map((col, ci) => {
        values.push(row[col] ?? null);
        return `$${ri * cols.length + ci + 1}`;
      });
      return `(${ph.join(', ')})`;
    });
    const updateCols = cols.filter(c => !pk.includes(c));
    const updateSet  = updateCols.length
      ? updateCols.map(c => `${c} = EXCLUDED.${c}`).join(', ')
      : `${pk[0]} = EXCLUDED.${pk[0]}`; // fallback si toutes les cols sont PK

    const sql = `
      INSERT INTO ${table} (${cols.join(', ')})
      VALUES ${placeholders.join(', ')}
      ON CONFLICT (${pk.join(', ')}) DO UPDATE SET ${updateSet}
    `;
    await pg.query(sql, values);
    total += chunk.length;
  }
  return total;
}

/**
 * Full refresh : TRUNCATE puis INSERT en masse.
 */
async function fullRefresh(pg, table, rows) {
  await pg.query(`TRUNCATE TABLE ${table} CASCADE`);
  if (!rows.length) return 0;

  const cols  = Object.keys(rows[0]);
  const CHUNK = 500;
  let total   = 0;

  for (let i = 0; i < rows.length; i += CHUNK) {
    const chunk  = rows.slice(i, i + CHUNK);
    const values = [];
    const placeholders = chunk.map((row, ri) => {
      const ph = cols.map((col, ci) => {
        values.push(row[col] ?? null);
        return `$${ri * cols.length + ci + 1}`;
      });
      return `(${ph.join(', ')})`;
    });
    await pg.query(
      `INSERT INTO ${table} (${cols.join(', ')}) VALUES ${placeholders.join(', ')}`,
      values
    );
    total += chunk.length;
  }
  return total;
}

async function logSync(pg, tableName, rowsSynced, status, errorMsg = null) {
  await pg.query(`
    INSERT INTO sync_log (table_name, last_sync, rows_synced, status, error_msg)
    VALUES ($1, NOW(), $2, $3, $4)
    ON CONFLICT (table_name) DO UPDATE SET
      last_sync   = NOW(),
      rows_synced = $2,
      status      = $3,
      error_msg   = $4
  `, [tableName, rowsSynced, status, errorMsg]);
}

async function getLastSync(pg, tableName) {
  const res = await pg.query(
    `SELECT last_sync FROM sync_log WHERE table_name = $1 AND status = 'ok'`,
    [tableName]
  );
  return res.rows[0]?.last_sync || null;
}

module.exports = { batchUpsert, fullRefresh, logSync, getLastSync };