npm install express pg amqplib pino pino-http zod @techcorp/comun-http && npm install -D nodemon dotenv

servicio-pedidos/
├── src/
│   ├── servidor.js, app.js, config.js (04-03), salud.js
│   ├── rutas/pedidos.js                    # POST /v1/pedidos, GET /v1/pedidos/:id
│   ├── casos-uso/crearPedido.js
│   ├── dominio/pedido.js                   # agregado: crear, calcular total, aplicar evento
│   ├── dominio/maquinaEstadosPedido.js     # TRANSICIONES, MOTIVOS, transicionar (02-05, tal cual)
│   ├── repositorios/pedidoRepositorio.js   # guardarConEventos, obtener, claves de idempotencia, procesarUnaVez
│   ├── clientes/catalogoCliente.js, clientes/clientesCliente.js
│   ├── traductores/traductorProducto.js    # ACL (02-03, 03-06)
│   ├── infra/postgres.js
│   └── mensajeria/relayOutbox.js, consumidorSaga.js, consumidorClientes.js
├── migraciones/001-esquema-inicial.sql … 004-claves-idempotencia.sql
├── scripts/migrar.js, scripts/publicarEvento.js
└── contratos/openapi.yaml, contratos/asyncapi.yaml

// src/infra/postgres.js
const { Pool } = require('pg');

function crearPoolPostgres({ url, logger }) {
  const pool = new Pool({ connectionString: url, max: 10, connectionTimeoutMillis: 5000, idleTimeoutMillis: 30000 });
  pool.on('error', (err) => logger.error({ err }, 'error en conexión inactiva del pool'));   // sin esto, un error tumba el proceso
  return {
    consultar: (sql, params) => pool.query(sql, params),                 // fuera de transacción
    // Ejecuta fn(tx) dentro de BEGIN/COMMIT; ROLLBACK si fn lanza. tx.consultar usa SIEMPRE la misma conexión.
    async transaccion(fn) {
      const cliente = await pool.connect();
      try {
        await cliente.query('BEGIN');
        const resultado = await fn({ consultar: (sql, params) => cliente.query(sql, params) });
        await cliente.query('COMMIT');
        return resultado;
      } catch (err) { await cliente.query('ROLLBACK'); throw err; }
      finally { cliente.release(); }
    },
    ping: () => pool.query('SELECT 1'),
    cerrar: () => pool.end()
  };
}
module.exports = { crearPoolPostgres };

// src/traductores/traductorProducto.js — el lector tolerante de 03-06, sin cambios
function aProductoDePedidos(dto) {
  return { productoId: dto.id, nombre: dto.nombre, precioUnitario: Number(dto.precio), disponible: dto.disponible !== false };
}
module.exports = { aProductoDePedidos };

// src/clientes/catalogoCliente.js
const { ErrorNegocio } = require('@techcorp/comun-http');
const { aProductoDePedidos } = require('../traductores/traductorProducto');

function crearCatalogoCliente({ urlBase, timeoutMs }) {
  return {
    // Devuelve un Map productoId → { productoId, nombre, precioUnitario, disponible }
    async obtenerProductos(ids, { requestId }) {
      const url = `${urlBase}/v1/productos?ids=${encodeURIComponent(ids.join(','))}`;
      let respuesta;
      try {
        respuesta = await fetch(url, { headers: { Accept: 'application/json', 'X-Request-Id': requestId }, signal: AbortSignal.timeout(timeoutMs) });
      } catch (err) {                                     // timeout o red: 503 para nuestro cliente (03-01)
        throw new ErrorNegocio('DEPENDENCIA_NO_DISPONIBLE', `Catálogo no disponible: ${err.name}`, 503);
      }
      if (!respuesta.ok) {
        const problema = await respuesta.json().catch(() => ({}));
        if (respuesta.status >= 500) throw new ErrorNegocio('DEPENDENCIA_NO_DISPONIBLE', `Catálogo respondió ${respuesta.status}`, 503);
        throw new ErrorNegocio('PETICION_INVALIDA', `Catálogo rechazó la petición (${problema.codigo ?? respuesta.status})`, 400);
      }
      const { datos, noEncontrados } = await respuesta.json();
      const productos = datos.map(aProductoDePedidos);
      const noVendibles = [...noEncontrados, ...productos.filter((p) => !p.disponible).map((p) => p.productoId)];
      if (noVendibles.length > 0) throw new ErrorNegocio('PRODUCTO_NO_DISPONIBLE', `Productos no disponibles: ${noVendibles.join(', ')}`, 422);
      return new Map(productos.map((p) => [p.productoId, p]));
    }
  };
}
module.exports = { crearCatalogoCliente };

// src/dominio/pedido.js
const { randomUUID } = require('node:crypto');
const { transicionar, MOTIVOS } = require('./maquinaEstadosPedido');

const aCentimos = (n) => Math.round(n * 100);

// Construye un Pedido PENDIENTE a partir de la petición validada y de lo que dijeron Clientes y Catálogo
function crearPedido({ clienteId, lineas, direccionEnvio }, { cliente, productos }) {
  const lineasCongeladas = lineas.map((l, i) => {
    const p = productos.get(l.productoId);            // catalogoCliente ya garantizó que existe y es vendible
    return { linea: i + 1, productoId: p.productoId, nombreProducto: p.nombre, precioUnitario: p.precioUnitario, cantidad: l.cantidad };
  });
  const totalCentimos = lineasCongeladas.reduce((acc, l) => acc + aCentimos(l.precioUnitario) * l.cantidad, 0);
  return {
    pedidoId: `ped-${randomUUID().slice(0, 8)}`,       // id opaco generado por el dueño (02-04)
    clienteId, cliente: { nombre: cliente.nombre, email: cliente.email },
    estado: 'PENDIENTE', lineas: lineasCongeladas, direccionEnvio,
    total: totalCentimos / 100, motivoCancelacion: null, creadoEn: new Date().toISOString()
  };
}

// Aplica un evento de la saga; devuelve el nuevo estado o null si no hay transición (evento tardío/duplicado)
function aplicarEventoSaga(pedido, tipoEvento) {
  const nuevoEstado = transicionar(pedido.estado, tipoEvento);
  if (!nuevoEstado) return null;
  pedido.estado = nuevoEstado;
  if (nuevoEstado === 'CANCELADO') pedido.motivoCancelacion = MOTIVOS[tipoEvento];
  return nuevoEstado;
}

// Carga que viaja en pedido.creado / pedido.confirmado / pedido.cancelado (contrato de 02-05 y AsyncAPI de 03-06)
function datosParaConsumidores(p) {
  return { pedidoId: p.pedidoId, clienteId: p.clienteId, cliente: p.cliente, direccionEnvio: p.direccionEnvio, total: p.total,
           lineas: p.lineas.map((l) => ({ productoId: l.productoId, nombre: l.nombreProducto, cantidad: l.cantidad, precioUnitario: l.precioUnitario })) };
}
module.exports = { crearPedido, aplicarEventoSaga, datosParaConsumidores };

// src/repositorios/pedidoRepositorio.js
const { randomUUID } = require('node:crypto');

function crearPedidoRepositorio(bd) {
  // Guarda pedido + líneas + réplica del cliente + eventos en el outbox, en UNA transacción (02-05 §7).
  // `tx` opcional: si el llamante ya está en una transacción (procesarUnaVez), reutilizamos la suya.
  async function guardarConEventos(pedido, eventos, tx) {
    const trabajo = async (t) => {
      await t.consultar(`INSERT INTO pedidos (pedido_id, cliente_id, estado, total, direccion_envio, motivo_cancelacion, creado_en, actualizado_en)
                         VALUES ($1,$2,$3,$4,$5,$6,$7,NOW())
                         ON CONFLICT (pedido_id) DO UPDATE SET estado = EXCLUDED.estado, motivo_cancelacion = EXCLUDED.motivo_cancelacion, actualizado_en = NOW()`,
        [pedido.pedidoId, pedido.clienteId, pedido.estado, pedido.total, pedido.direccionEnvio, pedido.motivoCancelacion, pedido.creadoEn]);
      for (const l of pedido.lineas) {                      // las líneas nunca cambian tras la creación: insert idempotente
        await t.consultar(`INSERT INTO lineas_pedido (pedido_id, linea, producto_id, nombre_producto, precio_unitario, cantidad)
                           VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT DO NOTHING`, [pedido.pedidoId, l.linea, l.productoId, l.nombreProducto, l.precioUnitario, l.cantidad]);
      }
      if (pedido.cliente) {                                 // réplica clientes_ref (02-04): la calentamos con lo que ya sabemos
        await t.consultar(`INSERT INTO clientes_ref (cliente_id, nombre, email, actualizado_en) VALUES ($1,$2,$3,NOW())
                           ON CONFLICT (cliente_id) DO NOTHING`, [pedido.clienteId, pedido.cliente.nombre, pedido.cliente.email]);
      }
      for (const ev of eventos) {                           // el outbox: mismos INSERT, misma transacción
        await t.consultar(`INSERT INTO outbox (evento_id, agregado_tipo, agregado_id, tipo, version, carga) VALUES ($1,'Pedido',$2,$3,$4,$5)`,
          [`evt-${randomUUID()}`, pedido.pedidoId, ev.tipo, ev.version ?? 1, ev.carga]);
      }
    };
    return tx ? trabajo(tx) : bd.transaccion(trabajo);
  }

  async function obtener(pedidoId, tx = bd) {
    const { rows } = await tx.consultar(`SELECT p.*, c.nombre AS cliente_nombre, c.email AS cliente_email
                                          FROM pedidos p LEFT JOIN clientes_ref c ON c.cliente_id = p.cliente_id WHERE p.pedido_id = $1`, [pedidoId]);
    if (rows.length === 0) return null;
    const lineas = (await tx.consultar('SELECT * FROM lineas_pedido WHERE pedido_id = $1 ORDER BY linea', [pedidoId])).rows;
    return aPedido(rows[0], lineas);                       // fila → agregado (nombres camelCase; ~10 líneas, omitidas)
  }

  // Idempotencia de la API (02-05 §8c, 03-01): clave → respuesta guardada
  const buscarClave = async (clave) => (await bd.consultar('SELECT huella, respuesta FROM claves_idempotencia WHERE clave = $1', [clave])).rows[0] ?? null;
  const guardarClave = (t, clave, pedidoId, huella, respuesta) =>
    t.consultar('INSERT INTO claves_idempotencia (clave, pedido_id, huella, respuesta) VALUES ($1,$2,$3,$4)', [clave, pedidoId, huella, respuesta]);

  // Idempotencia de consumidores (02-05 §8b, 03-02): efecto + registro en la MISMA transacción
  async function procesarUnaVez(eventoId, consumidor, fn) {
    return bd.transaccion(async (tx) => {
      const { rowCount } = await tx.consultar('INSERT INTO eventos_procesados (evento_id, consumidor) VALUES ($1,$2) ON CONFLICT DO NOTHING', [eventoId, consumidor]);
      if (rowCount === 0) return 'DUPLICADO';              // ya procesado: la transacción no hace nada más
      await fn(tx);                                        // el efecto real, con el mismo tx
      return 'PROCESADO';
    });
  }

  return { guardarConEventos, obtener, buscarClave, guardarClave, procesarUnaVez, transaccion: bd.transaccion };
}
module.exports = { crearPedidoRepositorio };

// src/casos-uso/crearPedido.js
const { createHash } = require('node:crypto');
const { ErrorNegocio } = require('@techcorp/comun-http');
const Pedido = require('../dominio/pedido');

const huellaDe = (cuerpo) => createHash('sha256').update(JSON.stringify(cuerpo)).digest('hex');

function crearCasoUsoCrearPedido({ repositorio, catalogoCliente, clientesCliente, logger }) {
  return async function crearPedido(peticion, { claveIdempotencia, requestId }) {
    // 1. Idempotencia: misma clave + mismo cuerpo → misma respuesta; misma clave + otro cuerpo → 422 (03-01)
    const huella = huellaDe(peticion);
    const previa = await repositorio.buscarClave(claveIdempotencia);
    if (previa) {
      if (previa.huella !== huella) throw new ErrorNegocio('CLAVE_IDEMPOTENCIA_REUTILIZADA', 'La Idempotency-Key ya se usó con otro cuerpo', 422);
      return { pedido: previa.respuesta, repetida: true };
    }
    // 2. Cliente y productos EN PARALELO: son independientes y así la latencia es la del más lento, no la suma
    const [cliente, productos] = await Promise.all([
      clientesCliente.obtenerCliente(peticion.clienteId, { requestId }),                            // CLIENTE_NO_EXISTE → 404
      catalogoCliente.obtenerProductos([...new Set(peticion.lineas.map((l) => l.productoId))], { requestId })   // PRODUCTO_NO_DISPONIBLE → 422
    ]);
    // 3. El agregado, en PENDIENTE, con nombres y precios congelados y el total calculado
    const pedido = Pedido.crearPedido(peticion, { cliente, productos });
    const respuesta = aRepresentacion(pedido);           // el JSON de 03-01 (id, estado, lineas, total, _links); ~8 líneas, omitidas
    // 4. Pedido + pedido.creado + clave de idempotencia: UNA transacción. Ni RabbitMQ ni HTTP aquí dentro.
    await repositorio.transaccion(async (tx) => {
      await repositorio.guardarConEventos(pedido, [{ tipo: 'pedido.creado', version: 1, carga: Pedido.datosParaConsumidores(pedido) }], tx);
      await repositorio.guardarClave(tx, claveIdempotencia, pedido.pedidoId, huella, respuesta);
    });
    logger.info({ pedidoId: pedido.pedidoId, clienteId: pedido.clienteId, total: pedido.total, requestId }, 'pedido creado');
    return { pedido: respuesta, repetida: false };
  };
}
module.exports = { crearCasoUsoCrearPedido };

// src/rutas/pedidos.js (fragmento POST)
enrutador.post('/v1/pedidos', async (req, res, next) => {
  try {
    const claveIdempotencia = req.get('Idempotency-Key');
    if (!claveIdempotencia) throw new ErrorNegocio('PETICION_INVALIDA', 'Falta la cabecera Idempotency-Key', 400);
    const peticion = esquemaNuevoPedido.parse(req.body);   // zod: clienteId, lineas[{productoId, cantidad ≥ 1}], direccionEnvio (400/422 vía middleware)
    const { pedido } = await crearPedido(peticion, { claveIdempotencia, requestId: req.id });
    res.status(202).location(`/v1/pedidos/${pedido.id}`).json(pedido);
  } catch (err) { next(err); }
});

// src/mensajeria/relayOutbox.js
const { construirSobre, publicarEvento } = require('@techcorp/comun-http/mensajeria/publicador');   // 03-02

function crearRelayOutbox({ bd, canalConfirm, intervaloMs, lote = 50, logger }) {
  let temporizador = null, parado = false;

  async function publicarPendientes() {
    // Una transacción por lote. FOR UPDATE SKIP LOCKED: si hay dos réplicas de Pedidos, cada una toma filas distintas.
    const publicados = await bd.transaccion(async (tx) => {
      const { rows } = await tx.consultar(
        `SELECT evento_id, tipo, version, carga FROM outbox WHERE publicado_en IS NULL ORDER BY creado_en LIMIT $1 FOR UPDATE SKIP LOCKED`, [lote]);
      if (rows.length === 0) return 0;
      for (const fila of rows) {
        // El sobre lleva el eventoId del outbox (no uno nuevo): si el relay reintenta, el consumidor lo reconoce como duplicado
        const sobre = { ...construirSobre(fila.tipo, fila.carga, { version: fila.version }), eventoId: fila.evento_id };
        publicarEvento(canalConfirm, sobre);
      }
      await canalConfirm.waitForConfirms();                // el broker confirma que ha recibido (y persistido) el lote
      await tx.consultar(`UPDATE outbox SET publicado_en = NOW() WHERE evento_id = ANY($1)`, [rows.map((r) => r.evento_id)]);
      return rows.length;                                  // COMMIT: solo ahora quedan marcadas
    });
    if (publicados > 0) logger.debug({ publicados }, 'outbox publicado');
    return publicados;
  }

  async function ciclo() {
    if (parado) return;
    try {
      const n = await publicarPendientes();
      temporizador = setTimeout(ciclo, n === lote ? 0 : intervaloMs);   // si el lote venía lleno, seguir sin esperar
    } catch (err) {
      logger.error({ err }, 'relay outbox: fallo, reintento en el siguiente ciclo');   // RabbitMQ caído: las filas siguen pendientes
      temporizador = setTimeout(ciclo, intervaloMs);
    }
  }
  return { iniciar: () => ciclo(), parar: () => { parado = true; clearTimeout(temporizador); } };
}
module.exports = { crearRelayOutbox };

// src/mensajeria/consumidorSaga.js
const { declararColaConsumidor } = require('@techcorp/comun-http/mensajeria/topologia');   // 03-02
const Pedido = require('../dominio/pedido');

const COLA = 'pedidos.saga';
const EVENTOS = ['stock.reservado', 'stock.rechazado', 'pago.confirmado', 'pago.rechazado'];

function crearConsumidorSaga({ canal, repositorio, logger }) {
  // Efecto de un evento de la saga. Se ejecuta DENTRO de procesarUnaVez, con su tx.
  async function manejar(sobre, tx) {
    const pedido = await repositorio.obtener(sobre.carga.pedidoId, tx);
    if (!pedido) { logger.warn({ sobre }, 'evento para pedido desconocido'); return; }   // no reintentar: a ack (queda registrado como procesado)

    let nuevoEstado = Pedido.aplicarEventoSaga(pedido, sobre.tipo);
    if (!nuevoEstado) { logger.info({ pedidoId: pedido.pedidoId, de: pedido.estado, evento: sobre.tipo }, 'transicion_ignorada'); return; }
    if (nuevoEstado === 'PAGADO') nuevoEstado = Pedido.aplicarEventoSaga(pedido, 'confirmar');   // T4 de 02-05: hoy es inmediata

    const eventos = [];
    if (nuevoEstado === 'CONFIRMADO') eventos.push({ tipo: 'pedido.confirmado', carga: Pedido.datosParaConsumidores(pedido) });
    if (nuevoEstado === 'CANCELADO')  eventos.push({ tipo: 'pedido.cancelado',  carga: { ...Pedido.datosParaConsumidores(pedido), motivo: pedido.motivoCancelacion } });
    await repositorio.guardarConEventos(pedido, eventos, tx);        // estado nuevo + eventos de salida, misma transacción que eventos_procesados
    logger.info({ pedidoId: pedido.pedidoId, estado: nuevoEstado, evento: sobre.tipo }, 'pedido actualizado');
  }

  async function iniciar() {
    await declararColaConsumidor(canal, COLA, EVENTOS);
    await canal.prefetch(10);
    await canal.consume(COLA, async (msg) => {
      if (!msg) return;
      let sobre;
      try { sobre = JSON.parse(msg.content.toString()); } catch { return canal.nack(msg, false, false); }   // ilegible → DLQ
      try {
        await repositorio.procesarUnaVez(sobre.eventoId, 'pedidos.saga', (tx) => manejar(sobre, tx));
        canal.ack(msg);
      } catch (err) {
        logger.error({ err, eventoId: sobre.eventoId }, 'error procesando evento de saga');
        canal.nack(msg, false, !msg.fields.redelivered);              // 1.ª vez: reencolar; 2.ª: DLQ (03-02)
      }
    });
  }
  return { iniciar };
}
module.exports = { crearConsumidorSaga };

// src/rutas/pedidos.js (fragmento GET)
enrutador.get('/v1/pedidos/:id', async (req, res, next) => {
  try {
    const pedido = await repositorio.obtener(req.params.id);
    if (!pedido) throw new ErrorNegocio('PEDIDO_NO_EXISTE', `No existe ${req.params.id}`, 404);
    const etag = `"${pedido.pedidoId}:${new Date(pedido.actualizadoEn).getTime()}"`;   // cambia con cada transición
    if (req.get('If-None-Match') === etag) return res.status(304).end();               // polling barato (03-01)
    res.set('ETag', etag).set('Cache-Control', 'no-cache').json(aRepresentacion(pedido));
  } catch (err) { next(err); }
});

docker run -d --name pg-pedidos -p 5432:5432 -e POSTGRES_USER=svc_pedidos -e POSTGRES_PASSWORD=dev-pedidos -e POSTGRES_DB=pedidos postgres:16
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
cp .env.ejemplo .env && npm run migrar && node scripts/stubClientes.js &   # stub en 3004
npm run dev

# 1. Crear el pedido de Ana → 202
curl -s -i -X POST http://localhost:3002/v1/pedidos -H 'Content-Type: application/json' -H 'Idempotency-Key: 7f3c9a2e-1b4d-4e8f-9c21-5a6b7c8d9e0f' \
  -d '{"clienteId":"c-1024","lineas":[{"productoId":"p-501","cantidad":1},{"productoId":"p-777","cantidad":2}],
       "direccionEnvio":{"calle":"Gran Vía 12","codigoPostal":"28013","ciudad":"Madrid","pais":"ES"}}'
# HTTP/1.1 202 Accepted   Location: /v1/pedidos/ped-3f9a1c2b   cuerpo: {"id":"ped-3f9a1c2b","estado":"PENDIENTE","total":79.7,...}

# 2. Repetir la misma petición → 202 con el MISMO id (idempotencia); cambiar cantidad con la misma clave → 422 CLAVE_IDEMPOTENCIA_REUTILIZADA

# 3. Consultar → PENDIENTE (y en RabbitMQ Management, cola inventario.pedidos: 1 mensaje pedido.creado si Inventario no está arrancado)
curl -s http://localhost:3002/v1/pedidos/ped-3f9a1c2b | jq .estado      # "PENDIENTE"

# 4. Simular a Inventario: publicar stock.reservado con el script
node scripts/publicarEvento.js stock.reservado '{"pedidoId":"ped-3f9a1c2b","reservaId":"res-4471"}'
curl -s http://localhost:3002/v1/pedidos/ped-3f9a1c2b | jq .estado      # "STOCK_RESERVADO"

# 5. Simular a Pagos → CONFIRMADO, y en la cola notificaciones.pedidos aparece pedido.confirmado
node scripts/publicarEvento.js pago.confirmado '{"pedidoId":"ped-3f9a1c2b","pagoId":"pag-9001","importe":79.70}'
curl -s http://localhost:3002/v1/pedidos/ped-3f9a1c2b | jq .estado      # "CONFIRMADO"

# 6. Repetir el paso 4 → el estado NO cambia (transicion_ignorada en el log): idempotencia + máquina de estados

// scripts/publicarEvento.js — publica un sobre estándar en techcorp.eventos: node scripts/publicarEvento.js <tipo> '<carga JSON>'
const { conectar } = require('@techcorp/comun-http/mensajeria/topologia');
const { construirSobre, publicarEvento } = require('@techcorp/comun-http/mensajeria/publicador');
(async () => {
  const [tipo, cargaJson] = process.argv.slice(2);
  const { conexion, canal } = await conectar(process.env.RABBITMQ_URL ?? 'amqp://localhost:5672');
  const sobre = construirSobre(tipo, JSON.parse(cargaJson));
  publicarEvento(canal, sobre);
  console.log('publicado', sobre.eventoId, tipo);
  await canal.close(); await conexion.close();
})();

async function manejar(sobre, tx) {
  const { clienteId, nombre, email, actualizadoEn } = sobre.carga;
  await tx.consultar(
    `INSERT INTO clientes_ref (cliente_id, nombre, email, actualizado_en) VALUES ($1,$2,$3,$4)
     ON CONFLICT (cliente_id) DO UPDATE SET nombre = EXCLUDED.nombre, email = EXCLUDED.email, actualizado_en = EXCLUDED.actualizado_en
     WHERE clientes_ref.actualizado_en < EXCLUDED.actualizado_en`,
    [clienteId, nombre, email, actualizadoEn]);
}