Transacciones, Cursores y Retry

import pg from "pg";

const { Pool } = pg;
const pool = new Pool({
  user: "postgres", host: "localhost",
  password: "postgres", database: "clientes", port: 5432,
});

pool.connect(async (error, client, release) => {
  await client.query("BEGIN");

  try {
    // Operación 1: descontar saldo
    await client.query(
      "UPDATE usuarios SET saldo = saldo - $1 WHERE email = $2",
      [200000, "joselyn@demo.com"]
    );

    // Operación 2: acreditar saldo
    await client.query(
      "UPDATE usuarios SET saldo = saldo + $1 WHERE email = $2",
      [200000, "pedro@demo.com"]
    );

    await client.query("COMMIT");
    console.log("Transferencia completada.");

  } catch (e) {
    await client.query("ROLLBACK");
    console.error("Transacción revertida.");
    // pg enriquece el error con metadatos:
    console.error("Código PG:", e.code);     // '23514' = check_violation, etc.
    console.error("Detalle:",  e.detail);    // qué constraint falló
    console.error("Tabla:",    e.table);     // tabla origen del error
    console.error("Constraint:", e.constraint);
  } finally {
    release(); // Siempre devolver al pool
    pool.end();
  }
});

// models/Cliente.js (fragmento — transferencia)
import db from "../config/db.js";

const pool = db.pool;

export const transferir = async (emailOrigen, emailDestino, monto) => {
  let client;
  try {
    client = await pool.connect();
    await client.query("BEGIN");

    // Verificar que el origen tenga saldo suficiente
    const { rows } = await client.query(
      "SELECT saldo FROM usuarios WHERE email = $1 FOR UPDATE",
      [emailOrigen]
    );

    if (!rows[0] || rows[0].saldo < monto) {
      const err = new Error("Saldo insuficiente.");
      err.statusCode = 422;
      throw err;
    }

    await client.query(
      "UPDATE usuarios SET saldo = saldo - $1 WHERE email = $2",
      [monto, emailOrigen]
    );

    await client.query(
      "UPDATE usuarios SET saldo = saldo + $1 WHERE email = $2",
      [monto, emailDestino]
    );

    await client.query("COMMIT");
    return { ok: true, mensaje: "Transferencia realizada." };

  } catch (error) {
    if (client) await client.query("ROLLBACK");
    throw error;
  } finally {
    if (client) client.release();
  }
};

npm install pg-cursor

// index.js — lectura de tabla grande con cursor
import pg from "pg";
import Cursor from "pg-cursor";

const { Pool } = pg;
const pool = new Pool({ /* config */ });

const leerConCursor = async () => {
  const client = await pool.connect();

  // Crear el cursor — NO ejecuta la query todavía
  const cursor = client.query(new Cursor("SELECT * FROM ventas ORDER BY id"));

  const BATCH_SIZE = 100; // leer 100 filas por vez
  let totalProcesadas = 0;

  try {
    while (true) {
      // read() devuelve un array de hasta BATCH_SIZE filas
      const filas = await cursor.read(BATCH_SIZE);

      if (filas.length === 0) break; // cursor agotado

      // Procesar el lote (ej: transformar, guardar en otro lado)
      totalProcesadas += filas.length;
      console.log(`Procesadas: ${totalProcesadas} filas`);
    }
    console.log(`Total: ${totalProcesadas} filas procesadas.`);
  } finally {
    await cursor.close(); // liberar recursos del cursor en el servidor
    client.release();
  }
};

leerConCursor();

// Errores de red/infraestructura que justifican reintento
const ERRORES_TRANSITORIOS = new Set([
  "ECONNRESET",   // Conexión cortada por el servidor
  "ECONNREFUSED", // PostgreSQL no arrancó aún
  "ETIMEDOUT",    // Timeout de red
  "57P03",        // cannot_connect_now (BD inicializando)
  "53300",        // too_many_connections
  "08006",        // connection_failure
]);

const esTransitorio = (error) =>
  ERRORES_TRANSITORIOS.has(error.code) || ERRORES_TRANSITORIOS.has(error.codigoPg);

/**
 * conReintento: envuelve cualquier función async con lógica de reintento.
 * @param fn          - La operación a ejecutar (debe ser async)
 * @param maxIntentos - Cuántas veces intentarlo (default: 3)
 * @param delayBaseMs - Tiempo base entre intentos en ms (default: 300)
 */
const conReintento = async (fn, maxIntentos = 3, delayBaseMs = 300) => {
  for (let intento = 1; intento <= maxIntentos; intento++) {
    try {
      return await fn();
    } catch (error) {
      // Errores permanentes o último intento → propagar inmediatamente
      if (!esTransitorio(error) || intento === maxIntentos) {
        throw error;
      }

      // Backoff exponencial: 300ms → 600ms → 1200ms
      const delay = delayBaseMs * Math.pow(2, intento - 1);
      console.warn(`[Retry] Intento ${intento}/${maxIntentos}. Reintentando en ${delay}ms...`);
      await new Promise((resolve) => setTimeout(resolve, delay));
    }
  }
};

// Envolver la operación con conReintento
export const crearCliente = async (datos) => {
  return conReintento(async () => {
    let client;
    try {
      client = await pool.connect();
      const { rows } = await client.query(
        "INSERT INTO clientes (nombre, email) VALUES ($1, $2) RETURNING *",
        [datos.nombre, datos.email]
      );
      return rows[0];
    } finally {
      if (client) client.release();
    }
  });
};

Transacción simple
  pool.connect() → BEGIN → query1 → query2 → COMMIT
                                   ↕ error → ROLLBACK
                   finally: client.release()

Cursor (tabla grande)
  pool.connect() → new Cursor(sql) → cursor.read(N) × bucle
                   finally: cursor.close() + client.release()

Retry (errores transitorios)
  conReintento(fn, 3, 300)
    └─ intento 1 → falla (ECONNRESET) → 300ms
    └─ intento 2 → falla (ECONNRESET) → 600ms
    └─ intento 3 → éxito ✅