// Dos deadlocks reales entre dos sesiones, A y B, sobre los envíos 501 y 502.
//
//   node deadlock.mjs          orden cruzado: A toma 501 y pide 502; B toma
//                              502 y pide 501
//   node deadlock.mjs upgrade  upgrade: A y B leen 501 con FOR SHARE y las
//                              dos quieren actualizarlo
//   node deadlock.mjs cross 1s 4s   con otros deadlock_timeout para A y B
//                              (por defecto, 4s y 2s)
//
// Mientras las dos esperan, una tercera sesión (el observador) muestra
// pg_locks, pg_blocking_pids y el waits-for graph armado con findCycle.
// deadlock_timeout es distinto en cada sesión para que se sepa de antemano
// quién encuentra el ciclo: en PostgreSQL, la sesión que lo encuentra es la
// que aborta, con el error 40P01.
import {
  openSession, launch, waitUntilBlocked, lockRows, blockingRows,
  psqlShow, closeAll, LOCKS_SQL, BLOCKING_SQL,
} from './sessions.mjs';
import { findCycle, edgesFromBlocking } from './waitsfor.mjs';

const UPDATE = 'UPDATE shipment SET price = price + 1 WHERE shipment_id = $1';
const FOR_SHARE = 'SELECT price FROM shipment WHERE shipment_id = $1 FOR SHARE';

// kind: 'cross' o 'upgrade'. timeouts: deadlock_timeout de A y de B.
// print: función que recibe cada línea de la traza (por defecto, nada).
export async function runDeadlock({
  kind = 'cross', timeoutA = '4s', timeoutB = '2s', print = () => {},
} = {}) {
  let a, b, obs;
  try {
    a = await openSession('A', { deadlockTimeout: timeoutA });
    b = await openSession('B', { deadlockTimeout: timeoutB });
    obs = await openSession('obs');
    const step = async (s, sql, params) => {
      print(`${s.name}: ${sql.replace('$1', params?.[0])}`);
      return s.query(sql, params);
    };

    await step(a, 'BEGIN');
    await step(b, 'BEGIN');
    const [xa, xb] = [
      (await a.query('SELECT pg_current_xact_id()::text AS x')).rows[0].x,
      (await b.query('SELECT pg_current_xact_id()::text AS x')).rows[0].x,
    ];
    print(`   (A es el pid ${a.pid}, xid ${xa}; B es el pid ${b.pid}, xid ${xb})`);

    let waitA, waitB;
    if (kind === 'cross') {
      await step(a, UPDATE, [501]);
      await step(b, UPDATE, [502]);
      print('A: ' + UPDATE.replace('$1', 502) + '   -- espera');
      waitA = launch(a, UPDATE, [502]);
      await waitUntilBlocked(obs, a.pid);
    } else {
      await step(a, FOR_SHARE, [501]);
      await step(b, FOR_SHARE, [501]);
      print('A: ' + UPDATE.replace('$1', 501) + '   -- espera');
      waitA = launch(a, UPDATE, [501]);
      await waitUntilBlocked(obs, a.pid);
    }
    const locksOneWaiting = await lockRows(obs, [a, b]);
    const shownOne = psqlShow(LOCKS_SQL, [a, b]);

    // En los dos casos, B termina pidiendo el envío 501.
    print('B: ' + UPDATE.replace('$1', 501) + '   -- espera');
    waitB = launch(b, UPDATE, [501]);
    await waitUntilBlocked(obs, b.pid);

    // Las dos esperan: todavía no pasó ningún deadlock_timeout.
    const locks = await lockRows(obs, [a, b]);
    const blocking = await blockingRows(obs, [a, b]);
    const shownLocks = psqlShow(LOCKS_SQL, [a, b]);
    const shownBlocking = psqlShow(BLOCKING_SQL, [a, b]);
    const graph = findCycle(edgesFromBlocking(blocking));
    print('');
    print('pg_locks con A esperando:');
    print(shownOne.trimEnd());
    print('');
    print('pg_locks con las dos esperando:');
    print(shownLocks.trimEnd());
    print('');
    print(shownBlocking.trimEnd());
    print('');
    print(`findCycle: ${JSON.stringify(graph)}`);

    const [ra, rb] = await Promise.all([waitA, waitB]);
    const results = { A: ra, B: rb };
    let victim = null;
    // Primero la víctima: su error es lo que deja continuar a la otra.
    print('');
    for (const [name, r] of Object.entries(results).sort(([, r]) => (r.error ? -1 : 1))) {
      if (r.error) {
        victim = name;
        print(`${name}: ERROR:  ${r.error.message}  (código ${r.error.code})`);
        print(`DETAIL:  ${r.error.detail.replaceAll('\n', '\n         ')}`);
      } else {
        print(`${name}: UPDATE ${r.rowCount}   -- continúa`);
      }
    }
    const survivor = victim === 'A' ? b : a;
    const loser = victim === 'A' ? a : b;
    await step(loser, 'ROLLBACK');
    await step(survivor, 'ROLLBACK'); // no cambiamos los precios de verdad

    return {
      pids: { A: a.pid, B: b.pid }, xids: { A: xa, B: xb },
      locksOneWaiting, locks, blocking, graph, victim,
      error: results[victim]?.error ?? null,
      shown: { one: shownOne, locks: shownLocks, blocking: shownBlocking },
    };
  } finally {
    await closeAll(a, b, obs);
  }
}

if (import.meta.url === `file://${process.argv[1]}`) {
  const kind = process.argv[2] === 'upgrade' ? 'upgrade' : 'cross';
  const [timeoutA = '4s', timeoutB = '2s'] = process.argv.slice(3);
  await runDeadlock({ kind, timeoutA, timeoutB, print: (line) => console.log(line) });
}

// Los locks que toma PostgreSQL en tres situaciones, vistos desde pg_locks:
//
//   1. A lee el envío 501 y luego lo actualiza: locks de tabla
//      (AccessShareLock, RowExclusiveLock) y el lock de A sobre su propio xid.
//   2. B quiere actualizar la misma fila: espera un ShareLock sobre el xid de
//      A. Cuando A termina, B continúa.
//   3. Granularidad: con A dentro de un UPDATE (ROW EXCLUSIVE en la tabla),
//      B pide LOCK TABLE shipment IN SHARE MODE y espera.
//
//   node locks.mjs           los tres casos
//   node locks.mjs matrix    la matriz de conflictos de los locks de tabla,
//                            medida con LOCK TABLE ... NOWAIT
import {
  openSession, launch, waitUntilBlocked, lockRows, blockingRows,
  psqlShow, closeAll, LOCKS_SQL, BLOCKING_SQL,
} from './sessions.mjs';

const UPDATE = 'UPDATE shipment SET price = price + 1 WHERE shipment_id = 501';

export async function runLocks({ print = () => {} } = {}) {
  let a, b, obs;
  const out = {};
  try {
    a = await openSession('A');
    b = await openSession('B');
    obs = await openSession('obs');
    const say = (s, sql) => print(`${s.name}: ${sql}`);
    const show = (sql) => {
      const text = psqlShow(sql, [a, b]);
      print(text.trimEnd() + '\n');
      return text;
    };

    // 1. Una lectura y una escritura.
    print('-- 1. Una lectura y una escritura');
    say(a, 'BEGIN');
    await a.query('BEGIN');
    say(a, 'SELECT price FROM shipment WHERE shipment_id = 501');
    await a.query('SELECT price FROM shipment WHERE shipment_id = 501');
    out.afterRead = await lockRows(obs, [a]);
    show(LOCKS_SQL);
    say(a, UPDATE);
    await a.query(UPDATE);
    out.xidA = (await a.query('SELECT pg_current_xact_id()::text AS x')).rows[0].x;
    out.afterWrite = await lockRows(obs, [a]);
    out.shownWrite = show(LOCKS_SQL);

    // 2. Una espera.
    print('\n-- 2. Una espera');
    say(b, 'BEGIN');
    await b.query('BEGIN');
    say(b, `${UPDATE}   -- espera`);
    const waiting = launch(b, UPDATE);
    await waitUntilBlocked(obs, b.pid);
    out.waiting = await lockRows(obs, [a, b]);
    out.blocking = await blockingRows(obs, [a, b]);
    out.shownWait = show(LOCKS_SQL);
    show(BLOCKING_SQL);
    say(a, 'ROLLBACK');
    await a.query('ROLLBACK');
    const r = await waiting;
    out.bAfter = r;
    print(`B: UPDATE ${r.rowCount}   -- continúa`);
    say(b, 'ROLLBACK');
    await b.query('ROLLBACK');

    // 3. Granularidad: ROW EXCLUSIVE (como IX) contra SHARE (como S).
    print('\n-- 3. Granularidad');
    say(a, 'BEGIN');
    await a.query('BEGIN');
    say(a, UPDATE);
    await a.query(UPDATE);
    say(b, 'BEGIN');
    await b.query('BEGIN');
    say(b, 'LOCK TABLE shipment IN SHARE MODE   -- espera');
    const tableLock = launch(b, 'LOCK TABLE shipment IN SHARE MODE');
    await waitUntilBlocked(obs, b.pid);
    out.tableWait = await lockRows(obs, [a, b]);
    out.shownTable = show(LOCKS_SQL);
    say(a, 'ROLLBACK');
    await a.query('ROLLBACK');
    out.tableAfter = await tableLock;
    print('B: LOCK TABLE   -- continúa');
    out.tableHeld = await lockRows(obs, [b]);
    show(LOCKS_SQL);
    say(b, 'ROLLBACK');
    await b.query('ROLLBACK');
    return out;
  } finally {
    await closeAll(a, b, obs);
  }
}

// Los ocho modos de lock de tabla de PostgreSQL, del más débil al más fuerte.
export const TABLE_MODES = [
  'ACCESS SHARE', 'ROW SHARE', 'ROW EXCLUSIVE', 'SHARE UPDATE EXCLUSIVE',
  'SHARE', 'SHARE ROW EXCLUSIVE', 'EXCLUSIVE', 'ACCESS EXCLUSIVE',
];

// La matriz de conflictos, medida: A toma un modo sobre agency y B pide otro
// con NOWAIT, que falla con 55P03 en vez de esperar.
export async function conflictMatrix({ print = () => {} } = {}) {
  let a, b;
  const matrix = {};
  try {
    a = await openSession('A');
    b = await openSession('B');
    for (const held of TABLE_MODES) {
      matrix[held] = {};
      await a.query('BEGIN');
      await a.query(`LOCK TABLE agency IN ${held} MODE`);
      for (const asked of TABLE_MODES) {
        await b.query('BEGIN');
        const r = await launch(b, `LOCK TABLE agency IN ${asked} MODE NOWAIT`);
        if (r.error && r.error.code !== '55P03') throw r.error;
        matrix[held][asked] = !r.error; // true = compatible
        await b.query('ROLLBACK');
      }
      await a.query('ROLLBACK');
    }
  } finally {
    await closeAll(a, b);
  }
  const short = (m) => m.split(' ').map((w) => w[0]).join('');
  print(`${'tiene \\ pide'.padEnd(24)}${TABLE_MODES.map((m) => short(m).padStart(5)).join('')}`);
  for (const held of TABLE_MODES) {
    print(`${held.padEnd(24)}${TABLE_MODES.map((m) => (matrix[held][m] ? 'ok' : 'X').padStart(5)).join('')}`);
  }
  return matrix;
}

if (import.meta.url === `file://${process.argv[1]}`) {
  const print = (line) => console.log(line);
  if (process.argv[2] === 'matrix') await conflictMatrix({ print });
  else await runLocks({ print });
}

-- Consultas para mirar los locks desde una tercera sesión de psql, mientras
-- otras dos trabajan. Se corren con \i locks.sql cada vez que quieran mirar.

-- Los locks de las demás sesiones: los concedidos primero y después los que
-- esperan (granted = f).
SELECT pid, locktype, relation::regclass, page, tuple, transactionid, mode, granted
FROM pg_locks
WHERE pid <> pg_backend_pid()
  AND locktype <> 'virtualxid'
ORDER BY pid, granted DESC;

-- Quién bloquea a quién: el waits-for graph, una arista por cada pid de
-- blocked_by.
SELECT pid, application_name, pg_blocking_pids(pid) AS blocked_by, state, query
FROM pg_stat_activity
WHERE datname = current_database()
  AND pid <> pg_backend_pid();

// Las versiones de una fila en PostgreSQL, vistas desde dos sesiones.
//
//   node mvcc.mjs versions   xmin, xmax y ctid del envío 501 antes, durante y
//                            después de un UPDATE de A; un SELECT ... FOR
//                            UPDATE que cambia xmax sin crear versión; el
//                            snapshot de B mientras A está abierta
//   node mvcc.mjs vacuum     tuplas muertas y VACUUM, sin y con una
//                            transacción REPEATABLE READ abierta
//
// Los dos dejan los datos como estaban (el precio vuelve a su valor), pero
// las filas quedan en otras posiciones (otro ctid).
import { openSession, closeAll } from './sessions.mjs';

const VERSION = 'SELECT ctid, xmin, xmax, price FROM shipment WHERE shipment_id = 501';
const UPDATE = 'UPDATE shipment SET price = price + 1 WHERE shipment_id = 501';

export async function runVersions({ print = () => {} } = {}) {
  let a, b;
  const out = {};
  try {
    a = await openSession('A');
    b = await openSession('B');
    const row = async (s, label) => {
      const r = (await s.query(VERSION)).rows[0];
      print(`${s.name}: ${VERSION}`);
      print(`   ctid=${r.ctid} xmin=${r.xmin} xmax=${r.xmax} price=${r.price}${label ? '   ' + label : ''}`);
      return r;
    };
    const run = async (s, sql) => {
      print(`${s.name}: ${sql}`);
      return s.query(sql);
    };

    out.before = await row(a);
    await run(a, 'BEGIN');
    await run(a, UPDATE);
    out.xidA = (await a.query('SELECT pg_current_xact_id()::text AS x')).rows[0].x;
    print(`   (xid de A: ${out.xidA})`);
    out.aSees = await row(a, '-- A ve su versión nueva');
    out.bSees = await row(b, '-- B ve la versión vieja');
    out.snapshot = (await b.query('SELECT pg_current_snapshot()::text AS s')).rows[0].s;
    print(`B: SELECT pg_current_snapshot();\n   ${out.snapshot}`);
    await run(a, 'COMMIT');
    out.bAfter = await row(b, '-- ahora B ve la nueva');

    // Un lock de fila sin versión nueva.
    await run(a, 'BEGIN');
    await run(a, 'SELECT price FROM shipment WHERE shipment_id = 501 FOR UPDATE');
    out.xidLock = (await a.query('SELECT pg_current_xact_id()::text AS x')).rows[0].x;
    print(`   (xid de A: ${out.xidLock})`);
    out.locked = await row(b, '-- mismo ctid, xmax es A');
    await run(a, 'ROLLBACK');
    out.afterLock = await row(b, '-- xmax sigue ahí, pero A abortó');

    // Dejar el precio como estaba.
    await a.query('UPDATE shipment SET price = price - 1 WHERE shipment_id = 501');
    return out;
  } finally {
    await closeAll(a, b);
  }
}

// La línea de VACUUM (VERBOSE) que cuenta las tuplas de la tabla. Llega como
// un aviso (notice), no como el resultado de una consulta.
async function vacuum(client) {
  const notices = [];
  const listen = (n) => notices.push(n.message);
  client.on('notice', listen);
  await client.query('VACUUM (VERBOSE) shipment');
  client.off('notice', listen);
  const report = notices.find((m) => m.startsWith('finished vacuuming') && m.includes('.public.shipment"'));
  const lines = report.split('\n');
  const line = lines.find((l) => l.startsWith('tuples:'));
  const cutoff = lines.find((l) => l.startsWith('removable cutoff:'));
  const [removed, remain, dead] = line
    .match(/tuples: (\d+) removed, (\d+) remain, (\d+) are dead/)
    .slice(1)
    .map(Number);
  return { line, cutoff, removed, remain, dead };
}

const TOUCH = 'UPDATE shipment SET price = price WHERE shipment_id BETWEEN 1 AND 100';

export async function runVacuum({ print = () => {} } = {}) {
  let a, b;
  const out = {};
  try {
    a = await openSession('A');
    b = await openSession('B');
    const run = async (s, sql) => {
      print(`${s.name}: ${sql}`);
      return s.query(sql);
    };
    const vac = async (label) => {
      print('A: VACUUM (VERBOSE) shipment');
      const v = await vacuum(a);
      print(`   ${v.line}\n   ${v.cutoff}`);
      out[label] = v;
    };

    await vac('clean'); // punto de partida: sin tuplas muertas
    const r = await run(a, TOUCH);
    out.touched = r.rowCount;
    print(`   UPDATE ${r.rowCount}`);
    await vac('first');

    await run(b, 'BEGIN ISOLATION LEVEL REPEATABLE READ');
    await run(b, 'SELECT count(*) FROM shipment');
    await run(a, TOUCH);
    await vac('blocked');
    await run(b, 'COMMIT');
    await vac('after');
    return out;
  } finally {
    await closeAll(a, b);
  }
}

if (import.meta.url === `file://${process.argv[1]}`) {
  const print = (line) => console.log(line);
  if (process.argv[2] === 'vacuum') await runVacuum({ print });
  else await runVersions({ print });
}

// Herramientas para los experimentos con varias sesiones: abrir clientes con
// nombre, lanzar una sentencia que se va a quedar esperando un lock, esperar
// (sin sleep) hasta que de verdad esté esperando, y mirar pg_locks y
// pg_stat_activity solo de nuestras sesiones (el servidor es compartido).
import pg from 'pg';
import { execFileSync } from 'node:child_process';

// Abre un cliente y le pone un nombre visible en pg_stat_activity.
// statement_timeout es una red de seguridad: si algo sale mal, nada se queda
// esperando para siempre.
export async function openSession(name, { deadlockTimeout = '1s' } = {}) {
  const client = new pg.Client({ application_name: name });
  await client.connect();
  await client.query("SELECT set_config('statement_timeout', '20s', false)");
  await client.query("SELECT set_config('deadlock_timeout', $1, false)", [deadlockTimeout]);
  const { rows } = await client.query('SELECT pg_backend_pid() AS pid');
  client.name = name;
  client.pid = rows[0].pid;
  return client;
}

// Lanza una sentencia sin esperarla. Devuelve una promesa que nunca se
// rechaza: se resuelve con { rows, rowCount } o con { error }.
export function launch(client, sql, params = []) {
  return client.query(sql, params).then(
    (res) => ({ rows: res.rows, rowCount: res.rowCount }),
    (error) => ({ error }),
  );
}

// Espera hasta que la sesión con ese pid esté bloqueada esperando un lock.
export async function waitUntilBlocked(observer, pid, timeoutMs = 10000) {
  const start = Date.now();
  while (Date.now() - start < timeoutMs) {
    const { rows } = await observer.query(
      `SELECT wait_event_type, cardinality(pg_blocking_pids(pid)) AS n
       FROM pg_stat_activity WHERE pid = $1`,
      [pid],
    );
    if (rows[0]?.wait_event_type === 'Lock' && rows[0].n > 0) return;
    await new Promise((r) => setTimeout(r, 20));
  }
  throw new Error(`la sesión ${pid} no llegó a esperar un lock`);
}

// Los locks de las sesiones dadas, sin los de las tablas del sistema y sin
// el virtualxid de cada transacción (siempre está y no aporta aquí).
export const LOCKS_SQL = `
SELECT a.application_name AS session, l.locktype,
       coalesce(l.relation::regclass::text, '') AS relation,
       coalesce(l.page::text || ',' || l.tuple::text, '') AS tuple,
       coalesce(l.transactionid::text, '') AS xid,
       l.mode, l.granted
FROM pg_locks l
  JOIN pg_stat_activity a ON a.pid = l.pid
WHERE l.pid = ANY ($1::int[])
  AND l.locktype <> 'virtualxid'
  AND (l.relation IS NULL OR l.relation >= 16384)
ORDER BY a.application_name, l.granted DESC, l.locktype, relation, l.mode`;

export async function lockRows(observer, sessions) {
  const { rows } = await observer.query(LOCKS_SQL, [sessions.map((s) => s.pid)]);
  return rows;
}

// Quién espera a quién, según pg_blocking_pids.
export const BLOCKING_SQL = `
SELECT application_name AS session, pid, pg_blocking_pids(pid) AS blocked_by,
       wait_event_type, state
FROM pg_stat_activity
WHERE pid = ANY ($1::int[])
ORDER BY application_name`;

export async function blockingRows(observer, sessions) {
  const { rows } = await observer.query(BLOCKING_SQL, [sessions.map((s) => s.pid)]);
  return rows;
}

// Muestra una consulta con el formato de psql, corriendo el psql de verdad.
// Los pids se pasan como variable de psql (:'pids'), no pegados al texto.
export function psqlShow(sql, sessions) {
  const pids = `{${sessions.map((s) => s.pid).join(',')}}`;
  const text = sql.replace('$1::int[]', ":'pids'::int[]");
  return execFileSync('psql', ['-X', '-q', '-v', `pids=${pids}`], {
    input: text + ';\n',
    encoding: 'utf8',
  });
}

// Cierra todas las sesiones, aunque alguna tenga una transacción abierta.
export async function closeAll(...clients) {
  for (const c of clients) {
    if (!c) continue;
    try {
      await c.query('ROLLBACK');
    } catch {
      // la conexión ya estaba rota o no había transacción
    }
    await c.end();
  }
}

// Write skew con la capacidad de un vehículo, y dos maneras de evitarlo.
//
// Preparación (setup): el viaje 12 usa el vehículo 11. Le dejamos una
// capacidad de 60 kg y tomamos dos envíos sin viaje: el 55 (48,20 kg) y el
// 444 (43,50 kg). Cada uno cabe solo; los dos juntos no.
//
// La empleada A quiere asignar el envío 55 y el empleado B, el 444. Cada uno
// revisa la carga del viaje y, si su envío cabe, lo asigna. El schedule es:
//
//   A: BEGIN, (lock del viaje), carga    B: BEGIN, (lock del viaje), carga
//   A: UPDATE, COMMIT                    B: UPDATE, COMMIT
//
//   node skew.mjs rr          REPEATABLE READ: las dos hacen commit (write skew)
//   node skew.mjs serializable SERIALIZABLE: una recibe 40001
//   node skew.mjs rc-lock     READ COMMITTED + SELECT ... FOR UPDATE del viaje
//   node skew.mjs rr-lock     REPEATABLE READ + SELECT ... FOR UPDATE del viaje
//
// Al final se devuelve todo a su estado original.
import { openSession, launch, waitUntilBlocked, psqlShow, closeAll } from './sessions.mjs';

export const TRIP = 12;
export const VEHICLE = 11;
export const CAPACITY = 60;
export const PICKS = { A: { shipment: 55, weight: 48.2 }, B: { shipment: 444, weight: 43.5 } };

export const LOAD_SQL = `
SELECT v.capacity_kg, coalesce(sum(s.weight_kg), 0) AS load_kg
FROM trip t
  JOIN vehicle v ON v.vehicle_id = t.vehicle_id
  LEFT JOIN shipment s ON s.trip_id = t.trip_id
WHERE t.trip_id = $1
GROUP BY v.capacity_kg`;
// Los locks de predicado de SSI: no bloquean a nadie, solo registran qué leyó
// cada transacción.
export const SIREAD_SQL = `
SELECT a.application_name AS session, l.locktype, l.relation::regclass AS relation,
       l.page, l.tuple, l.mode
FROM pg_locks l
  JOIN pg_stat_activity a ON a.pid = l.pid
WHERE l.pid = ANY ($1::int[]) AND l.mode = 'SIReadLock'
ORDER BY a.application_name, relation, l.page, l.tuple`;
const LOCK_TRIP = 'SELECT trip_id FROM trip WHERE trip_id = $1 FOR UPDATE';
const ASSIGN = 'UPDATE shipment SET trip_id = $1 WHERE shipment_id = $2';

const MODES = {
  rr: { level: 'REPEATABLE READ', lockTrip: false },
  serializable: { level: 'SERIALIZABLE', lockTrip: false },
  'rc-lock': { level: 'READ COMMITTED', lockTrip: true },
  'rr-lock': { level: 'REPEATABLE READ', lockTrip: true },
};

async function setup(client) {
  const { rows } = await client.query('SELECT capacity_kg FROM vehicle WHERE vehicle_id = $1', [VEHICLE]);
  await client.query('UPDATE vehicle SET capacity_kg = $1 WHERE vehicle_id = $2', [CAPACITY, VEHICLE]);
  await client.query('UPDATE shipment SET trip_id = NULL WHERE shipment_id = ANY ($1::int[])', [
    [PICKS.A.shipment, PICKS.B.shipment],
  ]);
  return rows[0].capacity_kg;
}

async function restore(client, capacity) {
  await client.query('UPDATE vehicle SET capacity_kg = $1 WHERE vehicle_id = $2', [capacity, VEHICLE]);
  await client.query('UPDATE shipment SET trip_id = NULL WHERE shipment_id = ANY ($1::int[])', [
    [PICKS.A.shipment, PICKS.B.shipment],
  ]);
}

export async function runSkew(mode, { print = () => {} } = {}) {
  const { level, lockTrip } = MODES[mode];
  let a, b, obs, original;
  const out = { mode, level, lockTrip, errors: {}, saw: {}, assigned: {} };
  try {
    a = await openSession('A');
    b = await openSession('B');
    obs = await openSession('obs');
    original = await setup(obs);

    // Una sentencia de una sesión; los errores se guardan, no se lanzan.
    const run = async (s, sql, params = [], show = sql) => {
      if (out.errors[s.name]) return null; // la transacción ya falló
      print(`${s.name}: ${show}`);
      const r = await launch(s, sql, params);
      if (r.error) {
        out.errors[s.name] = { code: r.error.code, message: r.error.message, at: show };
        print(`   ERROR:  ${r.error.message}  (código ${r.error.code})`);
      }
      return r;
    };
    const check = async (s) => {
      const r = await run(s, LOAD_SQL, [TRIP], 'carga del viaje 12');
      if (!r || r.error) return;
      const { capacity_kg, load_kg } = r.rows[0];
      const fits = Number(load_kg) + PICKS[s.name].weight <= capacity_kg;
      out.saw[s.name] = Number(load_kg);
      print(`   capacity_kg=${capacity_kg} load_kg=${load_kg} -> ${PICKS[s.name].weight} kg ${fits ? 'cabe' : 'no cabe'}`);
      return fits;
    };
    const assign = async (s, fits) => {
      if (!fits) return;
      const r = await run(s, ASSIGN, [TRIP, PICKS[s.name].shipment],
        `UPDATE shipment SET trip_id = ${TRIP} WHERE shipment_id = ${PICKS[s.name].shipment}`);
      if (r && !r.error) out.assigned[s.name] = true;
    };
    const finish = async (s) => {
      if (out.errors[s.name]) {
        await s.query('ROLLBACK');
        print(`${s.name}: ROLLBACK`);
        out.assigned[s.name] = false;
        return;
      }
      const r = await run(s, 'COMMIT');
      if (r.error) out.assigned[s.name] = false;
    };

    await run(a, `BEGIN ISOLATION LEVEL ${level}`);
    await run(b, `BEGIN ISOLATION LEVEL ${level}`);
    let fitsA, fitsB;
    if (!lockTrip) {
      fitsA = await check(a);
      fitsB = await check(b);
      await assign(a, fitsA);
      await assign(b, fitsB);
      if (level === 'SERIALIZABLE') {
        out.siread = (await obs.query(SIREAD_SQL, [[a.pid, b.pid]])).rows;
        print('');
        print(psqlShow(SIREAD_SQL, [a, b]).trimEnd());
        print('');
      }
      await finish(a);
      await finish(b);
    } else {
      await run(a, LOCK_TRIP, [TRIP], `SELECT trip_id FROM trip WHERE trip_id = ${TRIP} FOR UPDATE`);
      fitsA = await check(a);
      print(`B: SELECT trip_id FROM trip WHERE trip_id = ${TRIP} FOR UPDATE   -- espera`);
      const lockB = launch(b, LOCK_TRIP, [TRIP]);
      await waitUntilBlocked(obs, b.pid);
      out.bWaited = true;
      await assign(a, fitsA);
      await finish(a);
      const r = await lockB;
      if (r.error) {
        out.errors.B = { code: r.error.code, message: r.error.message, at: 'FOR UPDATE' };
        print(`   ERROR:  ${r.error.message}  (código ${r.error.code})`);
      } else {
        print('   -- B continúa');
      }
      fitsB = await check(b);
      await assign(b, fitsB);
      await finish(b);
    }

    const final = (await obs.query(LOAD_SQL, [TRIP])).rows[0];
    out.finalLoad = Number(final.load_kg);
    out.overloaded = out.finalLoad > CAPACITY;
    print('');
    print(psqlShow(LOAD_SQL.replace('$1', String(TRIP)), []).trimEnd());
    return out;
  } finally {
    if (obs && original !== undefined) await restore(obs, original);
    await closeAll(a, b, obs);
  }
}

if (import.meta.url === `file://${process.argv[1]}`) {
  const mode = process.argv[2] ?? 'rr';
  if (!MODES[mode]) {
    console.error(`uso: node skew.mjs ${Object.keys(MODES).join('|')}`);
    process.exit(1);
  }
  await runSkew(mode, { print: (line) => console.log(line) });
}

// Búsqueda de ciclos en un grafo dirigido dado como lista de aristas.
//
// Sirve para los dos grafos del curso que tienen una transacción por nodo:
//   - el grafo de precedencia de la semana 13 (Ti -> Tj: un conflicto de Ti
//     va antes que uno de Tj). Sin ciclo, `order` es un orden serial
//     equivalente.
//   - el waits-for graph de esta semana (Ti -> Tj: Ti espera un lock que
//     tiene Tj). Con ciclo, hay deadlock y `cycle` son sus transacciones.
//
//   findCycle([['T1', 'T2'], ['T2', 'T1']])
//     -> { acyclic: false, cycle: ['T1', 'T2'] }
//   findCycle([['T2', 'T1'], ['T1', 'T3']])
//     -> { acyclic: true, order: ['T2', 'T1', 'T3'] }
//
// Es una búsqueda en profundidad con tres colores: blanco (no visitado), gris
// (en la pila actual) y negro (terminado). Una arista hacia un nodo gris
// cierra un ciclo. Los nodos y vecinos se recorren en orden, para que el
// resultado no dependa del orden de las aristas.

const WHITE = 0;
const GRAY = 1;
const BLACK = 2;

const byName = (a, b) => (typeof a === 'number' && typeof b === 'number' ? a - b : String(a).localeCompare(String(b)));

export function findCycle(edges, nodes = []) {
  const next = new Map();
  const add = (n) => {
    if (!next.has(n)) next.set(n, new Set());
  };
  for (const n of nodes) add(n);
  for (const [from, to] of edges) {
    add(from);
    add(to);
    next.get(from).add(to);
  }

  const color = new Map([...next.keys()].map((n) => [n, WHITE]));
  const stack = []; // camino actual de la búsqueda
  const finished = []; // nodos en el orden en que se ponen negros

  function visit(n) {
    color.set(n, GRAY);
    stack.push(n);
    for (const m of [...next.get(n)].sort(byName)) {
      if (color.get(m) === GRAY) return stack.slice(stack.indexOf(m));
      if (color.get(m) === WHITE) {
        const cycle = visit(m);
        if (cycle) return cycle;
      }
    }
    stack.pop();
    color.set(n, BLACK);
    finished.push(n);
    return null;
  }

  for (const n of [...next.keys()].sort(byName)) {
    if (color.get(n) !== WHITE) continue;
    const cycle = visit(n);
    if (cycle) return { acyclic: false, cycle };
  }
  // Un nodo termina después de todos los que alcanza: al revés, es un orden
  // topológico (cada arista va de izquierda a derecha).
  return { acyclic: true, order: finished.reverse() };
}

// Aristas del waits-for graph de PostgreSQL a partir de filas con
// { pid, blocked_by }, como las que da pg_blocking_pids(pid).
export function edgesFromBlocking(rows) {
  return rows.flatMap((r) => r.blocked_by.map((holder) => [r.pid, holder]));
}

// Pruebas de deadlock.mjs: dos deadlocks reales y quién es la víctima.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { runDeadlock } from './deadlock.mjs';

const has = (rows, want) => rows.some((r) => Object.entries(want).every(([k, v]) => r[k] === v));

function checkCycle(out) {
  const { A, B } = out.pids;
  // Cada una espera un ShareLock sobre el xid de la otra.
  assert.ok(has(out.locks, { session: 'A', locktype: 'transactionid', xid: out.xids.B, mode: 'ShareLock', granted: false }));
  assert.ok(has(out.locks, { session: 'B', locktype: 'transactionid', xid: out.xids.A, mode: 'ShareLock', granted: false }));
  // El waits-for graph armado con pg_blocking_pids tiene el ciclo A -> B -> A.
  assert.deepEqual(out.blocking.map((r) => [r.session, r.blocked_by]), [['A', [B]], ['B', [A]]]);
  assert.equal(out.graph.acyclic, false);
  assert.deepEqual([...out.graph.cycle].sort(), [A, B].sort());
  // La víctima recibe 40P01, y el DETAIL nombra los dos procesos.
  assert.equal(out.error.code, '40P01');
  assert.equal(out.error.message, 'deadlock detected');
  assert.match(out.error.detail, new RegExp(`Process ${A} waits for ShareLock on transaction ${out.xids.B}; blocked by process ${B}`));
  assert.match(out.error.detail, new RegExp(`Process ${B} waits for ShareLock on transaction ${out.xids.A}; blocked by process ${A}`));
}

test('orden cruzado: B tiene el deadlock_timeout más corto y es la víctima', async () => {
  const out = await runDeadlock({ kind: 'cross', timeoutA: '6s', timeoutB: '2s' });
  checkCycle(out);
  assert.equal(out.victim, 'B');
  // Mientras solo A espera, A tiene el lock de la tupla que quiere.
  assert.ok(has(out.locksOneWaiting, { session: 'A', locktype: 'tuple', mode: 'ExclusiveLock', granted: true }));
});

test('orden cruzado: con los timeouts al revés, la víctima es A', async () => {
  const out = await runDeadlock({ kind: 'cross', timeoutA: '2s', timeoutB: '6s' });
  checkCycle(out);
  assert.equal(out.victim, 'A');
});

test('upgrade: las dos tienen FOR SHARE sobre 501 y las dos quieren escribirla', async () => {
  const out = await runDeadlock({ kind: 'upgrade', timeoutA: '6s', timeoutB: '2s' });
  checkCycle(out);
  assert.equal(out.victim, 'B');
  // FOR SHARE toma ROW SHARE en la tabla; el UPDATE, además, ROW EXCLUSIVE.
  assert.ok(has(out.locks, { session: 'B', relation: 'shipment', mode: 'RowShareLock', granted: true }));
  assert.ok(has(out.locks, { session: 'A', relation: 'shipment', mode: 'RowExclusiveLock', granted: true }));
});

// Pruebas de locks.mjs: lo que el apunte dice de pg_locks.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { runLocks, conflictMatrix, TABLE_MODES } from './locks.mjs';

const has = (rows, want) => rows.some((r) => Object.entries(want).every(([k, v]) => r[k] === v));

test('locks de una lectura, una escritura, una espera y un lock de tabla', async () => {
  const out = await runLocks();

  // Una lectura: ACCESS SHARE en la tabla y su índice, y todavía sin xid.
  assert.ok(has(out.afterRead, { relation: 'shipment', mode: 'AccessShareLock', granted: true }));
  assert.ok(has(out.afterRead, { relation: 'shipment_pkey', mode: 'AccessShareLock' }));
  assert.ok(!has(out.afterRead, { locktype: 'transactionid' }));

  // Una escritura: ROW EXCLUSIVE en la tabla, el xid propio en EXCLUSIVE y
  // ningún lock de fila en pg_locks.
  assert.ok(has(out.afterWrite, { relation: 'shipment', mode: 'RowExclusiveLock', granted: true }));
  assert.ok(has(out.afterWrite, { locktype: 'transactionid', xid: out.xidA, mode: 'ExclusiveLock' }));
  assert.ok(!has(out.afterWrite, { locktype: 'tuple' }));

  // Una espera: B pide ShareLock sobre el xid de A y no se lo dan; tiene el
  // lock de la tupla que quiere (su lugar en la fila de espera).
  const bRows = out.waiting.filter((r) => r.session === 'B');
  assert.ok(has(bRows, { locktype: 'transactionid', xid: out.xidA, mode: 'ShareLock', granted: false }));
  assert.ok(has(bRows, { locktype: 'tuple', relation: 'shipment', mode: 'ExclusiveLock', granted: true }));
  const blockedB = out.blocking.find((r) => r.session === 'B');
  const pidA = out.blocking.find((r) => r.session === 'A').pid;
  assert.deepEqual(blockedB.blocked_by, [pidA]);
  assert.equal(blockedB.wait_event_type, 'Lock');
  assert.equal(out.bAfter.rowCount, 1);

  // Granularidad: SHARE en la tabla espera a ROW EXCLUSIVE.
  assert.ok(has(out.tableWait, { session: 'B', relation: 'shipment', mode: 'ShareLock', granted: false }));
  assert.ok(has(out.tableHeld, { session: 'B', relation: 'shipment', mode: 'ShareLock', granted: true }));
});

test('la matriz de conflictos medida es la de la documentación de PostgreSQL', async () => {
  const m = await conflictMatrix();
  // Para cada modo, con qué modos entra en conflicto (la tabla "Conflicting Lock Modes" de Explicit Locking).
  const conflicts = {
    'ACCESS SHARE': ['ACCESS EXCLUSIVE'],
    'ROW SHARE': ['EXCLUSIVE', 'ACCESS EXCLUSIVE'],
    'ROW EXCLUSIVE': ['SHARE', 'SHARE ROW EXCLUSIVE', 'EXCLUSIVE', 'ACCESS EXCLUSIVE'],
    'SHARE UPDATE EXCLUSIVE': ['SHARE UPDATE EXCLUSIVE', 'SHARE', 'SHARE ROW EXCLUSIVE', 'EXCLUSIVE', 'ACCESS EXCLUSIVE'],
    SHARE: ['ROW EXCLUSIVE', 'SHARE UPDATE EXCLUSIVE', 'SHARE ROW EXCLUSIVE', 'EXCLUSIVE', 'ACCESS EXCLUSIVE'],
    'SHARE ROW EXCLUSIVE': TABLE_MODES.slice(2),
    EXCLUSIVE: TABLE_MODES.slice(1),
    'ACCESS EXCLUSIVE': TABLE_MODES,
  };
  for (const held of TABLE_MODES) {
    for (const asked of TABLE_MODES) {
      assert.equal(m[held][asked], !conflicts[held].includes(asked), `${held} / ${asked}`);
      assert.equal(m[held][asked], m[asked][held], 'la matriz es simétrica');
    }
  }
});

// Pruebas de mvcc.mjs: versiones de una fila, snapshots y VACUUM.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { runVersions, runVacuum } from './mvcc.mjs';

// pg_current_snapshot() se escribe xmin:xmax:xip1,xip2,...
function visibleIn(snapshot, xid) {
  const [xmin, xmax, list] = snapshot.split(':');
  const x = BigInt(xid);
  if (x < BigInt(xmin)) return true; // terminó antes del snapshot (si hizo commit)
  if (x >= BigInt(xmax)) return false; // empezó después o seguía en curso
  return !list.split(',').filter(Boolean).map(BigInt).includes(x);
}

test('un UPDATE escribe una versión nueva; cada sesión ve la suya', async () => {
  const out = await runVersions();
  // A ve la versión que creó: otro ctid, xmin = A, sin xmax.
  assert.notEqual(out.aSees.ctid, out.before.ctid);
  assert.equal(out.aSees.xmin, out.xidA);
  assert.equal(out.aSees.xmax, '0');
  assert.equal(Number(out.aSees.price), Number(out.before.price) + 1);
  // B ve la versión vieja, ya marcada con xmax = A.
  assert.equal(out.bSees.ctid, out.before.ctid);
  assert.equal(out.bSees.xmax, out.xidA);
  assert.equal(out.bSees.price, out.before.price);
  // En el snapshot de B, A no es visible.
  assert.equal(visibleIn(out.snapshot, out.xidA), false);
  // Después del commit, B ve la versión nueva.
  assert.equal(out.bAfter.ctid, out.aSees.ctid);
  // FOR UPDATE escribe xmax en la misma versión; el rollback no lo borra.
  assert.equal(out.locked.ctid, out.bAfter.ctid);
  assert.equal(out.locked.xmax, out.xidLock);
  assert.equal(out.locked.xmin, out.xidA);
  assert.equal(out.afterLock.xmax, out.xidLock);
});

test('VACUUM no puede quitar lo que un snapshot abierto todavía puede ver', async () => {
  const out = await runVacuum();
  assert.equal(out.touched, 100);
  assert.equal(out.clean.dead, 0);
  assert.deepEqual([out.first.removed, out.first.dead], [100, 0]);
  assert.deepEqual([out.blocked.removed, out.blocked.dead], [0, 100]);
  assert.deepEqual([out.after.removed, out.after.dead], [100, 0]);
});