-- Semana 13: transacciones en SQL con una sola sesión. Todo lo que cambia
-- datos termina en ROLLBACK, así que se puede correr varias veces:
--   psql -d chasqui_w13 -f examples.sql
-- Los errores son parte del ejemplo: no usen ON_ERROR_STOP.

SHOW transaction_isolation;

BEGIN ISOLATION LEVEL READ UNCOMMITTED;
SHOW transaction_isolation;
COMMIT;

-- SET TRANSACTION tiene que ser lo primero de la transacción.
BEGIN;
SELECT price FROM shipment WHERE shipment_id = 501;
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
ROLLBACK;

-- Después de un error, la transacción queda abortada: todo se ignora hasta
-- el ROLLBACK (o el COMMIT, que también hace rollback).
BEGIN;
UPDATE shipment SET price = price + 5 WHERE shipment_id = 501;
UPDATE shipment SET price = -1 WHERE shipment_id = 502;
SELECT price FROM shipment WHERE shipment_id = 501;
COMMIT;
SELECT price FROM shipment WHERE shipment_id = 501;

-- Con un savepoint se deshace solo la parte que falló.
BEGIN;
UPDATE shipment SET price = price + 5 WHERE shipment_id = 501;
SAVEPOINT before_discount;
UPDATE shipment SET price = -1 WHERE shipment_id = 502;
ROLLBACK TO SAVEPOINT before_discount;
SELECT shipment_id, price FROM shipment WHERE shipment_id IN (501, 502) ORDER BY shipment_id;
ROLLBACK;

// Muestra los conflictos, el grafo de precedencia y el veredicto de un
// schedule. Si el schedule tiene commits o aborts, dice además si es
// recuperable y si es cascadeless.
//   node graph.mjs "r1(A) r2(B) w1(A) r3(A) w2(B) r3(B) w3(A) r1(B)"
import {
  parseSchedule,
  conflicts,
  conflictSerializable,
  serialSchedule,
  isRecoverable,
  isCascadeless,
  show,
} from './precedence.mjs';

const text = process.argv.slice(2).join(' ');
if (text.trim() === '') {
  console.error('uso: node graph.mjs "r1(A) w2(A) ..."');
  process.exit(2);
}

const ops = parseSchedule(text);
console.log(`schedule: ${ops.map(show).join(' ')}`);

console.log('\nconflictos:');
for (const [p, q] of conflicts(ops)) console.log(`  ${show(p)}–${show(q)}  T${p.txn} → T${q.txn}`);

const result = conflictSerializable(ops);
console.log('\naristas:');
for (const e of result.graph.edges) console.log(`  T${e.from} → T${e.to}  (${e.because.join(', ')})`);

if (result.serializable) {
  const order = result.order.map((t) => `T${t}`).join(', ');
  console.log(`\nconflict serializable; orden serial equivalente: ${order}`);
  console.log(`  ${serialSchedule(ops, result.order)}`);
} else {
  console.log(`\nno es conflict serializable; ciclo: ${result.cycle.map((t) => `T${t}`).join(' → ')}`);
}

if (ops.some((o) => o.op === 'c' || o.op === 'a')) {
  const rec = isRecoverable(ops);
  const cas = isCascadeless(ops);
  const why = (v) => `T${v.reader} lee ${v.item} de T${v.writer} (${show(v.write)} … ${show(v.read)})`;
  console.log(`\nrecuperable: ${rec.recoverable ? 'sí' : `no; ${why(rec.violation)} y hace commit antes que T${rec.violation.writer}`}`);
  console.log(`cascadeless: ${cas.cascadeless ? 'sí' : `no; ${why(cas.violation)} antes del commit de T${cas.violation.writer}`}`);
}

// Schedules, conflictos y grafo de precedencia (Silberschatz, secs. 17.6 y
// 17.7). Un schedule se escribe como en clase:
//
//   r1(A) w2(A) r2(B) w1(B) c1 c2
//
// r = read, w = write, c = commit, a = abort; el número es la transacción y
// la letra entre paréntesis, el ítem. Las comas y los espacios extra se
// ignoran. El módulo no ejecuta nada: solo mira el orden de las operaciones,
// igual que el control de concurrencia del DBMS, que no ve los valores.

const TOKEN = /^([rwca])(\d+)(?:\(([A-Za-z_][A-Za-z0-9_]*)\))?$/;

// "r1(A) w2(B) c1" -> [{ op: 'r', txn: 1, item: 'A', pos: 0 }, ...]
export function parseSchedule(text) {
  const words = text.split(/[\s,;]+/).filter((w) => w !== '');
  return words.map((word, pos) => {
    const m = TOKEN.exec(word);
    if (!m) throw new Error(`operación no válida: "${word}" (se espera r1(A), w2(B), c1 o a1)`);
    const [, op, txn, item] = m;
    const isAccess = op === 'r' || op === 'w';
    if (isAccess && item === undefined) throw new Error(`a "${word}" le falta el ítem: ${op}${txn}(A)`);
    if (!isAccess && item !== undefined) throw new Error(`"${word}": commit y abort no llevan ítem`);
    return { op, txn: Number(txn), item: isAccess ? item : null, pos };
  });
}

// Texto de una operación, para mensajes y para los motivos de las aristas.
export function show(o) {
  return o.item === null ? `${o.op}${o.txn}` : `${o.op}${o.txn}(${o.item})`;
}

function ops(schedule) {
  return typeof schedule === 'string' ? parseSchedule(schedule) : schedule;
}

// Las transacciones que aparecen en el schedule, ordenadas por número.
export function transactions(schedule) {
  return [...new Set(ops(schedule).map((o) => o.txn))].sort((a, b) => a - b);
}

// Dos operaciones están en conflicto si son de transacciones distintas, sobre
// el mismo ítem, y al menos una es una escritura. Devuelve los pares [p, q]
// con p antes que q en el schedule.
export function conflicts(schedule) {
  const list = ops(schedule).filter((o) => o.op === 'r' || o.op === 'w');
  const pairs = [];
  for (let i = 0; i < list.length; i++) {
    for (let j = i + 1; j < list.length; j++) {
      const p = list[i];
      const q = list[j];
      if (p.txn !== q.txn && p.item === q.item && (p.op === 'w' || q.op === 'w')) pairs.push([p, q]);
    }
  }
  return pairs;
}

// Grafo de precedencia: un nodo por transacción y una arista Ti -> Tj por
// cada par en conflicto en que la operación de Ti va primero. Cada arista
// guarda los pares que la justifican ("w1(A)–r3(A)").
export function precedenceGraph(schedule) {
  const all = ops(schedule);
  const nodes = transactions(all);
  const byKey = new Map();
  for (const [p, q] of conflicts(all)) {
    const key = `${p.txn}->${q.txn}`;
    if (!byKey.has(key)) byKey.set(key, { from: p.txn, to: q.txn, because: [] });
    byKey.get(key).because.push(`${show(p)}–${show(q)}`);
  }
  const edges = [...byKey.values()].sort((a, b) => a.from - b.from || a.to - b.to);
  return { nodes, edges };
}

// Busca un ciclo con una búsqueda en profundidad. Devuelve la lista de
// transacciones del ciclo, cerrada (la primera se repite al final), o null.
export function findCycle({ nodes, edges }) {
  const next = new Map(nodes.map((n) => [n, []]));
  for (const e of edges) next.get(e.from).push(e.to);
  const state = new Map(nodes.map((n) => [n, 'new'])); // new, open, done
  const path = [];

  function visit(n) {
    state.set(n, 'open');
    path.push(n);
    for (const m of next.get(n)) {
      if (state.get(m) === 'open') return [...path.slice(path.indexOf(m)), m];
      if (state.get(m) === 'new') {
        const cycle = visit(m);
        if (cycle) return cycle;
      }
    }
    path.pop();
    state.set(n, 'done');
    return null;
  }

  for (const n of nodes) {
    if (state.get(n) === 'new') {
      const cycle = visit(n);
      if (cycle) return cycle;
    }
  }
  return null;
}

// Orden topológico (Kahn). Si hay varias transacciones sin aristas de
// entrada, toma la de número menor, así el resultado es siempre el mismo.
// Devuelve null si el grafo tiene un ciclo.
export function topologicalOrder({ nodes, edges }) {
  const indegree = new Map(nodes.map((n) => [n, 0]));
  for (const e of edges) indegree.set(e.to, indegree.get(e.to) + 1);
  const ready = nodes.filter((n) => indegree.get(n) === 0);
  const order = [];
  while (ready.length > 0) {
    ready.sort((a, b) => a - b);
    const n = ready.shift();
    order.push(n);
    for (const e of edges) {
      if (e.from !== n) continue;
      indegree.set(e.to, indegree.get(e.to) - 1);
      if (indegree.get(e.to) === 0) ready.push(e.to);
    }
  }
  return order.length === nodes.length ? order : null;
}

// La prueba de conflict serializability: el schedule es conflict serializable
// si y solo si su grafo de precedencia no tiene ciclos.
//   { serializable: true,  order: [2, 1, 3], graph }
//   { serializable: false, cycle: [1, 2, 1], graph }
export function conflictSerializable(schedule) {
  const graph = precedenceGraph(schedule);
  const cycle = findCycle(graph);
  if (cycle) return { serializable: false, cycle, graph };
  return { serializable: true, order: topologicalOrder(graph), graph };
}

// El schedule serial de un orden: todas las operaciones de la primera
// transacción, después las de la segunda, etc. Sirve para comparar.
export function serialSchedule(schedule, order) {
  const all = ops(schedule);
  return order.flatMap((t) => all.filter((o) => o.txn === t)).map(show).join(' ');
}

// Relación "lee de" (reads-from): Tj lee X de Ti si la última escritura de X
// antes de esa lectura es de Ti (otra transacción) y Ti no abortó antes de
// la lectura. Devuelve [{ reader, writer, item, read, write }].
export function readsFrom(schedule) {
  const all = ops(schedule);
  const aborted = new Set();
  const result = [];
  for (let i = 0; i < all.length; i++) {
    const o = all[i];
    if (o.op === 'a') aborted.add(o.txn);
    if (o.op !== 'r') continue;
    for (let k = i - 1; k >= 0; k--) {
      const w = all[k];
      if (w.op !== 'w' || w.item !== o.item || aborted.has(w.txn)) continue;
      if (w.txn !== o.txn) result.push({ reader: o.txn, writer: w.txn, item: o.item, read: o, write: w });
      break;
    }
  }
  return result;
}

function positionOf(all, op, txn) {
  const found = all.find((o) => o.op === op && o.txn === txn);
  return found ? found.pos : Infinity;
}

// Recuperable: si Tj lee algo que escribió Ti, Ti hace commit antes que Tj.
// Si Tj no hace commit en el schedule, todavía no hay nada que objetar.
//   { recoverable: true } o { recoverable: false, violation: {...} }
export function isRecoverable(schedule) {
  const all = ops(schedule);
  for (const rf of readsFrom(all)) {
    const readerCommit = positionOf(all, 'c', rf.reader);
    if (readerCommit === Infinity) continue;
    if (positionOf(all, 'c', rf.writer) > readerCommit) return { recoverable: false, violation: rf };
  }
  return { recoverable: true };
}

// Cascadeless: cada lectura de un dato escrito por otra transacción ocurre
// después del commit de esa transacción. Así un abort nunca arrastra a otra.
export function isCascadeless(schedule) {
  const all = ops(schedule);
  for (const rf of readsFrom(all)) {
    if (positionOf(all, 'c', rf.writer) > rf.read.pos) return { cascadeless: false, violation: rf };
  }
  return { cascadeless: true };
}

// Corre un escenario de scenarios.mjs con dos procesos psql y muestra lo que
// se vería en las dos terminales, en orden: A=# y B=# son los prompts de
// cada sesión. Cuando una sentencia espera un lock se marca con "-- B espera" (o A),
// y su salida aparece después, en el paso en que pudo continuar.
//   PGDATABASE=chasqui_w13 node psql2.mjs lost-update rc
// Niveles: ru, rc, rr, sr. Escenarios: los de scenarios.mjs. Con --verbose,
// psql corre con VERBOSITY=verbose y los errores muestran su SQLSTATE.
import { LEVELS, SCENARIOS } from './scenarios.mjs';
import { runScenario } from './sessions.mjs';

const args = process.argv.slice(2);
const verbose = args.includes('--verbose');
const [name, level = 'rc'] = args.filter((a) => a !== '--verbose');
if (!SCENARIOS[name] || !LEVELS[level]) {
  console.error('uso: node psql2.mjs <escenario> [ru|rc|rr|sr] [--verbose]');
  console.error(`escenarios: ${Object.keys(SCENARIOS).join(', ')}`);
  process.exit(2);
}

const { steps } = await runScenario(name, level, {
  kind: 'psql',
  psqlFlags: verbose ? ['-v', 'VERBOSITY=verbose'] : [],
});
const lines = [];
const print = (step) => {
  if (step.output !== '') lines.push(step.output);
};
steps.forEach((step, i) => {
  lines.push(`${step.session}=# ${step.sql}`);
  if (step.blocked) lines.push(`-- ${step.session} espera`);
  else print(step);
  for (const resumed of steps.filter((s) => s.resumedAt === i)) {
    lines.push(`-- ${resumed.session} continúa`);
    print(resumed);
  }
});
console.log(lines.join('\n'));

// Los schedules de dos sesiones del capítulo. Cada escenario es una lista de
// pasos [sesión, sentencia, etiqueta]; la etiqueta (opcional) sirve para que
// las pruebas encuentren el resultado de ese paso. sessions.mjs los ejecuta
// con dos clientes de pg; psql2.mjs, con dos procesos psql.
//
// Todos trabajan sobre los envíos 501 y 502 y el viaje 5708 de chasqui_w13.
// Antes de empezar, prepare() deja los precios en 50.00 y 45.00 para que las
// cuentas sean fáciles; restore() los devuelve a como estaban.

export const LEVELS = {
  ru: 'READ UNCOMMITTED',
  rc: 'READ COMMITTED',
  rr: 'REPEATABLE READ',
  sr: 'SERIALIZABLE',
};

const PRICE_501 = 'SELECT price FROM shipment WHERE shipment_id = 501;';
const SUM_501_502 = 'SELECT sum(price) FROM shipment WHERE shipment_id IN (501, 502);';
const TRIP_5708 = 'SELECT count(*), sum(weight_kg) FROM shipment WHERE trip_id = 5708;';
const NEW_SHIPMENT =
  'INSERT INTO shipment (shipment_id, sender_id, recipient_id, origin_agency_id, ' +
  'dest_agency_id, trip_id, weight_kg, price, status, created_at) ' +
  "VALUES (12001, 1078, 96, 25, 27, 5708, 10, 30, 'registered', now());";

const begin = (level) => `BEGIN ISOLATION LEVEL ${LEVELS[level]};`;

export const SCENARIOS = {
  // B cambia el precio y no hace commit; A intenta leer el cambio.
  'dirty-read': (level) => [
    ['B', 'BEGIN;'],
    ['B', 'UPDATE shipment SET price = 0 WHERE shipment_id = 501;'],
    ['A', begin(level)],
    ['A', PRICE_501, 'read'],
    ['B', 'ROLLBACK;'],
    ['A', 'COMMIT;'],
  ],

  // A lee dos veces la misma fila; entre las dos lecturas, B la cambia.
  'non-repeatable-read': (level) => [
    ['A', begin(level)],
    ['A', PRICE_501, 'first'],
    ['B', 'UPDATE shipment SET price = price + 10 WHERE shipment_id = 501;'],
    ['A', PRICE_501, 'second'],
    ['A', 'COMMIT;'],
  ],

  // En PostgreSQL el snapshot de REPEATABLE READ se toma en la primera
  // consulta de la transacción, no en el BEGIN.
  'snapshot-start': (level) => [
    ['A', begin(level)],
    ['B', 'UPDATE shipment SET price = price + 10 WHERE shipment_id = 501;'],
    ['A', PRICE_501, 'first'],
    ['B', 'UPDATE shipment SET price = price + 10 WHERE shipment_id = 501;'],
    ['A', PRICE_501, 'second'],
    ['A', 'COMMIT;'],
  ],

  // A repite una consulta con condición; entre las dos, B inserta una fila
  // que la cumple.
  phantom: (level) => [
    ['A', begin(level)],
    ['A', TRIP_5708, 'first'],
    ['B', NEW_SHIPMENT],
    ['A', TRIP_5708, 'second'],
    ['A', 'COMMIT;'],
  ],

  // Read-then-write: las dos leen 50.00 y escriben un valor calculado en la
  // aplicación. A cobra 5 soles de embalaje; B aplica 10 de descuento.
  'lost-update': (level) => [
    ['A', begin(level)],
    ['B', begin(level)],
    ['A', PRICE_501, 'aRead'],
    ['B', PRICE_501, 'bRead'],
    ['A', 'UPDATE shipment SET price = 55 WHERE shipment_id = 501;'],
    ['B', 'UPDATE shipment SET price = 40 WHERE shipment_id = 501;', 'bWrite'],
    ['A', 'COMMIT;'],
    ['B', 'COMMIT;', 'bCommit'],
    ['A', PRICE_501, 'final'],
  ],

  // Lo mismo, pero cada una escribe con una sola sentencia que lee y
  // escribe la fila (price = price + 5).
  'atomic-update': (level) => [
    ['A', begin(level)],
    ['B', begin(level)],
    ['A', 'UPDATE shipment SET price = price + 5 WHERE shipment_id = 501;'],
    ['B', 'UPDATE shipment SET price = price - 10 WHERE shipment_id = 501;', 'bWrite'],
    ['A', 'COMMIT;'],
    ['B', 'COMMIT;', 'bCommit'],
    ['A', PRICE_501, 'final'],
  ],

  // Read-then-write con SELECT ... FOR UPDATE: la lectura ya toma el lock.
  'for-update': (level) => [
    ['A', begin(level)],
    ['B', begin(level)],
    ['A', 'SELECT price FROM shipment WHERE shipment_id = 501 FOR UPDATE;', 'aRead'],
    ['B', 'SELECT price FROM shipment WHERE shipment_id = 501 FOR UPDATE;', 'bRead'],
    ['A', 'UPDATE shipment SET price = 55 WHERE shipment_id = 501;'],
    ['A', 'COMMIT;'],
    ['B', 'UPDATE shipment SET price = 45 WHERE shipment_id = 501;'],
    ['B', 'COMMIT;', 'bCommit'],
    ['A', PRICE_501, 'final'],
  ],

  // Write skew. Regla de la promoción: entre los envíos 501 y 502 el cliente
  // paga al menos 90 soles. Cada empleado revisa la suma (95) y descuenta 5
  // soles a un envío distinto. Cada uno, solo, respeta la regla.
  'write-skew': (level) => [
    ['A', begin(level)],
    ['B', begin(level)],
    ['A', SUM_501_502, 'aSum'],
    ['B', SUM_501_502, 'bSum'],
    ['A', 'UPDATE shipment SET price = price - 5 WHERE shipment_id = 501;'],
    ['B', 'UPDATE shipment SET price = price - 5 WHERE shipment_id = 502;', 'bWrite'],
    ['A', 'COMMIT;', 'aCommit'],
    ['B', 'COMMIT;', 'bCommit'],
    ['A', SUM_501_502, 'final'],
  ],

  // El lost update en SERIALIZABLE, y B reintenta la transacción completa:
  // ROLLBACK, BEGIN otra vez, vuelve a leer y vuelve a calcular.
  retry: () => [
    ['A', begin('sr')],
    ['B', begin('sr')],
    ['A', PRICE_501, 'aRead'],
    ['B', PRICE_501, 'bRead'],
    ['A', 'UPDATE shipment SET price = 55 WHERE shipment_id = 501;'],
    ['B', 'UPDATE shipment SET price = 40 WHERE shipment_id = 501;', 'bWrite'],
    ['A', 'COMMIT;'],
    ['B', 'ROLLBACK;'],
    ['B', begin('sr')],
    ['B', PRICE_501, 'bReread'],
    ['B', 'UPDATE shipment SET price = 45 WHERE shipment_id = 501;'],
    ['B', 'COMMIT;', 'bCommit'],
    ['A', PRICE_501, 'final'],
  ],
};

// Deja los datos en el estado de partida y devuelve lo necesario para
// restaurarlos. Se corre con un cliente aparte, fuera de las dos sesiones.
export async function prepare(client) {
  const { rows } = await client.query(
    'SELECT shipment_id, price FROM shipment WHERE shipment_id IN (501, 502) ORDER BY shipment_id',
  );
  await client.query('DELETE FROM shipment WHERE shipment_id = 12001');
  await client.query('UPDATE shipment SET price = 50 WHERE shipment_id = 501');
  await client.query('UPDATE shipment SET price = 45 WHERE shipment_id = 502');
  return rows;
}

export async function restore(client, saved) {
  await client.query('DELETE FROM shipment WHERE shipment_id = 12001');
  for (const { shipment_id: id, price } of saved) {
    await client.query('UPDATE shipment SET price = $1 WHERE shipment_id = $2', [price, id]);
  }
}

// Ejecuta un escenario de scenarios.mjs con dos sesiones, paso a paso, como
// se haría a mano con dos terminales. Las sesiones pueden ser dos clientes
// de pg (para las pruebas) o dos procesos psql (para mostrar la salida real
// de psql, con psql2.mjs).
//
// Después de mandar cada sentencia se espera a que termine o a que quede
// esperando un lock. Para saber si espera, una tercera conexión (monitor)
// pregunta a PostgreSQL con pg_blocking_pids(pid), que lista los procesos
// que la bloquean. Si espera, se pasa al paso siguiente y la sentencia
// termina más tarde, cuando la otra sesión suelte el lock.
import { spawn } from 'node:child_process';
import pg from 'pg';
import { SCENARIOS, prepare, restore } from './scenarios.mjs';

const POLL_MS = 10;
const TIMEOUT_MS = 10000;
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// Una sesión con un cliente de pg. Cada paso guarda rows, command o error.
class PgSession {
  constructor(name) {
    this.client = new pg.Client({ application_name: `w13_${name}` });
  }

  async open() {
    await this.client.connect();
    this.pid = this.client.processID;
  }

  send(step) {
    this.client.query(step.sql).then(
      (r) => Object.assign(step, { command: r.command, rowCount: r.rowCount, rows: r.rows }),
      (e) => Object.assign(step, { error: { code: e.code, message: e.message } }),
    ).finally(() => {
      step.done = true;
    });
  }

  async close() {
    await this.client.end();
  }
}

// Una sesión con psql. Después de cada sentencia se manda \echo con una
// marca; cuando la marca aparece en la salida, la sentencia terminó y todo
// lo anterior es su salida (stdout y stderr, donde psql escribe los errores).
class PsqlSession {
  constructor(name, monitor, flags = []) {
    this.flags = flags;
    this.name = `w13_${name}_${process.pid}`;
    this.monitor = monitor;
    this.out = '';
    this.err = '';
    this.queue = [];
  }

  async open() {
    this.proc = spawn('psql', ['-X', ...this.flags], { env: { ...process.env, PGAPPNAME: this.name } });
    this.proc.stdout.on('data', (d) => {
      this.out += d;
      this.collect();
    });
    this.proc.stderr.on('data', (d) => {
      this.err += d;
    });
    // El pid del backend sale de pg_stat_activity, por application_name.
    for (let waited = 0; this.pid === undefined; waited += POLL_MS) {
      if (waited > TIMEOUT_MS) throw new Error('psql no se conectó');
      const { rows } = await this.monitor.query(
        'SELECT pid FROM pg_stat_activity WHERE application_name = $1',
        [this.name],
      );
      if (rows.length > 0) this.pid = rows[0].pid;
      else await sleep(POLL_MS);
    }
  }

  send(step) {
    step.marker = `@@done-${Math.random().toString(36).slice(2)}@@`;
    this.queue.push(step);
    this.proc.stdin.write(`${step.sql}\n\\echo ${step.marker}\n`);
  }

  collect() {
    for (const step of this.queue) {
      const at = this.out.indexOf(`${step.marker}\n`);
      if (at < 0) return;
      step.stdout = this.out.slice(0, at);
      this.out = this.out.slice(at + step.marker.length + 1);
      // stderr llega por otro canal: se le da un momento antes de leerlo.
      setTimeout(() => {
        step.output = (step.stdout + this.err).replace(/\n+$/, '');
        this.err = '';
        step.done = true;
      }, 30);
      this.queue = this.queue.filter((s) => s !== step);
    }
  }

  async close() {
    this.proc.stdin.end();
    await new Promise((resolve) => this.proc.on('close', resolve));
  }
}

// Espera hasta que cada sentencia en curso haya terminado o esté esperando
// un lock. Si una que esperaba termina, anota en qué paso continuó.
async function settle(monitor, sessions, running, index) {
  for (let waited = 0; ; waited += POLL_MS) {
    let stable = true;
    for (const name of Object.keys(running)) {
      const step = running[name];
      if (step === null) continue;
      if (step.done) {
        if (step.blocked) step.resumedAt = index;
        running[name] = null;
        continue;
      }
      const { rows } = await monitor.query(
        'SELECT cardinality(pg_blocking_pids($1)) > 0 AS waiting',
        [sessions[name].pid],
      );
      if (rows[0].waiting) step.blocked = true;
      else stable = false;
    }
    if (stable) return;
    if (waited > TIMEOUT_MS) throw new Error('una sentencia no termina ni espera un lock');
    await sleep(POLL_MS);
  }
}

// Ejecuta el escenario `name` con el nivel `level` ('ru', 'rc', 'rr', 'sr').
// kind es 'pg' o 'psql'. Devuelve los pasos, cada uno con:
//   session, sql, label
//   blocked     true si tuvo que esperar un lock
//   resumedAt   índice del paso después del cual continuó (si esperó)
//   rows, command, error { code, message }   (con pg)
//   output                                    (con psql)
// y byLabel, para buscar un paso por su etiqueta.
export async function runScenario(name, level, { kind = 'pg', psqlFlags = [] } = {}) {
  const build = SCENARIOS[name];
  if (!build) throw new Error(`no existe el escenario ${name}`);
  const monitor = new pg.Client({ application_name: 'w13_monitor' });
  await monitor.connect();
  const { rows } = await monitor.query('SELECT current_database() AS db');
  if (!rows[0].db.startsWith('chasqui_w13')) {
    await monitor.end();
    throw new Error('los escenarios solo corren sobre chasqui_w13 (PGDATABASE=chasqui_w13)');
  }

  const saved = await prepare(monitor);
  const make = (s) => (kind === 'psql' ? new PsqlSession(s, monitor, psqlFlags) : new PgSession(s));
  const sessions = { A: make('A'), B: make('B') };
  const running = { A: null, B: null };
  const steps = [];
  try {
    await sessions.A.open();
    await sessions.B.open();
    for (const [session, sql, label] of build(level)) {
      if (running[session] !== null) throw new Error(`la sesión ${session} todavía está esperando`);
      const step = { session, sql, label, blocked: false, done: false };
      steps.push(step);
      running[session] = step;
      sessions[session].send(step);
      await settle(monitor, sessions, running, steps.length - 1);
    }
  } finally {
    await sessions.A.close();
    await sessions.B.close();
    await restore(monitor, saved);
    await monitor.end();
  }
  const byLabel = Object.fromEntries(steps.filter((s) => s.label).map((s) => [s.label, s]));
  return { steps, byLabel };
}

// Las anomalías del capítulo, nivel por nivel, en PostgreSQL. Cada prueba
// corre un escenario de scenarios.mjs con dos clientes de pg y comprueba lo
// que se observó: qué leyó cada sesión, quién esperó un lock y quién recibió
// el error 40001 (serialization_failure).
import { test } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import { runScenario } from './sessions.mjs';

const ALL = ['ru', 'rc', 'rr', 'sr'];

// Lo que permite PostgreSQL en cada nivel (la tabla del capítulo).
const ALLOWS = {
  dirtyRead: [],
  nonRepeatableRead: ['ru', 'rc'],
  phantom: ['ru', 'rc'],
  lostUpdate: ['ru', 'rc'],
  writeSkew: ['ru', 'rc', 'rr'],
};

for (const level of ALL) {
  test(`${level}: no hay dirty reads`, async () => {
    const { byLabel } = await runScenario('dirty-read', level);
    assert.equal(byLabel.read.rows[0].price, '50.00'); // no ve el 0 de B
    assert.equal(ALLOWS.dirtyRead.includes(level), false);
  });

  test(`${level}: non-repeatable read ${ALLOWS.nonRepeatableRead.includes(level) ? 'sí' : 'no'}`, async () => {
    const { byLabel } = await runScenario('non-repeatable-read', level);
    assert.equal(byLabel.first.rows[0].price, '50.00');
    const expected = ALLOWS.nonRepeatableRead.includes(level) ? '60.00' : '50.00';
    assert.equal(byLabel.second.rows[0].price, expected);
  });

  test(`${level}: phantom ${ALLOWS.phantom.includes(level) ? 'sí' : 'no'}`, async () => {
    const { byLabel } = await runScenario('phantom', level);
    assert.deepEqual(byLabel.first.rows[0], { count: '6', sum: '166.60' });
    const expected = ALLOWS.phantom.includes(level) ? { count: '7', sum: '176.60' } : { count: '6', sum: '166.60' };
    assert.deepEqual(byLabel.second.rows[0], expected);
  });

  test(`${level}: lost update ${ALLOWS.lostUpdate.includes(level) ? 'sí' : 'no, 40001'}`, async () => {
    const { byLabel } = await runScenario('lost-update', level);
    assert.equal(byLabel.aRead.rows[0].price, '50.00');
    assert.equal(byLabel.bRead.rows[0].price, '50.00');
    assert.equal(byLabel.bWrite.blocked, true); // B espera el lock de la fila
    if (ALLOWS.lostUpdate.includes(level)) {
      assert.equal(byLabel.bWrite.error, undefined);
      assert.equal(byLabel.final.rows[0].price, '40.00'); // se perdió el embalaje
    } else {
      assert.equal(byLabel.bWrite.error.code, '40001');
      assert.match(byLabel.bWrite.error.message, /concurrent update/);
      assert.equal(byLabel.bCommit.command, 'ROLLBACK'); // COMMIT de una transacción abortada
      assert.equal(byLabel.final.rows[0].price, '55.00');
    }
  });

  test(`${level}: write skew ${ALLOWS.writeSkew.includes(level) ? 'sí' : 'no, 40001 en el COMMIT'}`, async () => {
    const { byLabel } = await runScenario('write-skew', level);
    assert.equal(byLabel.aSum.rows[0].sum, '95.00');
    assert.equal(byLabel.bSum.rows[0].sum, '95.00');
    assert.equal(byLabel.bWrite.blocked, false); // filas distintas: nadie espera
    assert.equal(byLabel.aCommit.error, undefined);
    if (ALLOWS.writeSkew.includes(level)) {
      assert.equal(byLabel.bCommit.error, undefined);
      assert.equal(byLabel.final.rows[0].sum, '85.00'); // la regla (>= 90) se rompió
    } else {
      assert.equal(byLabel.bCommit.error.code, '40001');
      assert.match(byLabel.bCommit.error.message, /read\/write dependencies/);
      assert.equal(byLabel.final.rows[0].sum, '90.00');
    }
  });
}

test('rc: UPDATE ... SET price = price + 5 espera y relee: no se pierde nada', async () => {
  const { byLabel } = await runScenario('atomic-update', 'rc');
  assert.equal(byLabel.bWrite.blocked, true);
  assert.equal(byLabel.bWrite.error, undefined);
  assert.equal(byLabel.final.rows[0].price, '45.00'); // 50 + 5 - 10
});

test('rr: el mismo UPDATE de una sola sentencia recibe 40001', async () => {
  const { byLabel } = await runScenario('atomic-update', 'rr');
  assert.equal(byLabel.bWrite.blocked, true);
  assert.equal(byLabel.bWrite.error.code, '40001');
  assert.equal(byLabel.final.rows[0].price, '55.00');
});

test('rc: con SELECT ... FOR UPDATE, B espera en la lectura y lee 55', async () => {
  const { byLabel } = await runScenario('for-update', 'rc');
  assert.equal(byLabel.aRead.rows[0].price, '50.00');
  assert.equal(byLabel.bRead.blocked, true);
  assert.equal(byLabel.bRead.rows[0].price, '55.00');
  assert.equal(byLabel.final.rows[0].price, '45.00');
});

test('sr: B recibe 40001, reintenta la transacción completa y el resultado es serial', async () => {
  const { byLabel } = await runScenario('retry', 'sr');
  assert.equal(byLabel.bWrite.error.code, '40001');
  assert.equal(byLabel.bReread.rows[0].price, '55.00');
  assert.equal(byLabel.bCommit.command, 'COMMIT');
  assert.equal(byLabel.final.rows[0].price, '45.00'); // como A y después B
});

test('psql2: la salida de psql del lost update en READ COMMITTED', async () => {
  const { steps, byLabel } = await runScenario('lost-update', 'rc', { kind: 'psql' });
  assert.equal(byLabel.bWrite.blocked, true);
  assert.equal(byLabel.bWrite.output, 'UPDATE 1');
  assert.equal(steps[6].output, 'COMMIT');
  assert.equal(byLabel.bWrite.resumedAt, 6); // continúa después del COMMIT de A
  assert.match(byLabel.final.output, /40\.00/);
});

test('los escenarios dejan los datos como estaban', async () => {
  const client = new pg.Client();
  await client.connect();
  try {
    const { rows } = await client.query(
      'SELECT shipment_id, price FROM shipment WHERE shipment_id IN (501, 502, 12001) ORDER BY shipment_id',
    );
    assert.deepEqual(rows, [
      { shipment_id: 501, price: '50.25' },
      { shipment_id: 502, price: '45.26' },
    ]);
  } finally {
    await client.end();
  }
});

test('rr: el snapshot se toma en la primera consulta, no en el BEGIN', async () => {
  const { byLabel } = await runScenario('snapshot-start', 'rr');
  assert.equal(byLabel.first.rows[0].price, '60.00'); // ve el primer UPDATE de B
  assert.equal(byLabel.second.rows[0].price, '60.00'); // no ve el segundo
});

// Lo que muestra examples.sql, comprobado con un cliente de pg.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';

async function withClient(fn) {
  const client = new pg.Client();
  await client.connect();
  try {
    return await fn(client);
  } finally {
    await client.end();
  }
}

const price = async (c, id) =>
  (await c.query('SELECT price FROM shipment WHERE shipment_id = $1', [id])).rows[0].price;

test('el nivel por defecto es READ COMMITTED; READ UNCOMMITTED se acepta', () =>
  withClient(async (c) => {
    assert.equal((await c.query('SHOW transaction_isolation')).rows[0].transaction_isolation, 'read committed');
    await c.query('BEGIN ISOLATION LEVEL READ UNCOMMITTED');
    assert.equal((await c.query('SHOW transaction_isolation')).rows[0].transaction_isolation, 'read uncommitted');
    await c.query('COMMIT');
  }));

test('SET TRANSACTION después de una consulta falla', () =>
  withClient(async (c) => {
    await c.query('BEGIN');
    await price(c, 501);
    await assert.rejects(c.query('SET TRANSACTION ISOLATION LEVEL SERIALIZABLE'), {
      message: 'SET TRANSACTION ISOLATION LEVEL must be called before any query',
    });
    await c.query('ROLLBACK');
  }));

test('después de un error la transacción queda abortada y COMMIT hace rollback', () =>
  withClient(async (c) => {
    await c.query('BEGIN');
    await c.query('UPDATE shipment SET price = price + 5 WHERE shipment_id = 501');
    await assert.rejects(c.query('UPDATE shipment SET price = -1 WHERE shipment_id = 502'), { code: '23514' });
    await assert.rejects(price(c, 501), { code: '25P02' });
    assert.equal((await c.query('COMMIT')).command, 'ROLLBACK');
    assert.equal(await price(c, 501), '50.25');
  }));

test('ROLLBACK TO SAVEPOINT deshace solo lo que vino después', () =>
  withClient(async (c) => {
    await c.query('BEGIN');
    await c.query('UPDATE shipment SET price = price + 5 WHERE shipment_id = 501');
    await c.query('SAVEPOINT before_discount');
    await assert.rejects(c.query('UPDATE shipment SET price = -1 WHERE shipment_id = 502'), { code: '23514' });
    await c.query('ROLLBACK TO SAVEPOINT before_discount');
    assert.equal(await price(c, 501), '55.25');
    assert.equal(await price(c, 502), '45.26');
    await c.query('ROLLBACK');
    assert.equal(await price(c, 501), '50.25');
  }));

// Los schedules del capítulo: conflictos, grafo de precedencia, orden serial
// o ciclo, y recuperabilidad.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  parseSchedule,
  conflicts,
  precedenceGraph,
  conflictSerializable,
  serialSchedule,
  readsFrom,
  isRecoverable,
  isCascadeless,
  show,
} from './precedence.mjs';

const edgesOf = (s) => precedenceGraph(s).edges.map((e) => `T${e.from}->T${e.to}`);

test('parseSchedule lee lecturas, escrituras, commits y aborts', () => {
  assert.deepEqual(parseSchedule('r1(A) w2(B), c1 a2'), [
    { op: 'r', txn: 1, item: 'A', pos: 0 },
    { op: 'w', txn: 2, item: 'B', pos: 1 },
    { op: 'c', txn: 1, item: null, pos: 2 },
    { op: 'a', txn: 2, item: null, pos: 3 },
  ]);
  assert.throws(() => parseSchedule('r1(A) x2(B)'), /no válida: "x2\(B\)"/);
  assert.throws(() => parseSchedule('r1'), /le falta el ítem/);
  assert.throws(() => parseSchedule('c1(A)'), /no llevan ítem/);
});

test('dos lecturas no entran en conflicto; r-w, w-r y w-w sí', () => {
  const pairs = conflicts('r1(A) r2(A) w1(B) r2(B) w2(B) w1(C) w2(C)').map(([p, q]) => `${show(p)}–${show(q)}`);
  assert.deepEqual(pairs, ['w1(B)–r2(B)', 'w1(B)–w2(B)', 'w1(C)–w2(C)']);
});

test('el ejemplo resuelto: T2 → T1, T1 → T3, T2 → T3; orden T2, T1, T3', () => {
  const s = 'r1(A) r2(B) w1(A) r3(A) w2(B) r3(B) w3(A) r1(B)';
  const graph = precedenceGraph(s);
  assert.deepEqual(graph.edges, [
    { from: 1, to: 3, because: ['r1(A)–w3(A)', 'w1(A)–r3(A)', 'w1(A)–w3(A)'] },
    { from: 2, to: 1, because: ['w2(B)–r1(B)'] },
    { from: 2, to: 3, because: ['w2(B)–r3(B)'] },
  ]);
  const result = conflictSerializable(s);
  assert.equal(result.serializable, true);
  assert.deepEqual(result.order, [2, 1, 3]);
  assert.equal(serialSchedule(s, result.order), 'r2(B) w2(B) r1(A) w1(A) r1(B) r3(A) r3(B) w3(A)');
});

test('el orden serial no tiene que ser el de llegada', () => {
  // T1 empieza primero, pero lee el A que escribió T2: equivale a T2, T1.
  const result = conflictSerializable('r1(B) r2(A) w2(A) r1(A)');
  assert.deepEqual(result.order, [2, 1]);
});

test('un schedule serial es serializable en su propio orden', () => {
  const result = conflictSerializable('r2(A) w2(A) r1(A) w1(A) r3(B)');
  assert.deepEqual(result.order, [2, 1, 3]);
});

test('el lost update: r1(A) r2(A) w1(A) w2(A) tiene el ciclo T1 → T2 → T1', () => {
  const result = conflictSerializable('r1(A) r2(A) w1(A) w2(A)');
  assert.equal(result.serializable, false);
  assert.deepEqual(result.cycle, [1, 2, 1]);
  assert.deepEqual(edgesOf('r1(A) r2(A) w1(A) w2(A)'), ['T1->T2', 'T2->T1']);
});

test('view serializable pero no conflict serializable: r1(A) w2(A) w1(A) w3(A)', () => {
  const s = 'r1(A) w2(A) w1(A) w3(A)';
  assert.deepEqual(edgesOf(s), ['T1->T2', 'T1->T3', 'T2->T1', 'T2->T3']);
  assert.deepEqual(conflictSerializable(s).cycle, [1, 2, 1]);
});

test('el grafo no mira los valores: las dos transferencias tienen ciclo', () => {
  const s = 'r1(A) w1(A) r2(B) w2(B) r1(B) w1(B) r2(A) w2(A)';
  assert.deepEqual(edgesOf(s), ['T1->T2', 'T2->T1']);
  assert.equal(conflictSerializable(s).serializable, false);
});

test('un ciclo de tres transacciones', () => {
  const result = conflictSerializable('r1(A) w2(A) r2(B) w3(B) r3(C) w1(C)');
  assert.deepEqual(result.cycle, [1, 2, 3, 1]);
});

test('la boleta: T2 lee el precio que escribió T1 y hace commit antes de que T1 aborte', () => {
  const s = 'w1(A) r2(A) c2 a1';
  assert.deepEqual(readsFrom(s).map((r) => [r.reader, r.writer, r.item]), [[2, 1, 'A']]);
  assert.equal(conflictSerializable(s).serializable, true); // serializable, y aun así está mal
  assert.equal(isRecoverable(s).recoverable, false);
  assert.equal(isCascadeless(s).cascadeless, false);
});

test('si T2 espera el commit de T1 para hacer commit, el schedule es recuperable', () => {
  const s = 'w1(A) r2(A) c1 c2';
  assert.equal(isRecoverable(s).recoverable, true);
  assert.equal(isCascadeless(s).cascadeless, false); // T2 leyó antes del commit de T1
});

test('rollback en cascada: T3 lee de T2, que leyó de T1, y T1 aborta', () => {
  const s = 'w1(A) r2(A) w2(B) r3(B) a1';
  assert.deepEqual(readsFrom(s).map((r) => `T${r.reader} lee ${r.item} de T${r.writer}`), [
    'T2 lee A de T1',
    'T3 lee B de T2',
  ]);
  assert.equal(isRecoverable(s).recoverable, true); // nadie hizo commit todavía
  assert.equal(isCascadeless(s).cascadeless, false);
});

test('cascadeless: cada lectura llega después del commit de quien escribió', () => {
  const s = 'w1(A) c1 r2(A) w2(B) c2 r3(B) c3';
  assert.equal(isCascadeless(s).cascadeless, true);
  assert.equal(isRecoverable(s).recoverable, true);
});

test('después de un abort, la lectura ve la escritura anterior', () => {
  // w1(A) se deshizo; r3(A) lee el A que escribió T2 (que ya hizo commit).
  const s = 'w2(A) c2 w1(A) a1 r3(A) c3';
  assert.deepEqual(readsFrom(s).map((r) => [r.reader, r.writer]), [[3, 2]]);
  assert.equal(isCascadeless(s).cascadeless, true);
});