-- Semana 16: el envío 501 como un documento JSON, al estilo de un document
-- store, armado desde las tablas de Chasqui con jsonb (semana 8):
--   psql -d chasqui_w16 -f documents.sql
-- Solo lee.

-- El documento: el envío con su lista de eventos adentro, y el nombre de la
-- agencia copiado en cada evento. \a y \t muestran el texto tal cual.
\a
\t
-- name: document_501
SELECT jsonb_pretty(jsonb_build_object(
  'shipment_id', s.shipment_id,
  'status', s.status,
  'weight_kg', s.weight_kg,
  'origin', o.name,
  'destination', d.name,
  'events', (SELECT jsonb_agg(jsonb_build_object(
                      'seq', e.seq, 'kind', e.kind, 'agency', a.name)
                    ORDER BY e.seq)
             FROM shipment_event e JOIN agency a ON a.agency_id = e.agency_id
             WHERE e.shipment_id = s.shipment_id))) AS doc
FROM shipment s
  JOIN agency o ON o.agency_id = s.origin_agency_id
  JOIN agency d ON d.agency_id = s.dest_agency_id
WHERE s.shipment_id = 501;
\a
\t

-- La copia tiene un precio: si la agencia 1 cambia de nombre, ¿cuántos
-- documentos habría que reescribir? Todos los envíos que pasaron por ella.
-- name: docs_to_rewrite
SELECT count(DISTINCT shipment_id) AS documents, count(*) AS copies
FROM shipment_event
WHERE agency_id = 1;

// Un MapReduce de juguete, en un solo proceso, para ver sus fases: la entrada
// se parte en splits, cada split pasa por map (y por el combiner, si hay),
// el shuffle agrupa los pares por clave y los reparte entre los reducers por
// hash de la clave, y cada reducer llama a reduce(clave, valores).
//
//   PGDATABASE=chasqui_w16 node mapreduce.mjs
// cuenta las entregas por ciudad de destino a partir de los eventos de
// seguimiento, con y sin combiner, y las compara con un GROUP BY.

// Un hash de texto pequeño y determinista (FNV-1a de 32 bits).
function hash(text) {
  let h = 0x811c9dc5;
  for (const ch of text) {
    h ^= ch.codePointAt(0);
    h = Math.imul(h, 0x01000193) >>> 0;
  }
  return h;
}

// records: arreglo de entradas. map(record, emit) llama a emit(clave, valor)
// las veces que quiera. reduce(clave, valores) devuelve un valor.
// Devuelve { result: Map(clave → valor), stats }.
export function mapReduce(records, map, reduce, { splits = 4, reducers = 3, combiner } = {}) {
  const stats = { splits, reducers, mapped: 0, shuffled: 0, perReducer: Array(reducers).fill(0) };

  // Fase map: cada split produce sus pares (clave, valor).
  const size = Math.ceil(records.length / splits);
  const mapOutputs = [];
  for (let s = 0; s < splits; s++) {
    let pairs = [];
    for (const record of records.slice(s * size, (s + 1) * size)) {
      map(record, (key, value) => pairs.push([key, value]));
    }
    stats.mapped += pairs.length;
    // Combiner: un reduce local, en la máquina del map, antes de mandar nada.
    if (combiner) pairs = [...groupByKey(pairs)].map(([k, vs]) => [k, combiner(k, vs)]);
    mapOutputs.push(pairs);
  }

  // Shuffle: cada par viaja al reducer hash(clave) mod reducers.
  const inboxes = Array.from({ length: reducers }, () => []);
  for (const pairs of mapOutputs) {
    for (const [key, value] of pairs) {
      const r = hash(String(key)) % reducers;
      inboxes[r].push([key, value]);
      stats.shuffled++;
      stats.perReducer[r]++;
    }
  }

  // Fase reduce: cada reducer agrupa lo que recibió y reduce cada clave.
  const result = new Map();
  for (const inbox of inboxes) {
    for (const [key, values] of groupByKey(inbox)) result.set(key, reduce(key, values));
  }
  return { result, stats };
}

function groupByKey(pairs) {
  const groups = new Map();
  for (const [key, value] of pairs) {
    if (!groups.has(key)) groups.set(key, []);
    groups.get(key).push(value);
  }
  return groups;
}

// El ejemplo de clase: map emite (ciudad, 1) por cada evento de entrega;
// reduce suma. El mismo reduce sirve de combiner porque la suma es
// asociativa y conmutativa.
export const deliveriesMap = (event, emit) => {
  if (event.kind === 'delivered') emit(event.city, 1);
};
export const sumReduce = (_key, values) => values.reduce((a, b) => a + b, 0);

// Los "archivos" de eventos: una línea por evento de seguimiento, con la
// ciudad de la agencia donde ocurrió.
export const EVENTS_SQL = `
  SELECT e.shipment_id, e.seq, e.kind, a.city
  FROM shipment_event e JOIN agency a ON a.agency_id = e.agency_id
  ORDER BY e.shipment_id, e.seq`;

// La consulta SQL equivalente.
export const DELIVERIES_SQL = `
  SELECT a.city, count(*)::int AS n
  FROM shipment_event e JOIN agency a ON a.agency_id = e.agency_id
  WHERE e.kind = 'delivered'
  GROUP BY a.city`;

if (import.meta.url === `file://${process.argv[1]}`) {
  const { default: pg } = await import('pg');
  const pool = new pg.Pool();
  try {
    const events = (await pool.query(EVENTS_SQL)).rows;
    console.log(`${events.length} eventos`);
    for (const combiner of [undefined, sumReduce]) {
      const { result, stats } = mapReduce(events, deliveriesMap, sumReduce, { combiner });
      console.log(`\n${combiner ? 'con' : 'sin'} combiner: ${stats.mapped} pares emitidos por map, `
        + `${stats.shuffled} enviados en el shuffle (${stats.perReducer.join(' + ')})`);
      const top = [...result].sort((a, b) => b[1] - a[1]).slice(0, 5);
      console.log(top.map(([city, n]) => `  ${city.padEnd(10)} ${n}`).join('\n'));
    }
    const sql = (await pool.query(`${DELIVERIES_SQL} ORDER BY n DESC LIMIT 5`)).rows;
    console.log('\nGROUP BY en PostgreSQL:');
    console.log(sql.map(({ city, n }) => `  ${city.padEnd(10)} ${n}`).join('\n'));
  } finally {
    await pool.end();
  }
}

-- Semana 16: tres copias particionadas de shipment (y una de shipment_event)
-- dentro de chasqui_w16. `make` corre este archivo; se puede volver a correr:
--   psql -d chasqui_w16 -f partitions.sql
-- Las tablas originales no se tocan.

SET client_min_messages = warning;
DROP TABLE IF EXISTS shipment_by_month, shipment_by_hash, shipment_event_by_hash,
  shipment_by_zone;

-- 1. Por rango: una partición por mes de created_at. La clave primaria de una
-- tabla particionada tiene que incluir la clave de particionamiento.
CREATE TABLE shipment_by_month (
  LIKE shipment INCLUDING DEFAULTS INCLUDING CONSTRAINTS,
  PRIMARY KEY (shipment_id, created_at)
) PARTITION BY RANGE (created_at);

CREATE TABLE shipment_2026_01 PARTITION OF shipment_by_month
  FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');
CREATE TABLE shipment_2026_02 PARTITION OF shipment_by_month
  FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');
CREATE TABLE shipment_2026_03 PARTITION OF shipment_by_month
  FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');
CREATE TABLE shipment_2026_04 PARTITION OF shipment_by_month
  FOR VALUES FROM ('2026-04-01') TO ('2026-05-01');
CREATE TABLE shipment_2026_05 PARTITION OF shipment_by_month
  FOR VALUES FROM ('2026-05-01') TO ('2026-06-01');
CREATE TABLE shipment_2026_06 PARTITION OF shipment_by_month
  FOR VALUES FROM ('2026-06-01') TO ('2026-07-01');
-- Lo que no cae en ningún mes va aquí (sin ella, el INSERT falla).
CREATE TABLE shipment_month_default PARTITION OF shipment_by_month DEFAULT;

INSERT INTO shipment_by_month SELECT * FROM shipment;

-- 2. Por hash de shipment_id, en 4 particiones; shipment_event, con la misma
-- clave y el mismo número de particiones, para hacer el join por partes.
CREATE TABLE shipment_by_hash (
  LIKE shipment INCLUDING DEFAULTS INCLUDING CONSTRAINTS,
  PRIMARY KEY (shipment_id)
) PARTITION BY HASH (shipment_id);

CREATE TABLE shipment_h0 PARTITION OF shipment_by_hash FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE shipment_h1 PARTITION OF shipment_by_hash FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE shipment_h2 PARTITION OF shipment_by_hash FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE shipment_h3 PARTITION OF shipment_by_hash FOR VALUES WITH (MODULUS 4, REMAINDER 3);

CREATE TABLE shipment_event_by_hash (
  LIKE shipment_event INCLUDING DEFAULTS INCLUDING CONSTRAINTS,
  PRIMARY KEY (shipment_id, seq)
) PARTITION BY HASH (shipment_id);

CREATE TABLE shipment_event_h0 PARTITION OF shipment_event_by_hash FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE shipment_event_h1 PARTITION OF shipment_event_by_hash FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE shipment_event_h2 PARTITION OF shipment_event_by_hash FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE shipment_event_h3 PARTITION OF shipment_event_by_hash FOR VALUES WITH (MODULUS 4, REMAINDER 3);

INSERT INTO shipment_by_hash SELECT * FROM shipment;
INSERT INTO shipment_event_by_hash SELECT * FROM shipment_event;

-- 3. Por lista de agencias de origen: cuatro zonas, como cuatro servidores.
-- Sin partición DEFAULT: una agencia nueva que no está en ninguna lista falla.
CREATE TABLE shipment_by_zone (
  LIKE shipment INCLUDING DEFAULTS INCLUDING CONSTRAINTS,
  PRIMARY KEY (shipment_id, origin_agency_id)
) PARTITION BY LIST (origin_agency_id);

CREATE TABLE shipment_lima PARTITION OF shipment_by_zone
  FOR VALUES IN (1, 2, 3);
CREATE TABLE shipment_norte PARTITION OF shipment_by_zone
  FOR VALUES IN (7, 8, 9, 10, 11, 12, 22);
CREATE TABLE shipment_sur PARTITION OF shipment_by_zone
  FOR VALUES IN (4, 5, 6, 13, 17, 18, 19, 20, 21, 28);
CREATE TABLE shipment_centro_oriente PARTITION OF shipment_by_zone
  FOR VALUES IN (14, 15, 16, 23, 24, 25, 26, 27, 29);

INSERT INTO shipment_by_zone SELECT * FROM shipment;

ANALYZE shipment_by_month, shipment_by_hash, shipment_event_by_hash, shipment_by_zone;

-- Semana 16: PREPARE TRANSACTION a mano, en la base oriente del cluster
-- desechable (make cluster):
--   psql -h /tmp -p 5416 -U postgres -d oriente -f prepared.sql
-- Una transacción preparada sobrevive al fin de la sesión y a una caída del
-- servidor, y conserva sus locks hasta COMMIT PREPARED o ROLLBACK PREPARED.

BEGIN;
UPDATE trip SET booked_kg = booked_kg + 5 WHERE trip_id = 7001;
PREPARE TRANSACTION 'manual-1@oriente';

-- La sesión ya no tiene una transacción abierta, pero la transacción existe.
-- name: prepared_xacts
SELECT gid, database, owner, transaction AS xid
FROM pg_prepared_xacts;

-- Sus locks siguen ahí, sin un proceso que sea su owner (pid vacío).
-- name: prepared_locks
SELECT locktype, relation::regclass AS relation, transactionid, mode, granted, pid
FROM pg_locks
WHERE pid IS NULL
ORDER BY locktype, relation;

-- Lo que ve cualquier otra transacción (también una nueva de esta sesión):
-- el cambio todavía no está confirmado.
-- name: before_commit
SELECT booked_kg FROM trip WHERE trip_id = 7001;

-- Fase 2. Puede ejecutarla cualquier sesión en la misma base de datos.
COMMIT PREPARED 'manual-1@oriente';

-- name: after_commit
SELECT booked_kg FROM trip WHERE trip_id = 7001;

-- Para dejarlo como estaba:
UPDATE trip SET booked_kg = booked_kg - 5 WHERE trip_id = 7001;

-- Semana 16: particiones, partition pruning y join por particiones, sobre las
-- tablas que crea partitions.sql (make las crea). Solo lee:
--   psql -d chasqui_w16 -f pruning.sql
-- Cada consulta empieza con una marca "-- name:" que usan las pruebas.

-- Cuántas filas quedaron en cada partición. tableoid es la tabla física de
-- cada fila; ::regclass la muestra con su nombre.
-- name: by_month
SELECT tableoid::regclass AS partition, count(*) AS n
FROM shipment_by_month
GROUP BY tableoid
ORDER BY tableoid::regclass::text;

-- name: by_hash
SELECT tableoid::regclass AS partition, count(*) AS n
FROM shipment_by_hash
GROUP BY tableoid
ORDER BY tableoid::regclass::text;

-- name: by_zone
SELECT tableoid::regclass AS partition, count(*) AS n,
  round(100.0 * count(*) / sum(count(*)) OVER (), 1) AS pct
FROM shipment_by_zone
GROUP BY tableoid
ORDER BY n DESC;

-- Los límites de cada partición, como los guarda el catálogo.
-- name: bounds
SELECT c.relname AS partition, pg_get_expr(c.relpartbound, c.oid) AS bound
FROM pg_inherits i
  JOIN pg_class c ON c.oid = i.inhrelid
WHERE i.inhparent = 'shipment_by_month'::regclass
ORDER BY c.relname;

-- Pruning en la planificación: la condición sobre created_at descarta todas
-- las particiones menos la de marzo.
-- name: prune_march
EXPLAIN (COSTS OFF)
SELECT count(*)
FROM shipment_by_month
WHERE created_at >= '2026-03-01' AND created_at < '2026-04-01';

-- Sin condición sobre la clave de particionamiento, hay que leer todas.
-- name: no_prune
EXPLAIN (COSTS OFF)
SELECT count(*)
FROM shipment_by_month
WHERE sender_id = 777;

-- Con hash, solo sirve la igualdad: un rango de shipment_id lee las cuatro.
-- name: hash_eq
EXPLAIN (COSTS OFF)
SELECT * FROM shipment_by_hash WHERE shipment_id = 501;

-- name: hash_range
EXPLAIN (COSTS OFF)
SELECT count(*) FROM shipment_by_hash WHERE shipment_id BETWEEN 500 AND 510;

-- Pruning en la ejecución: con un parámetro, el plan genérico no sabe qué
-- mes es; al ejecutarlo, descarta las particiones que sobran.
SET plan_cache_mode = force_generic_plan;
PREPARE since_day(timestamptz) AS
  SELECT count(*) FROM shipment_by_month WHERE created_at >= $1;
-- name: prune_runtime
EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF)
EXECUTE since_day('2026-06-15');
DEALLOCATE since_day;
RESET plan_cache_mode;

-- name: since_june15
SELECT count(*) FROM shipment WHERE created_at >= '2026-06-15';

-- Join por particiones: shipment y shipment_event están particionadas por
-- hash de shipment_id con el mismo módulo. Con enable_partitionwise_join, el
-- join se hace partición contra partición, como lo haría cada servidor con
-- sus propios datos.
SET enable_partitionwise_join = on;
-- name: partitionwise
EXPLAIN (COSTS OFF)
SELECT s.status, count(*)
FROM shipment_by_hash s
  JOIN shipment_event_by_hash e ON e.shipment_id = s.shipment_id
GROUP BY s.status;
RESET enable_partitionwise_join;

-- name: events_per_status
SELECT s.status, count(*)
FROM shipment_by_hash s
  JOIN shipment_event_by_hash e ON e.shipment_id = s.shipment_id
GROUP BY s.status
ORDER BY s.status;

-- Execution skew: todos los envíos nuevos caen en la partición del último
-- mes. Se insertan 50 y se cuentan por partición; ROLLBACK los quita.
BEGIN;
-- name: insert_skew
WITH nuevos AS (
  INSERT INTO shipment_by_month
  SELECT shipment_id + 100000, sender_id, recipient_id, origin_agency_id,
    dest_agency_id, NULL, weight_kg, declared_value, price, 'registered',
    timestamptz '2026-06-30 08:00' + shipment_id * interval '1 minute', NULL
  FROM shipment
  WHERE shipment_id <= 50
  RETURNING tableoid::regclass AS partition
)
SELECT partition, count(*) AS n FROM nuevos GROUP BY partition;
ROLLBACK;

-- Una fila que no cae en ninguna lista: no hay partición DEFAULT en zonas.
-- name: no_partition
INSERT INTO shipment_by_zone
SELECT shipment_id + 100000, sender_id, recipient_id, 30, dest_agency_id,
  NULL, weight_kg, declared_value, price, status, created_at, delivered_at
FROM shipment
WHERE shipment_id = 501;

// Lee un archivo .sql con consultas marcadas "-- name: nombre" y devuelve un
// objeto { nombre: texto }. Así las pruebas corren el mismo texto que el
// archivo que se muestra en clase. Cada consulta va desde la marca hasta la
// primera línea que termina en punto y coma; lo que sigue (comentarios,
// BEGIN, SET, VACUUM) no es parte de ella.
import { readFileSync } from 'node:fs';

export function namedQueries(file) {
  const text = readFileSync(new URL(file, import.meta.url), 'utf8');
  const parts = text.split(/^-- name: (\w+)\n/m);
  const queries = {};
  for (let i = 1; i < parts.length; i += 2) {
    const end = parts[i + 1].search(/;\s*$/m);
    queries[parts[i]] = parts[i + 1].slice(0, end + 1).trim();
  }
  return queries;
}

// Muestra el protocolo 2PC paso a paso en cuatro escenarios, sobre el cluster
// desechable (make cluster):
//   node twopc-demo.mjs
// 1. Todo bien: los dos votan que sí y se hace commit.
// 2. El viaje se llena: oriente vota que no en PREPARE; sur hace rollback.
// 3. El coordinador cae después de la fase 1: los participantes quedan en
//    duda, con locks tomados; la recuperación aborta (presumed abort).
// 4. El coordinador cae después de decidir: la recuperación hace commit.
import pg from 'pg';
import { Coordinator, CoordinatorCrash, bookShipment, seedDemo } from './twopc.mjs';

const conn = (database) => new pg.Pool({ host: '/tmp', port: 5416, user: 'postgres', database });
const pools = { sur: conn('sur'), oriente: conn('oriente'), coordinador: conn('coordinador') };
const coordinator = new Coordinator(
  pools.coordinador,
  { sur: pools.sur, oriente: pools.oriente },
  { trace: (step) => console.log(`  ${step}`) },
);

const state = async () => {
  const trip = (await pools.oriente.query('SELECT booked_kg FROM trip WHERE trip_id = 7001')).rows[0];
  const n = (await pools.sur.query('SELECT count(*)::int AS n FROM shipment')).rows[0].n;
  console.log(`  estado: ${n} envíos en sur; viaje 7001 con ${trip.booked_kg} kg de 1000`);
};

async function scenario(title, gid, booking, options) {
  console.log(`\n${title}`);
  try {
    console.log(`  resultado: ${await coordinator.run(gid, bookShipment(booking), options)}`);
  } catch (err) {
    if (!(err instanceof CoordinatorCrash)) throw err;
    console.log(`  ${err.message}`);
  }
}

try {
  await seedDemo(pools);
  await state();

  await scenario('1. Envío de 35 kg', 'chasqui-1',
    { shipmentId: 90002, trackingCode: 'CUS-IQT-0002', tripId: 7001, weightKg: 35 });
  await state();

  await scenario('2. Envío de 40 kg (ya no cabe)', 'chasqui-2',
    { shipmentId: 90003, trackingCode: 'CUS-IQT-0003', tripId: 7001, weightKg: 40 });
  await state();

  await scenario('3. Envío de 10 kg, el coordinador cae después de la fase 1', 'chasqui-3',
    { shipmentId: 90004, trackingCode: 'CUS-IQT-0004', tripId: 7001, weightKg: 10 },
    { crashAt: 'prepared' });
  console.log('  en duda:', JSON.stringify(await coordinator.inDoubt()));
  const blocked = await pools.oriente.connect();
  try {
    await blocked.query("SET lock_timeout = '500ms'");
    await blocked.query('UPDATE trip SET booked_kg = booked_kg WHERE trip_id = 7001');
  } catch (err) {
    console.log(`  otra transacción sobre el viaje 7001: ${err.code} ${err.message}`);
  } finally {
    blocked.release();
  }
  console.log('  recuperación:', JSON.stringify(await coordinator.recover()));
  await state();

  await scenario('4. Envío de 10 kg, el coordinador cae después de decidir', 'chasqui-4',
    { shipmentId: 90005, trackingCode: 'CUS-IQT-0005', tripId: 7001, weightKg: 10 },
    { crashAt: 'decided' });
  console.log('  en duda:', JSON.stringify(await coordinator.inDoubt()));
  console.log('  recuperación:', JSON.stringify(await coordinator.recover()));
  await state();
} finally {
  await Promise.all(Object.values(pools).map((p) => p.end()));
}

-- Semana 16: las bases de datos de la demostración de 2PC, en el cluster
-- desechable de la semana (puerto 5416, con max_prepared_transactions > 0).
-- `make cluster` lo crea y corre este archivo:
--   psql -h /tmp -p 5416 -U postgres -d postgres -f twopc-setup.sql
--
-- Hacen de tres servidores:
--   sur          los envíos registrados en las agencias del sur (Cusco)
--   oriente      los viajes que salen hacia el oriente (Iquitos)
--   coordinador  el log de decisiones del coordinador de 2PC

SET client_min_messages = warning;
DROP DATABASE IF EXISTS sur;
DROP DATABASE IF EXISTS oriente;
DROP DATABASE IF EXISTS coordinador;
CREATE DATABASE sur;
CREATE DATABASE oriente;
CREATE DATABASE coordinador;

\connect sur
SET client_min_messages = warning;
CREATE TABLE shipment (
  shipment_id   integer PRIMARY KEY,
  tracking_code text NOT NULL,
  origin        text NOT NULL,
  destination   text NOT NULL,
  trip_id       integer NOT NULL,
  weight_kg     numeric(7, 2) NOT NULL CHECK (weight_kg > 0),
  CONSTRAINT shipment_tracking_code_key UNIQUE (tracking_code)
);
-- Un envío ya registrado en el viaje 7001 (seedDemo de twopc.mjs vuelve a
-- dejar estos mismos datos).
INSERT INTO shipment VALUES (90001, 'CUS-IQT-0001', 'Cusco', 'Iquitos', 7001, 940);

\connect oriente
SET client_min_messages = warning;
CREATE TABLE trip (
  trip_id     integer PRIMARY KEY,
  route       text NOT NULL,
  capacity_kg integer NOT NULL CHECK (capacity_kg > 0),
  booked_kg   numeric(8, 2) NOT NULL DEFAULT 0 CHECK (booked_kg >= 0)
);

-- La capacidad se verifica con un constraint trigger diferido: corre al final
-- de la transacción, es decir, en PREPARE TRANSACTION. Así este participante
-- puede votar "no" en la fase 1, cuando el otro ya votó que sí.
CREATE FUNCTION check_trip_capacity() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  IF NEW.booked_kg > NEW.capacity_kg THEN
    RAISE EXCEPTION 'el viaje % no tiene capacidad: % kg de %',
      NEW.trip_id, NEW.booked_kg, NEW.capacity_kg
      USING ERRCODE = 'check_violation';
  END IF;
  RETURN NULL;
END;
$$;

CREATE CONSTRAINT TRIGGER trip_capacity_check
  AFTER INSERT OR UPDATE ON trip
  DEFERRABLE INITIALLY DEFERRED
  FOR EACH ROW EXECUTE FUNCTION check_trip_capacity();

INSERT INTO trip (trip_id, route, capacity_kg, booked_kg)
VALUES (7001, 'Cusco - Iquitos', 1000, 940);

\connect coordinador
SET client_min_messages = warning;
-- Una fila por transacción global que llegó a decidirse commit. Con
-- presumed abort, una transacción sin fila aquí se considera abortada.
CREATE TABLE decision (
  gid        text PRIMARY KEY,
  outcome    text NOT NULL CHECK (outcome = 'commit'),
  decided_at timestamptz NOT NULL DEFAULT now(),
  ended      boolean NOT NULL DEFAULT false
);

// Un coordinador de two-phase commit (2PC) sobre varias bases de datos de
// PostgreSQL. Cada participante es un pg.Pool; la fase 1 es PREPARE
// TRANSACTION y la fase 2, COMMIT PREPARED o ROLLBACK PREPARED. La decisión
// se guarda en la tabla decision de otra base (el log del coordinador), con
// presumed abort: si no hay fila de commit, la transacción se aborta.
//
// Necesita servidores con max_prepared_transactions > 0: el cluster
// desechable que crea `make cluster` (ver twopc-setup.sql).

// Prefijo de los gid de este coordinador: recover() solo toca los suyos.
export const GID_PREFIX = 'chasqui-';

// Una caída simulada del coordinador. at dice en qué punto: 'prepared'
// (todos votaron que sí, no hay decisión) o 'decided' (la decisión está en el
// log, la fase 2 no empezó).
export class CoordinatorCrash extends Error {
  constructor(at) {
    super(`el coordinador cayó (${at})`);
    this.name = 'CoordinatorCrash';
    this.at = at;
  }
}

// En PostgreSQL el gid de una transacción preparada es único en todo el
// servidor, no por base de datos. Aquí sur y oriente están en el mismo
// cluster, así que cada participante prepara su parte con un gid propio:
// el global, '@' y el nombre del participante (como el branch qualifier de
// XA). En servidores distintos bastaría el gid global.
const branchGid = (gid, name) => `${gid}@${name}`;
const globalGid = (branch) => branch.slice(0, branch.lastIndexOf('@'));

// PREPARE TRANSACTION, COMMIT PREPARED y ROLLBACK PREPARED no aceptan
// parámetros ($1): el gid va como literal. Por eso se valida con una lista
// blanca de caracteres y además se escapa.
function gidLiteral(client, gid) {
  if (!/^[a-z0-9-]{1,100}@[a-z]{1,30}$/.test(gid)) throw new Error(`gid inválido: ${gid}`);
  return client.escapeLiteral(gid);
}

export class Coordinator {
  // logPool: la base del coordinador. participants: { nombre: pg.Pool }.
  // trace: función que recibe cada paso, para mostrar el protocolo.
  constructor(logPool, participants, { trace = () => {} } = {}) {
    this.log = logPool;
    this.participants = participants;
    this.trace = trace;
  }

  // Ejecuta work(clients) como una transacción global con identificador gid.
  // clients tiene un pg.Client por participante, con la transacción abierta.
  // Devuelve 'commit' o 'abort'. crashAt simula la caída del coordinador.
  async run(gid, work, { crashAt } = {}) {
    if (!gid.startsWith(GID_PREFIX) || !/^[a-z0-9-]{1,90}$/.test(gid)) throw new Error(`gid inválido: ${gid}`);
    const names = Object.keys(this.participants);
    const clients = {};
    for (const name of names) clients[name] = await this.participants[name].connect();
    const prepared = [];
    try {
      for (const name of names) await clients[name].query('BEGIN');
      try {
        await work(clients);
      } catch (err) {
        // Un participante falló antes de la fase 1: se aborta en todos.
        this.trace(`trabajo falló: ${err.message}`);
        for (const name of names) await clients[name].query('ROLLBACK');
        this.trace('decisión: abort (sin log, presumed abort)');
        return 'abort';
      }

      // Fase 1: cada participante fuerza su <ready T> en su log.
      let allReady = true;
      for (const name of names) {
        try {
          await clients[name].query(`PREPARE TRANSACTION ${gidLiteral(clients[name], branchGid(gid, name))}`);
          prepared.push(name);
          this.trace(`${name}: ready`);
        } catch (err) {
          // PREPARE falló: la transacción de ese participante ya abortó.
          allReady = false;
          this.trace(`${name}: no (${err.message})`);
          break;
        }
      }
      if (!allReady) {
        for (const name of names) {
          if (prepared.includes(name)) {
            await clients[name].query(`ROLLBACK PREPARED ${gidLiteral(clients[name], branchGid(gid, name))}`);
            this.trace(`${name}: rollback prepared`);
          } else {
            await clients[name].query('ROLLBACK'); // no hace nada si ya abortó
          }
        }
        this.trace('decisión: abort (sin log, presumed abort)');
        return 'abort';
      }
      if (crashAt === 'prepared') throw new CoordinatorCrash(crashAt);

      // La decisión: una fila en el log del coordinador. Este INSERT en
      // autocommit es el <commit T> forzado; desde aquí, T hizo commit.
      await this.log.query('INSERT INTO decision (gid, outcome) VALUES ($1, $2)', [gid, 'commit']);
      this.trace('decisión: commit (en el log del coordinador)');
      if (crashAt === 'decided') throw new CoordinatorCrash(crashAt);

      // Fase 2.
      await this.finish(gid, 'commit', names);
      return 'commit';
    } finally {
      // Si el coordinador "cae", las sesiones se cierran; las transacciones
      // preparadas siguen vivas en cada participante, con sus locks.
      for (const name of names) clients[name].release();
    }
  }

  // Fase 2 para gid en los participantes indicados, y <end T> en el log.
  async finish(gid, outcome, names) {
    const verb = outcome === 'commit' ? 'COMMIT PREPARED' : 'ROLLBACK PREPARED';
    for (const name of names) {
      const client = await this.participants[name].connect();
      try {
        await client.query(`${verb} ${gidLiteral(client, branchGid(gid, name))}`);
        this.trace(`${name}: ${verb.toLowerCase()}`);
      } finally {
        client.release();
      }
    }
    if (outcome === 'commit') {
      await this.log.query('UPDATE decision SET ended = true WHERE gid = $1', [gid]);
      this.trace('end');
    }
  }

  // Las transacciones en duda (preparadas y sin terminar) de este
  // coordinador, por participante: { nombre: [gid global, ...] }.
  async inDoubt() {
    const result = {};
    for (const [name, pool] of Object.entries(this.participants)) {
      const { rows } = await pool.query(
        `SELECT gid FROM pg_prepared_xacts
         WHERE database = current_database() AND starts_with(gid, $1)
           AND gid LIKE $2
         ORDER BY gid`,
        [GID_PREFIX, `%@${name}`],
      );
      result[name] = rows.map((r) => globalGid(r.gid));
    }
    return result;
  }

  // Recuperación después de una caída del coordinador: cada transacción en
  // duda se termina según el log. Con commit en el log, COMMIT PREPARED; sin
  // fila, ROLLBACK PREPARED (presumed abort). Devuelve { gid: outcome }.
  async recover() {
    const doubt = await this.inDoubt();
    const gids = [...new Set(Object.values(doubt).flat())];
    const outcomes = {};
    for (const gid of gids) {
      const { rows } = await this.log.query('SELECT outcome FROM decision WHERE gid = $1', [gid]);
      const outcome = rows.length > 0 ? 'commit' : 'abort';
      const names = Object.keys(doubt).filter((n) => doubt[n].includes(gid));
      this.trace(`recuperación de ${gid}: ${rows.length > 0 ? 'commit en el log' : 'sin registro'} → ${outcome}`);
      await this.finish(gid, outcome, names);
      outcomes[gid] = outcome;
    }
    // Decisiones de commit cuya fase 2 ya terminó en todos (por ejemplo, si
    // el coordinador cayó justo antes de anotar el end).
    await this.log.query(
      'UPDATE decision SET ended = true WHERE NOT ended AND NOT (gid = ANY ($1))',
      [gids],
    );
    return outcomes;
  }
}

// La transacción de ejemplo: registrar en el sur el envío y descontar su
// peso de la capacidad del viaje en el oriente. Las dos cosas o ninguna.
export function bookShipment({ shipmentId, trackingCode, tripId, weightKg }) {
  return async ({ sur, oriente }) => {
    await sur.query(
      `INSERT INTO shipment (shipment_id, tracking_code, origin, destination, trip_id, weight_kg)
       VALUES ($1, $2, 'Cusco', 'Iquitos', $3, $4)`,
      [shipmentId, trackingCode, tripId, weightKg],
    );
    const { rowCount } = await oriente.query(
      'UPDATE trip SET booked_kg = booked_kg + $1 WHERE trip_id = $2',
      [weightKg, tripId],
    );
    if (rowCount !== 1) throw new Error(`no existe el viaje ${tripId}`);
  };
}

// Deja las tres bases como al principio de la demostración.
export async function seedDemo({ sur, oriente, coordinador }) {
  await sur.query('TRUNCATE shipment');
  await sur.query(
    `INSERT INTO shipment (shipment_id, tracking_code, origin, destination, trip_id, weight_kg)
     VALUES (90001, 'CUS-IQT-0001', 'Cusco', 'Iquitos', 7001, 940)`,
  );
  await oriente.query('TRUNCATE trip');
  await oriente.query(
    `INSERT INTO trip (trip_id, route, capacity_kg, booked_kg)
     VALUES (7001, 'Cusco - Iquitos', 1000, 940)`,
  );
  await coordinador.query('TRUNCATE decision');
}

// Pruebas del MapReduce de juguete: el resultado coincide con el GROUP BY, y
// el combiner reduce lo que viaja en el shuffle sin cambiar el resultado.
import { test, after } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import { mapReduce, deliveriesMap, sumReduce, EVENTS_SQL, DELIVERIES_SQL } from './mapreduce.mjs';

const pool = new pg.Pool();
after(() => pool.end());

test('un conteo de palabras pequeño', () => {
  const lines = ['lima cusco lima', 'iquitos lima', 'cusco'];
  const { result, stats } = mapReduce(
    lines,
    (line, emit) => line.split(' ').forEach((w) => emit(w, 1)),
    sumReduce,
    { splits: 2, reducers: 2 },
  );
  assert.deepEqual(Object.fromEntries(result), { lima: 3, cusco: 2, iquitos: 1 });
  assert.equal(stats.mapped, 6);
  assert.equal(stats.shuffled, 6);
});

test('entregas por ciudad: igual que GROUP BY, con y sin combiner', async () => {
  const events = (await pool.query(EVENTS_SQL)).rows;
  assert.equal(events.length, 46421);
  const expected = new Map((await pool.query(DELIVERIES_SQL)).rows.map((r) => [r.city, r.n]));

  const plain = mapReduce(events, deliveriesMap, sumReduce);
  const combined = mapReduce(events, deliveriesMap, sumReduce, { combiner: sumReduce });
  for (const { result } of [plain, combined]) {
    assert.deepEqual(new Map([...result].sort()), new Map([...expected].sort()));
  }
  assert.equal(plain.result.get('Lima'), 3632);
  assert.equal(plain.stats.mapped, 11115);
  assert.equal(plain.stats.shuffled, 11115);
  assert.equal(combined.stats.shuffled, 100);
  // El reducer que recibe Lima recibe mucho más que los otros (skew).
  assert.deepEqual(plain.stats.perReducer, [1373, 7407, 2335]);
});

// Pruebas de pruning.sql y documents.sql sobre chasqui_w16 (make crea las
// tablas particionadas): las cifras y los planes que muestra el apunte.
import { test, before, after } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import { namedQueries } from './sqlfile.mjs';

const q = { ...namedQueries('./pruning.sql'), ...namedQueries('./documents.sql') };
let client;
before(async () => {
  client = new pg.Client();
  await client.connect();
});
after(() => client.end());

const rows = async (name) => (await client.query(q[name])).rows;
const plan = async (name) => (await rows(name)).map((r) => r['QUERY PLAN']).join('\n');
const counts = (list) => Object.fromEntries(list.map((r) => [r.partition, Number(r.n)]));

test('por mes: seis particiones con todos los envíos, la DEFAULT vacía', async () => {
  const byMonth = counts(await rows('by_month'));
  assert.deepEqual(byMonth, {
    shipment_2026_01: 1813, shipment_2026_02: 1938, shipment_2026_03: 2149,
    shipment_2026_04: 2054, shipment_2026_05: 2157, shipment_2026_06: 1889,
  });
  const bounds = await rows('bounds');
  assert.equal(bounds.length, 7);
  assert.equal(bounds.at(-1).bound, 'DEFAULT');
});

test('por hash: cuatro particiones parejas; por zona: Lima concentra un tercio', async () => {
  const byHash = Object.values(counts(await rows('by_hash')));
  assert.equal(byHash.length, 4);
  assert.equal(byHash.reduce((a, b) => a + b), 12000);
  assert.ok(Math.max(...byHash) - Math.min(...byHash) < 150);
  const byZone = await rows('by_zone');
  assert.equal(byZone[0].partition, 'shipment_lima');
  assert.equal(Number(byZone[0].n), 3965);
  assert.equal(byZone[0].pct, '33.0');
});

test('pruning: marzo lee una sola partición; sender_id lee las siete', async () => {
  const march = await plan('prune_march');
  assert.match(march, /Seq Scan on shipment_2026_03/);
  assert.doesNotMatch(march, /Append|shipment_2026_0[124-6]|default/);
  const all = await plan('no_prune');
  assert.equal(all.match(/Seq Scan on/g).length, 7);
});

test('hash: la igualdad poda, el rango no', async () => {
  const eq = await plan('hash_eq');
  assert.equal(eq.match(/ on shipment_h\d/g).length, 1);
  const range = await plan('hash_range');
  assert.equal(range.match(/ on shipment_h\d/g).length, 4);
});

test('pruning en la ejecución: Subplans Removed: 5 y 854 filas', async () => {
  await client.query('SET plan_cache_mode = force_generic_plan');
  await client.query(`PREPARE since_day(timestamptz) AS
    SELECT count(*) FROM shipment_by_month WHERE created_at >= $1`);
  try {
    const p = await plan('prune_runtime');
    assert.match(p, /Subplans Removed: 5/);
    assert.match(p, /Seq Scan on shipment_2026_06 .*actual rows=854/);
  } finally {
    await client.query('DEALLOCATE since_day');
    await client.query('RESET plan_cache_mode');
  }
  assert.equal(Number((await rows('since_june15'))[0].count), 854);
});

test('join por particiones: cuatro hash joins, partición contra partición', async () => {
  await client.query('SET enable_partitionwise_join = on');
  try {
    const p = await plan('partitionwise');
    assert.equal(p.match(/Hash Join/g).length, 4);
    for (const i of [0, 1, 2, 3]) assert.match(p, new RegExp(`shipment_event_h${i}`));
  } finally {
    await client.query('RESET enable_partitionwise_join');
  }
  const total = (await rows('events_per_status')).reduce((a, r) => a + Number(r.count), 0);
  assert.equal(total, 46421);
});

test('execution skew: los 50 envíos nuevos caen en junio', async () => {
  await client.query('BEGIN');
  try {
    assert.deepEqual(counts(await rows('insert_skew')), { shipment_2026_06: 50 });
  } finally {
    await client.query('ROLLBACK');
  }
});

test('sin partición para la agencia 30: error 23514', async () => {
  await assert.rejects(client.query(q.no_partition), (err) => {
    assert.equal(err.code, '23514');
    assert.match(err.message, /no partition of relation "shipment_by_zone" found for row/);
    return true;
  });
});

test('el envío 501 como documento, y cuántas copias tiene el nombre de la agencia 1', async () => {
  const doc = JSON.parse((await rows('document_501'))[0].doc);
  assert.equal(doc.events.length, 4);
  assert.equal(doc.events[0].agency, 'Huánuco - Centro');
  assert.deepEqual(await rows('docs_to_rewrite'), [{ documents: '3531', copies: '7002' }]);
});

// Pruebas de streaming replication en el cluster desechable: primario en el
// puerto 5416 y réplica (hot standby) en el 5417, creada con pg_basebackup.
// Al final la réplica queda detenida; make test borra el cluster.
import { test, after } from 'node:test';
import assert from 'node:assert/strict';
import { execFileSync } from 'node:child_process';
import { setTimeout as sleep } from 'node:timers/promises';
import pg from 'pg';

const primary = new pg.Pool({ host: '/tmp', port: 5416, user: 'postgres', database: 'postgres' });
const replica = new pg.Pool({ host: '/tmp', port: 5417, user: 'postgres', database: 'postgres' });
// Vuelve a la replicación asíncrona: con la réplica detenida, cualquier
// commit en el primario esperaría para siempre.
async function asyncAgain() {
  await primary.query('ALTER SYSTEM RESET synchronous_standby_names');
  await primary.query('SELECT pg_reload_conf()');
}
after(async () => {
  await asyncAgain();
  await primary.query('DROP TABLE IF EXISTS replica_demo');
  await primary.end();
  await replica.end().catch(() => {});
});

// Espera hasta que check() sea verdadero (la réplica va un poco atrás).
async function eventually(check, ms = 5000) {
  const until = Date.now() + ms;
  while (Date.now() < until) {
    if (await check()) return;
    await sleep(50);
  }
  assert.fail('no ocurrió a tiempo');
}

test('el primario ve a replica1 en streaming asíncrono', async () => {
  const { rows } = await primary.query(
    'SELECT application_name, state, sync_state FROM pg_stat_replication');
  assert.deepEqual(rows, [{ application_name: 'replica1', state: 'streaming', sync_state: 'async' }]);
});

test('la réplica está en recovery y no acepta escrituras (25006)', async () => {
  const { rows } = await replica.query('SELECT pg_is_in_recovery() AS r');
  assert.equal(rows[0].r, true);
  await assert.rejects(replica.query('CREATE TABLE x (a integer)'), (err) => {
    assert.equal(err.code, '25006');
    assert.equal(err.message, 'cannot execute CREATE TABLE in a read-only transaction');
    return true;
  });
});

test('lo que se escribe en el primario llega a la réplica; también un DELETE', async () => {
  await primary.query('CREATE TABLE replica_demo (id integer PRIMARY KEY, note text)');
  await primary.query("INSERT INTO replica_demo VALUES (1, 'envío 501'), (2, 'envío 502')");
  const count = async () => {
    const r = await replica.query("SELECT to_regclass('replica_demo') IS NOT NULL AS t");
    return r.rows[0].t ? Number((await replica.query('SELECT count(*) FROM replica_demo')).rows[0].count) : -1;
  };
  await eventually(async () => (await count()) === 2);
  // Una réplica no es un backup: el DELETE equivocado también se replica.
  await primary.query('DELETE FROM replica_demo');
  await eventually(async () => (await count()) === 0);
});

test('síncrona: sin la réplica, el commit espera; al cancelarlo ya hizo commit local', async () => {
  await primary.query("ALTER SYSTEM SET synchronous_standby_names = 'replica1'");
  await primary.query('SELECT pg_reload_conf()');
  await eventually(async () =>
    (await primary.query('SELECT sync_state FROM pg_stat_replication')).rows[0]?.sync_state === 'sync');

  // Con la réplica activa, un commit espera su confirmación y vuelve enseguida.
  await primary.query("INSERT INTO replica_demo VALUES (3, 'con réplica')");

  // Se detiene la réplica: el próximo commit queda esperando (SyncRep).
  await replica.end();
  const args = ['-D', '/tmp/w16-replica', '-m', 'fast', '-w', '-s', 'stop'];
  if (process.getuid() === 0) execFileSync('gosu', ['postgres', 'pg_ctl', ...args]);
  else execFileSync('pg_ctl', args);

  const client = new pg.Client({ host: '/tmp', port: 5416, user: 'postgres', database: 'postgres' });
  await client.connect();
  const notices = [];
  client.on('notice', (n) => notices.push(n.message));
  const insert = client.query("INSERT INTO replica_demo VALUES (4, 'sin réplica')");
  let waiting;
  await eventually(async () => {
    waiting = (await primary.query(
      "SELECT pid FROM pg_stat_activity WHERE wait_event = 'SyncRep'")).rows;
    return waiting.length === 1;
  });
  await primary.query('SELECT pg_cancel_backend($1)', [waiting[0].pid]);
  await insert; // no falla: el commit ya estaba hecho en el primario
  await client.end();
  await asyncAgain();
  assert.deepEqual(notices, ['canceling wait for synchronous replication due to user request']);
  const { rows } = await primary.query('SELECT id FROM replica_demo ORDER BY id');
  assert.deepEqual(rows.map((r) => r.id), [3, 4]);
});

// Pruebas del coordinador de 2PC y de prepared.sql, sobre el cluster
// desechable (make test lo crea y lo borra al final).
import { test, beforeEach, after } from 'node:test';
import assert from 'node:assert/strict';
import { execFileSync } from 'node:child_process';
import pg from 'pg';
import { Coordinator, CoordinatorCrash, bookShipment, seedDemo } from './twopc.mjs';
import { namedQueries } from './sqlfile.mjs';

const PRIMARY = '/tmp/w16-primary';
const conn = (database) => new pg.Pool({ host: '/tmp', port: 5416, user: 'postgres', database });
const pools = { sur: conn('sur'), oriente: conn('oriente'), coordinador: conn('coordinador') };
const steps = [];
const coordinator = new Coordinator(
  pools.coordinador,
  { sur: pools.sur, oriente: pools.oriente },
  { trace: (s) => steps.push(s) },
);
after(() => Promise.all(Object.values(pools).map((p) => p.end())));

beforeEach(async () => {
  steps.length = 0;
  await seedDemo(pools);
});

const booked = async () =>
  Number((await pools.oriente.query('SELECT booked_kg FROM trip WHERE trip_id = 7001')).rows[0].booked_kg);
const shipments = async () =>
  (await pools.sur.query('SELECT shipment_id FROM shipment ORDER BY 1')).rows.map((r) => r.shipment_id);
const booking = (id, weightKg) =>
  bookShipment({ shipmentId: id, trackingCode: `CUS-IQT-${id}`, tripId: 7001, weightKg });

test('commit: los dos votan que sí y los dos hacen commit', async () => {
  assert.equal(await coordinator.run('chasqui-1', booking(90002, 35)), 'commit');
  assert.deepEqual(steps, [
    'sur: ready', 'oriente: ready', 'decisión: commit (en el log del coordinador)',
    'sur: commit prepared', 'oriente: commit prepared', 'end',
  ]);
  assert.equal(await booked(), 975);
  assert.deepEqual(await shipments(), [90001, 90002]);
});

test('voto no en la fase 1: oriente no tiene capacidad y sur hace rollback prepared', async () => {
  assert.equal(await coordinator.run('chasqui-2', booking(90003, 61)), 'abort');
  assert.match(steps[1], /^oriente: no \(el viaje 7001 no tiene capacidad: 1001.00 kg de 1000\)$/);
  assert.equal(steps[2], 'sur: rollback prepared');
  assert.equal(await booked(), 940);
  assert.deepEqual(await shipments(), [90001]);
  const { rows } = await pools.coordinador.query('SELECT count(*)::int AS n FROM decision');
  assert.equal(rows[0].n, 0); // presumed abort: el abort no se anota
});

test('fallo antes de la fase 1: el código de seguimiento repetido aborta todo', async () => {
  const work = bookShipment({ shipmentId: 90004, trackingCode: 'CUS-IQT-0001', tripId: 7001, weightKg: 5 });
  assert.equal(await coordinator.run('chasqui-3', work), 'abort');
  assert.match(steps[0], /duplicate key value violates unique constraint "shipment_tracking_code_key"/);
  assert.equal(await booked(), 940);
});

test('el coordinador cae después de la fase 1: bloqueo y presumed abort', async () => {
  await assert.rejects(coordinator.run('chasqui-4', booking(90005, 10), { crashAt: 'prepared' }), CoordinatorCrash);
  assert.deepEqual(await coordinator.inDoubt(), { sur: ['chasqui-4'], oriente: ['chasqui-4'] });

  // Los participantes no pueden decidir solos y conservan sus locks: otra
  // transacción que quiere el mismo viaje espera (aquí, hasta el lock_timeout).
  const other = await pools.oriente.connect();
  try {
    await other.query("SET lock_timeout = '300ms'");
    await assert.rejects(other.query('UPDATE trip SET booked_kg = booked_kg WHERE trip_id = 7001'),
      (err) => err.code === '55P03');
  } finally {
    other.release();
  }
  // Mientras tanto, el cambio no se ve.
  assert.equal(await booked(), 940);

  assert.deepEqual(await coordinator.recover(), { 'chasqui-4': 'abort' });
  assert.deepEqual(await coordinator.inDoubt(), { sur: [], oriente: [] });
  assert.equal(await booked(), 940);
  assert.deepEqual(await shipments(), [90001]);
});

test('el coordinador cae después de decidir: la recuperación hace commit', async () => {
  await assert.rejects(coordinator.run('chasqui-5', booking(90006, 10), { crashAt: 'decided' }), CoordinatorCrash);
  assert.equal(await booked(), 940); // decidido, pero todavía sin fase 2
  assert.deepEqual(await coordinator.recover(), { 'chasqui-5': 'commit' });
  assert.equal(await booked(), 950);
  assert.deepEqual(await shipments(), [90001, 90006]);
  const { rows } = await pools.coordinador.query('SELECT gid, ended FROM decision');
  assert.deepEqual(rows, [{ gid: 'chasqui-5', ended: true }]);
});

test('gid inválido: no se arma SQL con él', async () => {
  await assert.rejects(coordinator.run("chasqui-x'; DROP TABLE trip; --", booking(90007, 1)), /gid inválido/);
  await assert.rejects(coordinator.run('otro-1', booking(90007, 1)), /gid inválido/);
});

test('prepared.sql: la transacción preparada no tiene proceso y conserva sus locks', async () => {
  const q = namedQueries('./prepared.sql');
  const c = await pools.oriente.connect();
  try {
    await c.query('BEGIN');
    await c.query('UPDATE trip SET booked_kg = booked_kg + 5 WHERE trip_id = 7001');
    await c.query("PREPARE TRANSACTION 'manual-1@oriente'");
    const [xact] = (await c.query(q.prepared_xacts)).rows;
    assert.equal(xact.gid, 'manual-1@oriente');
    const locks = (await c.query(q.prepared_locks)).rows;
    assert.deepEqual(locks.map((l) => [l.locktype, l.relation, l.mode, l.pid]), [
      ['relation', 'trip', 'RowExclusiveLock', null],
      ['relation', 'trip_pkey', 'RowExclusiveLock', null],
      ['transactionid', null, 'ExclusiveLock', null],
    ]);
    assert.equal((await c.query(q.before_commit)).rows[0].booked_kg, '940.00');
    await c.query("COMMIT PREPARED 'manual-1@oriente'");
    assert.equal((await c.query(q.after_commit)).rows[0].booked_kg, '945.00');
  } finally {
    c.release();
  }
});

test('una transacción preparada sobrevive a una caída del servidor', async () => {
  const c = await pools.oriente.connect();
  await c.query('BEGIN');
  await c.query('UPDATE trip SET booked_kg = booked_kg + 5 WHERE trip_id = 7001');
  await c.query("PREPARE TRANSACTION 'caida-1@oriente'");
  c.release();
  // Caída del primario del cluster desechable (nunca del servidor del curso).
  const pgctl = (...args) => process.getuid() === 0
    ? execFileSync('gosu', ['postgres', 'pg_ctl', '-D', PRIMARY, '-w', '-s', ...args])
    : execFileSync('pg_ctl', ['-D', PRIMARY, '-w', '-s', ...args]);
  await Promise.all(Object.values(pools).map((p) => p.end()));
  pgctl('-m', 'immediate', 'stop');
  pgctl('-l', `${PRIMARY}.log`, '-o',
    '-p 5416 -k /tmp -c listen_addresses= -c max_prepared_transactions=10 -c cluster_name=primary', 'start');
  const oriente = conn('oriente');
  try {
    const { rows } = await oriente.query('SELECT gid FROM pg_prepared_xacts');
    assert.deepEqual(rows, [{ gid: 'caida-1@oriente' }]);
    await oriente.query("ROLLBACK PREPARED 'caida-1@oriente'");
    assert.equal((await oriente.query('SELECT booked_kg FROM trip')).rows[0].booked_kg, '940.00');
  } finally {
    await oriente.end();
  }
  for (const name of Object.keys(pools)) pools[name] = conn(name);
});