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

orders-service/
├── src/
│   ├── server.js, app.js, config.js (04-03), health.js
│   ├── routes/orders.js                    # POST /v1/orders, GET /v1/orders/:id
│   ├── use-cases/createOrder.js
│   ├── domain/order.js                     # aggregate: create, compute total, apply event
│   ├── domain/orderStateMachine.js         # TRANSITIONS, REASONS, transition (02-05, as is)
│   ├── repositories/orderRepository.js     # saveWithEvents, get, idempotency keys, processOnce
│   ├── clients/catalogClient.js, clients/customersClient.js
│   ├── translators/productTranslator.js    # ACL (02-03, 03-06)
│   ├── infra/postgres.js
│   └── messaging/outboxRelay.js, sagaConsumer.js, customersConsumer.js
├── migrations/001-initial-schema.sql … 004-idempotency-keys.sql
├── scripts/migrate.js, scripts/publishEvent.js
└── contracts/openapi.yaml, contracts/asyncapi.yaml

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

function createPostgresPool({ url, logger }) {
  const pool = new Pool({ connectionString: url, max: 10, connectionTimeoutMillis: 5000, idleTimeoutMillis: 30000 });
  pool.on('error', (err) => logger.error({ err }, 'error on idle pool connection'));   // without this, an error takes the process down
  return {
    query: (sql, params) => pool.query(sql, params),                     // outside a transaction
    // Runs fn(tx) inside BEGIN/COMMIT; ROLLBACK if fn throws. tx.query ALWAYS uses the same connection.
    async transaction(fn) {
      const client = await pool.connect();
      try {
        await client.query('BEGIN');
        const result = await fn({ query: (sql, params) => client.query(sql, params) });
        await client.query('COMMIT');
        return result;
      } catch (err) { await client.query('ROLLBACK'); throw err; }
      finally { client.release(); }
    },
    ping: () => pool.query('SELECT 1'),
    close: () => pool.end()
  };
}
module.exports = { createPostgresPool };

// src/translators/productTranslator.js — the tolerant reader from 03-06, unchanged
function toOrdersProduct(dto) {
  return { productId: dto.id, name: dto.name, unitPrice: Number(dto.price), available: dto.available !== false };
}
module.exports = { toOrdersProduct };

// src/clients/catalogClient.js
const { BusinessError } = require('@techcorp/common-http');
const { toOrdersProduct } = require('../translators/productTranslator');

function createCatalogClient({ baseUrl, timeoutMs }) {
  return {
    // Returns a Map productId → { productId, name, unitPrice, available }
    async getProducts(ids, { requestId }) {
      const url = `${baseUrl}/v1/products?ids=${encodeURIComponent(ids.join(','))}`;
      let response;
      try {
        response = await fetch(url, { headers: { Accept: 'application/json', 'X-Request-Id': requestId }, signal: AbortSignal.timeout(timeoutMs) });
      } catch (err) {                                     // timeout or network: 503 for our caller (03-01)
        throw new BusinessError('DEPENDENCY_UNAVAILABLE', `Catalog unavailable: ${err.name}`, 503);
      }
      if (!response.ok) {
        const problem = await response.json().catch(() => ({}));
        if (response.status >= 500) throw new BusinessError('DEPENDENCY_UNAVAILABLE', `Catalog responded ${response.status}`, 503);
        throw new BusinessError('INVALID_REQUEST', `Catalog rejected the request (${problem.code ?? response.status})`, 400);
      }
      const { data, notFound } = await response.json();
      const products = data.map(toOrdersProduct);
      const notSellable = [...notFound, ...products.filter((p) => !p.available).map((p) => p.productId)];
      if (notSellable.length > 0) throw new BusinessError('PRODUCT_UNAVAILABLE', `Unavailable products: ${notSellable.join(', ')}`, 422);
      return new Map(products.map((p) => [p.productId, p]));
    }
  };
}
module.exports = { createCatalogClient };

// src/domain/order.js
const { randomUUID } = require('node:crypto');
const { transition, REASONS } = require('./orderStateMachine');

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

// Builds a PENDING Order from the validated request and from what Customers and Catalog said
function createOrder({ customerId, lines, shippingAddress }, { customer, products }) {
  const frozenLines = lines.map((l, i) => {
    const p = products.get(l.productId);              // catalogClient already guaranteed it exists and is sellable
    return { line: i + 1, productId: p.productId, productName: p.name, unitPrice: p.unitPrice, quantity: l.quantity };
  });
  const totalCents = frozenLines.reduce((acc, l) => acc + toCents(l.unitPrice) * l.quantity, 0);
  return {
    orderId: `ord-${randomUUID().slice(0, 8)}`,       // opaque id generated by the owner (02-04)
    customerId, customer: { name: customer.name, email: customer.email },
    status: 'PENDING', lines: frozenLines, shippingAddress,
    total: totalCents / 100, cancellationReason: null, createdAt: new Date().toISOString()
  };
}

// Applies a saga event; returns the new status or null if there is no transition (late/duplicate event)
function applySagaEvent(order, eventType) {
  const newStatus = transition(order.status, eventType);
  if (!newStatus) return null;
  order.status = newStatus;
  if (newStatus === 'CANCELLED') order.cancellationReason = REASONS[eventType];
  return newStatus;
}

// Payload that travels in order.created / order.confirmed / order.cancelled (02-05 contract and 03-06 AsyncAPI)
function dataForConsumers(o) {
  return { orderId: o.orderId, customerId: o.customerId, customer: o.customer, shippingAddress: o.shippingAddress, total: o.total,
           lines: o.lines.map((l) => ({ productId: l.productId, name: l.productName, quantity: l.quantity, unitPrice: l.unitPrice })) };
}
module.exports = { createOrder, applySagaEvent, dataForConsumers };

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

function createOrderRepository(db) {
  // Saves order + lines + customer replica + events in the outbox, in ONE transaction (02-05 §7).
  // `tx` optional: if the caller is already in a transaction (processOnce), we reuse theirs.
  async function saveWithEvents(order, events, tx) {
    const work = async (t) => {
      await t.query(`INSERT INTO orders (order_id, customer_id, status, total, shipping_address, cancellation_reason, created_at, updated_at)
                     VALUES ($1,$2,$3,$4,$5,$6,$7,NOW())
                     ON CONFLICT (order_id) DO UPDATE SET status = EXCLUDED.status, cancellation_reason = EXCLUDED.cancellation_reason, updated_at = NOW()`,
        [order.orderId, order.customerId, order.status, order.total, order.shippingAddress, order.cancellationReason, order.createdAt]);
      for (const l of order.lines) {                        // lines never change after creation: idempotent insert
        await t.query(`INSERT INTO order_lines (order_id, line, product_id, product_name, unit_price, quantity)
                       VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT DO NOTHING`, [order.orderId, l.line, l.productId, l.productName, l.unitPrice, l.quantity]);
      }
      if (order.customer) {                                 // customers_ref replica (02-04): we warm it up with what we already know
        await t.query(`INSERT INTO customers_ref (customer_id, name, email, updated_at) VALUES ($1,$2,$3,NOW())
                       ON CONFLICT (customer_id) DO NOTHING`, [order.customerId, order.customer.name, order.customer.email]);
      }
      for (const ev of events) {                            // the outbox: same INSERTs, same transaction
        await t.query(`INSERT INTO outbox (event_id, aggregate_type, aggregate_id, type, version, payload) VALUES ($1,'Order',$2,$3,$4,$5)`,
          [`evt-${randomUUID()}`, order.orderId, ev.type, ev.version ?? 1, ev.payload]);
      }
    };
    return tx ? work(tx) : db.transaction(work);
  }

  async function get(orderId, tx = db) {
    const { rows } = await tx.query(`SELECT o.*, c.name AS customer_name, c.email AS customer_email
                                     FROM orders o LEFT JOIN customers_ref c ON c.customer_id = o.customer_id WHERE o.order_id = $1`, [orderId]);
    if (rows.length === 0) return null;
    const lines = (await tx.query('SELECT * FROM order_lines WHERE order_id = $1 ORDER BY line', [orderId])).rows;
    return toOrder(rows[0], lines);                        // row → aggregate (camelCase names; ~10 lines, omitted)
  }

  // API idempotency (02-05 §8c, 03-01): key → stored response
  const findKey = async (key) => (await db.query('SELECT fingerprint, response FROM idempotency_keys WHERE key = $1', [key])).rows[0] ?? null;
  const saveKey = (t, key, orderId, fingerprint, response) =>
    t.query('INSERT INTO idempotency_keys (key, order_id, fingerprint, response) VALUES ($1,$2,$3,$4)', [key, orderId, fingerprint, response]);

  // Consumer idempotency (02-05 §8b, 03-02): effect + record in the SAME transaction
  async function processOnce(eventId, consumer, fn) {
    return db.transaction(async (tx) => {
      const { rowCount } = await tx.query('INSERT INTO processed_events (event_id, consumer) VALUES ($1,$2) ON CONFLICT DO NOTHING', [eventId, consumer]);
      if (rowCount === 0) return 'DUPLICATE';              // already processed: the transaction does nothing else
      await fn(tx);                                        // the real effect, with the same tx
      return 'PROCESSED';
    });
  }

  return { saveWithEvents, get, findKey, saveKey, processOnce, transaction: db.transaction };
}
module.exports = { createOrderRepository };

// src/use-cases/createOrder.js
const { createHash } = require('node:crypto');
const { BusinessError } = require('@techcorp/common-http');
const Order = require('../domain/order');

const fingerprintOf = (body) => createHash('sha256').update(JSON.stringify(body)).digest('hex');

function createCreateOrderUseCase({ repository, catalogClient, customersClient, logger }) {
  return async function createOrder(request, { idempotencyKey, requestId }) {
    // 1. Idempotency: same key + same body → same response; same key + different body → 422 (03-01)
    const fingerprint = fingerprintOf(request);
    const previous = await repository.findKey(idempotencyKey);
    if (previous) {
      if (previous.fingerprint !== fingerprint) throw new BusinessError('IDEMPOTENCY_KEY_REUSED', 'The Idempotency-Key was already used with a different body', 422);
      return { order: previous.response, repeated: true };
    }
    // 2. Customer and products IN PARALLEL: they are independent, so latency is that of the slowest, not the sum
    const [customer, products] = await Promise.all([
      customersClient.getCustomer(request.customerId, { requestId }),                              // CUSTOMER_NOT_FOUND → 404
      catalogClient.getProducts([...new Set(request.lines.map((l) => l.productId))], { requestId })  // PRODUCT_UNAVAILABLE → 422
    ]);
    // 3. The aggregate, PENDING, with frozen names and prices and the computed total
    const order = Order.createOrder(request, { customer, products });
    const response = toRepresentation(order);           // the JSON from 03-01 (id, status, lines, total, _links); ~8 lines, omitted
    // 4. Order + order.created + idempotency key: ONE transaction. No RabbitMQ or HTTP in here.
    await repository.transaction(async (tx) => {
      await repository.saveWithEvents(order, [{ type: 'order.created', version: 1, payload: Order.dataForConsumers(order) }], tx);
      await repository.saveKey(tx, idempotencyKey, order.orderId, fingerprint, response);
    });
    logger.info({ orderId: order.orderId, customerId: order.customerId, total: order.total, requestId }, 'order created');
    return { order: response, repeated: false };
  };
}
module.exports = { createCreateOrderUseCase };

// src/routes/orders.js (POST excerpt)
router.post('/v1/orders', async (req, res, next) => {
  try {
    const idempotencyKey = req.get('Idempotency-Key');
    if (!idempotencyKey) throw new BusinessError('INVALID_REQUEST', 'Missing Idempotency-Key header', 400);
    const request = newOrderSchema.parse(req.body);   // zod: customerId, lines[{productId, quantity ≥ 1}], shippingAddress (400/422 via middleware)
    const { order } = await createOrder(request, { idempotencyKey, requestId: req.id });
    res.status(202).location(`/v1/orders/${order.id}`).json(order);
  } catch (err) { next(err); }
});

// src/messaging/outboxRelay.js
const { buildEnvelope, publishEvent } = require('@techcorp/common-http/messaging/publisher');   // 03-02

function createOutboxRelay({ db, confirmChannel, intervalMs, batchSize = 50, logger }) {
  let timer = null, stopped = false;

  async function publishPending() {
    // One transaction per batch. FOR UPDATE SKIP LOCKED: if there are two Orders replicas, each takes different rows.
    const published = await db.transaction(async (tx) => {
      const { rows } = await tx.query(
        `SELECT event_id, type, version, payload FROM outbox WHERE published_at IS NULL ORDER BY created_at LIMIT $1 FOR UPDATE SKIP LOCKED`, [batchSize]);
      if (rows.length === 0) return 0;
      for (const row of rows) {
        // The envelope carries the outbox eventId (not a new one): if the relay retries, the consumer recognizes it as a duplicate
        const envelope = { ...buildEnvelope(row.type, row.payload, { version: row.version }), eventId: row.event_id };
        publishEvent(confirmChannel, envelope);
      }
      await confirmChannel.waitForConfirms();              // the broker confirms it has received (and persisted) the batch
      await tx.query(`UPDATE outbox SET published_at = NOW() WHERE event_id = ANY($1)`, [rows.map((r) => r.event_id)]);
      return rows.length;                                  // COMMIT: only now are they marked
    });
    if (published > 0) logger.debug({ published }, 'outbox published');
    return published;
  }

  async function cycle() {
    if (stopped) return;
    try {
      const n = await publishPending();
      timer = setTimeout(cycle, n === batchSize ? 0 : intervalMs);   // if the batch was full, keep going without waiting
    } catch (err) {
      logger.error({ err }, 'outbox relay: failure, retrying next cycle');   // RabbitMQ down: the rows stay pending
      timer = setTimeout(cycle, intervalMs);
    }
  }
  return { start: () => cycle(), stop: () => { stopped = true; clearTimeout(timer); } };
}
module.exports = { createOutboxRelay };

// src/messaging/sagaConsumer.js
const { declareConsumerQueue } = require('@techcorp/common-http/messaging/topology');   // 03-02
const Order = require('../domain/order');

const QUEUE = 'orders.saga';
const EVENTS = ['stock.reserved', 'stock.rejected', 'payment.confirmed', 'payment.rejected'];

function createSagaConsumer({ channel, repository, logger }) {
  // Effect of a saga event. Runs INSIDE processOnce, with its tx.
  async function handle(envelope, tx) {
    const order = await repository.get(envelope.payload.orderId, tx);
    if (!order) { logger.warn({ envelope }, 'event for unknown order'); return; }   // do not retry: ack (it stays recorded as processed)

    let newStatus = Order.applySagaEvent(order, envelope.type);
    if (!newStatus) { logger.info({ orderId: order.orderId, from: order.status, event: envelope.type }, 'transition_ignored'); return; }
    if (newStatus === 'PAID') newStatus = Order.applySagaEvent(order, 'confirm');   // T4 from 02-05: today it is immediate

    const events = [];
    if (newStatus === 'CONFIRMED') events.push({ type: 'order.confirmed', payload: Order.dataForConsumers(order) });
    if (newStatus === 'CANCELLED') events.push({ type: 'order.cancelled', payload: { ...Order.dataForConsumers(order), reason: order.cancellationReason } });
    await repository.saveWithEvents(order, events, tx);              // new status + outgoing events, same transaction as processed_events
    logger.info({ orderId: order.orderId, status: newStatus, event: envelope.type }, 'order updated');
  }

  async function start() {
    await declareConsumerQueue(channel, QUEUE, EVENTS);
    await channel.prefetch(10);
    await channel.consume(QUEUE, async (msg) => {
      if (!msg) return;
      let envelope;
      try { envelope = JSON.parse(msg.content.toString()); } catch { return channel.nack(msg, false, false); }   // unreadable → DLQ
      try {
        await repository.processOnce(envelope.eventId, 'orders.saga', (tx) => handle(envelope, tx));
        channel.ack(msg);
      } catch (err) {
        logger.error({ err, eventId: envelope.eventId }, 'error processing saga event');
        channel.nack(msg, false, !msg.fields.redelivered);              // 1st time: requeue; 2nd: DLQ (03-02)
      }
    });
  }
  return { start };
}
module.exports = { createSagaConsumer };

// src/routes/orders.js (GET excerpt)
router.get('/v1/orders/:id', async (req, res, next) => {
  try {
    const order = await repository.get(req.params.id);
    if (!order) throw new BusinessError('ORDER_NOT_FOUND', `${req.params.id} does not exist`, 404);
    const etag = `"${order.orderId}:${new Date(order.updatedAt).getTime()}"`;   // changes with every transition
    if (req.get('If-None-Match') === etag) return res.status(304).end();       // cheap polling (03-01)
    res.set('ETag', etag).set('Cache-Control', 'no-cache').json(toRepresentation(order));
  } catch (err) { next(err); }
});

docker run -d --name pg-orders -p 5432:5432 -e POSTGRES_USER=svc_orders -e POSTGRES_PASSWORD=dev-orders -e POSTGRES_DB=orders postgres:16
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
cp .env.example .env && npm run migrate && node scripts/stubCustomers.js &   # stub on 3004
npm run dev

# 1. Create Ana's order → 202
curl -s -i -X POST http://localhost:3002/v1/orders -H 'Content-Type: application/json' -H 'Idempotency-Key: 7f3c9a2e-1b4d-4e8f-9c21-5a6b7c8d9e0f' \
  -d '{"customerId":"c-1024","lines":[{"productId":"p-501","quantity":1},{"productId":"p-777","quantity":2}],
       "shippingAddress":{"street":"Gran Vía 12","postalCode":"28013","city":"Madrid","country":"ES"}}'
# HTTP/1.1 202 Accepted   Location: /v1/orders/ord-3f9a1c2b   body: {"id":"ord-3f9a1c2b","status":"PENDING","total":79.7,...}

# 2. Repeat the same request → 202 with the SAME id (idempotency); change a quantity with the same key → 422 IDEMPOTENCY_KEY_REUSED

# 3. Query → PENDING (and in RabbitMQ Management, queue inventory.orders: 1 order.created message if Inventory is not running)
curl -s http://localhost:3002/v1/orders/ord-3f9a1c2b | jq .status      # "PENDING"

# 4. Simulate Inventory: publish stock.reserved with the script
node scripts/publishEvent.js stock.reserved '{"orderId":"ord-3f9a1c2b","reservationId":"res-4471"}'
curl -s http://localhost:3002/v1/orders/ord-3f9a1c2b | jq .status      # "STOCK_RESERVED"

# 5. Simulate Payments → CONFIRMED, and order.confirmed shows up in the notifications.orders queue
node scripts/publishEvent.js payment.confirmed '{"orderId":"ord-3f9a1c2b","paymentId":"pay-9001","amount":79.70}'
curl -s http://localhost:3002/v1/orders/ord-3f9a1c2b | jq .status      # "CONFIRMED"

# 6. Repeat step 4 → the status does NOT change (transition_ignored in the log): idempotency + state machine

// scripts/publishEvent.js — publishes a standard envelope to techcorp.events: node scripts/publishEvent.js <type> '<JSON payload>'
const { connect } = require('@techcorp/common-http/messaging/topology');
const { buildEnvelope, publishEvent } = require('@techcorp/common-http/messaging/publisher');
(async () => {
  const [type, payloadJson] = process.argv.slice(2);
  const { connection, channel } = await connect(process.env.RABBITMQ_URL ?? 'amqp://localhost:5672');
  const envelope = buildEnvelope(type, JSON.parse(payloadJson));
  publishEvent(channel, envelope);
  console.log('published', envelope.eventId, type);
  await channel.close(); await connection.close();
})();

async function handle(envelope, tx) {
  const { customerId, name, email, updatedAt } = envelope.payload;
  await tx.query(
    `INSERT INTO customers_ref (customer_id, name, email, updated_at) VALUES ($1,$2,$3,$4)
     ON CONFLICT (customer_id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email, updated_at = EXCLUDED.updated_at
     WHERE customers_ref.updated_at < EXCLUDED.updated_at`,
    [customerId, name, email, updatedAt]);
}