// Primer contacto con pg: un Client para un script corto, un Pool para una
// aplicación, y cómo llegan los tipos de PostgreSQL a JavaScript.
// La configuración (host, usuario, base de datos) sale de PGHOST, PGUSER,
// PGDATABASE, etc.:
//   PGDATABASE=chasqui_w05 node connect.mjs
import pg from 'pg';

// 1. Client: una sola conexión. Se abre con connect() y se cierra con end().
const client = new pg.Client();
await client.connect();
try {
  const result = await client.query('SELECT name, city FROM agency WHERE region = $1', ['Puno']);
  console.log('rows:', result.rows);
  console.log('rowCount:', result.rowCount);
  console.log('fields:', result.fields.map((f) => f.name));
} finally {
  await client.end();
}

// 2. Pool: varias conexiones que se prestan. pool.query toma una libre,
//    ejecuta y la devuelve.
const pool = new pg.Pool({ max: 5 });
try {
  const { rows } = await pool.query(
    `SELECT s.shipment_id, s.weight_kg, s.price, s.created_at, s.delivered_at,
            count(e.seq) AS events
     FROM shipment s
       JOIN shipment_event e ON e.shipment_id = s.shipment_id
     WHERE s.shipment_id = $1
     GROUP BY s.shipment_id`,
    [1],
  );
  const s = rows[0];
  console.log(s);
  for (const key of Object.keys(s)) {
    const value = s[key];
    const kind = value instanceof Date ? 'Date' : typeof value;
    console.log(`${key.padEnd(13)} ${kind}`);
  }
  // numeric y bigint llegan como texto: para operar hay que convertir.
  console.log('suma con texto:', s.price + s.weight_kg);
  console.log('suma con Number:', Number(s.price) + Number(s.weight_kg));
} finally {
  await pool.end(); // sin esto, el proceso de Node no termina
}

// Marcar un envío como entregado: cambia shipment y agrega su evento
// 'delivered' en shipment_event. Las dos cosas o ninguna: una transacción
// con un mismo client del pool.

/**
 * Entrega el envío shipmentId en la agencia agencyId, a la hora at.
 * Devuelve el número de secuencia del evento nuevo.
 */
export async function deliverShipment(pool, shipmentId, agencyId, at = new Date()) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const updated = await client.query(
      `UPDATE shipment
       SET status = 'delivered', delivered_at = $2
       WHERE shipment_id = $1 AND status = 'in_transit'`,
      [shipmentId, at],
    );
    if (updated.rowCount !== 1) {
      throw new Error(`el envío ${shipmentId} no existe o no está en tránsito`);
    }
    const event = await client.query(
      `INSERT INTO shipment_event (shipment_id, seq, at, agency_id, kind)
       SELECT $1, max(seq) + 1, $2, $3, 'delivered'
       FROM shipment_event
       WHERE shipment_id = $1
       RETURNING seq`,
      [shipmentId, at, agencyId],
    );
    await client.query('COMMIT');
    return event.rows[0].seq;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

-- Semana 5: funciones y procedimientos.
-- Se cargan con: psql -d chasqui_w05 -f functions.sql (make lo hace).

-- 1. Función en SQL: el precio de un envío según la tarifa base de la ruta,
--    el peso y el valor declarado (1 % del valor declarado, si lo hay).
--    Es la misma fórmula con la que se generaron los datos de ejemplo.
CREATE OR REPLACE FUNCTION shipment_fare(p_route_id integer,
                                         p_weight numeric,
                                         p_declared numeric DEFAULT NULL)
RETURNS numeric
LANGUAGE sql
STABLE
AS $$
  SELECT round(r.base_fare + 0.9 * p_weight + 0.01 * coalesce(p_declared, 0), 2)
  FROM route r
  WHERE r.route_id = p_route_id;
$$;

-- 2. La misma idea en PL/pgSQL: busca la ruta por origen y destino, valida
--    los datos y avisa con RAISE EXCEPTION si algo no cuadra.
CREATE OR REPLACE FUNCTION quote_fare(p_origin integer,
                                      p_dest integer,
                                      p_weight numeric,
                                      p_declared numeric DEFAULT NULL)
RETURNS numeric
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
  v_fare numeric;
BEGIN
  IF p_weight IS NULL OR p_weight <= 0 THEN
    RAISE EXCEPTION 'peso inválido: %', p_weight
      USING ERRCODE = 'check_violation';
  END IF;

  SELECT base_fare INTO v_fare
  FROM route
  WHERE origin_agency_id = p_origin AND dest_agency_id = p_dest;

  IF NOT FOUND THEN
    RAISE EXCEPTION 'no hay ruta de la agencia % a la agencia %', p_origin, p_dest
      USING ERRCODE = 'no_data_found';
  END IF;

  RETURN round(v_fare + 0.9 * p_weight + 0.01 * coalesce(p_declared, 0), 2);
END;
$$;

-- 3. Una función que devuelve una tabla: las rutas que salen de las agencias
--    de una ciudad, con cuántos envíos llevó cada una. Se usa en el FROM.
CREATE OR REPLACE FUNCTION routes_from(p_city text)
RETURNS TABLE (route_id integer, origin text, dest text, base_fare numeric, shipments bigint)
LANGUAGE sql
STABLE
AS $$
  SELECT r.route_id, o.name, d.name, r.base_fare, count(s.shipment_id)
  FROM route r
    JOIN agency o ON o.agency_id = r.origin_agency_id
    JOIN agency d ON d.agency_id = r.dest_agency_id
    LEFT JOIN shipment s ON s.origin_agency_id = r.origin_agency_id
                        AND s.dest_agency_id = r.dest_agency_id
  WHERE o.city = p_city
  GROUP BY r.route_id, o.name, d.name, r.base_fare;
$$;

-- 4. Un procedimiento: recalcula una tabla de resumen por agencia. No
--    devuelve nada; se llama con CALL y puede hacer COMMIT por dentro.
CREATE TABLE IF NOT EXISTS agency_stats (
  agency_id    integer PRIMARY KEY REFERENCES agency,
  shipments    integer NOT NULL,
  revenue      numeric(12, 2) NOT NULL,
  refreshed_at timestamptz NOT NULL
);

CREATE OR REPLACE PROCEDURE refresh_agency_stats()
LANGUAGE plpgsql
AS $$
BEGIN
  DELETE FROM agency_stats;
  INSERT INTO agency_stats (agency_id, shipments, revenue, refreshed_at)
  SELECT a.agency_id, count(s.shipment_id), coalesce(sum(s.price), 0), now()
  FROM agency a
    LEFT JOIN shipment s ON s.origin_agency_id = a.agency_id
  GROUP BY a.agency_id;
  COMMIT;
END;
$$;

// Inyección SQL: la versión vulnerable frente a la parametrizada, con los
// mismos datos de entrada. Solo corre sobre chasqui_w05, y el ataque que
// borra filas se hace dentro de una transacción que termina en ROLLBACK.
//   PGDATABASE=chasqui_w05 node injection.mjs
import pg from 'pg';
import { findCustomerByDni } from './queries.mjs';
import { findCustomerByDniUnsafe } from './unsafe.mjs';

const pool = new pg.Pool();
const countEvents = async (db) =>
  (await db.query('SELECT count(*) AS n FROM shipment_event')).rows[0].n;

try {
  const { rows } = await pool.query('SELECT current_database() AS db');
  if (rows[0].db !== 'chasqui_w05') throw new Error('este script solo corre sobre chasqui_w05');

  const inputs = ['61830975', "' OR '1'='1"];
  for (const dni of inputs) {
    const unsafe = await findCustomerByDniUnsafe(pool, dni);
    const safe = await findCustomerByDni(pool, dni);
    console.log(`dni = ${JSON.stringify(dni)}`);
    console.log(`  vulnerable:    filas = ${unsafe.rowCount}`);
    console.log(`  parametrizada: filas = ${safe.length}`);
  }

  // Dos sentencias en un mismo texto: el protocolo simple las acepta.
  const attack = "'; DELETE FROM shipment_event; --";
  console.log(`dni = ${JSON.stringify(attack)}`);
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    console.log(`  eventos antes:  ${await countEvents(client)}`);
    const results = await findCustomerByDniUnsafe(client, attack);
    console.log(`  la vulnerable devolvió ${results.length} resultados:`,
      results.map((r) => `${r.command} ${r.rowCount}`).join(', '));
    console.log(`  eventos después: ${await countEvents(client)}`);
    await client.query('ROLLBACK');
    console.log(`  eventos tras ROLLBACK: ${await countEvents(client)}`);
  } finally {
    client.release();
  }
  const safe = await findCustomerByDni(pool, attack);
  console.log(`  parametrizada: filas = ${safe.length}, eventos = ${await countEvents(pool)}`);
} finally {
  await pool.end();
}

// Consultas de la semana 5 sobre Chasqui con el driver pg. Cada función
// recibe un pool (o un client) y no lo cierra: quien lo crea, lo cierra.
// Todas usan consultas parametrizadas ($1, $2, ...).

/** Los clientes con ese DNI (cero o uno, porque dni es UNIQUE). */
export async function findCustomerByDni(db, dni) {
  const result = await db.query(
    `SELECT customer_id, dni, first_name, last_name, phone, city
     FROM customer
     WHERE dni = $1`,
    [dni],
  );
  return result.rows;
}

// Lo que el usuario puede elegir para ordenar, y la columna que le
// corresponde. Un nombre de columna no puede ir en $1: se elige de esta lista.
const ROUTE_ORDER = {
  tarifa: 'r.base_fare',
  distancia: 'r.distance_km',
  destino: 'd.name',
};
const DIRECTION = { asc: 'ASC', desc: 'DESC' };

/** Las rutas que salen de una ciudad, en el orden que pidió el usuario. */
export async function routesFrom(db, city, order = 'tarifa', direction = 'asc') {
  if (!Object.hasOwn(ROUTE_ORDER, order) || !Object.hasOwn(DIRECTION, direction)) {
    throw new RangeError(`orden no permitido: ${order} ${direction}`);
  }
  const column = ROUTE_ORDER[order];
  const dir = DIRECTION[direction];
  const result = await db.query(
    `SELECT r.route_id, o.name AS origin, d.name AS dest, r.distance_km, r.base_fare
     FROM route r
       JOIN agency o ON o.agency_id = r.origin_agency_id
       JOIN agency d ON d.agency_id = r.dest_agency_id
     WHERE o.city = $1
     ORDER BY ${column} ${dir}, r.route_id`,
    [city],
  );
  return result.rows;
}

// Errores de PostgreSQL (SQLSTATE) que el programa sabe explicar.
const UNIQUE_VIOLATION = '23505';
const CHECK_VIOLATION = '23514';

/** Error con un mensaje para el usuario; el detalle técnico queda en cause. */
export class CustomerError extends Error {}

/** Registra un cliente y devuelve su customer_id. */
export async function addCustomer(db, { id, dni, firstName, lastName, phone = null, city }) {
  try {
    const result = await db.query(
      `INSERT INTO customer (customer_id, dni, first_name, last_name, phone, city)
       VALUES ($1, $2, $3, $4, $5, $6)
       RETURNING customer_id`,
      [id, dni, firstName, lastName, phone, city],
    );
    return result.rows[0].customer_id;
  } catch (err) {
    if (err.code === UNIQUE_VIOLATION && err.constraint === 'customer_dni_key') {
      throw new CustomerError(`el DNI ${dni} ya está registrado`, { cause: err });
    }
    if (err.code === CHECK_VIOLATION) {
      throw new CustomerError('el DNI o el celular no tienen el formato correcto', { cause: err });
    }
    throw err; // cualquier otro error (conexión caída, etc.) sigue de largo
  }
}

-- Semana 5: consultas recursivas sobre la red de rutas.
-- Cada consulta empieza con una línea "-- name: ..."; las pruebas
-- (sql.test.mjs) leen este archivo y corren cada consulta por su nombre.
-- Para verlas todas: psql -d chasqui_w05 -f recursive.sql

-- name: reachable_from_lima
-- Agencias a las que se llega desde Lima - La Victoria (agency_id 1) con una
-- o más rutas. UNION (sin ALL) descarta lo ya alcanzado, así que termina.
WITH RECURSIVE reachable (agency_id) AS (
    SELECT dest_agency_id
    FROM route
    WHERE origin_agency_id = 1
  UNION
    SELECT r.dest_agency_id
    FROM reachable re
      JOIN route r ON r.origin_agency_id = re.agency_id
)
SELECT count(*) AS agencies
FROM reachable;

-- name: unreachable_from_lima
-- Las agencias que no aparecen en la clausura.
WITH RECURSIVE reachable (agency_id) AS (
    SELECT dest_agency_id FROM route WHERE origin_agency_id = 1
  UNION
    SELECT r.dest_agency_id
    FROM reachable re
      JOIN route r ON r.origin_agency_id = re.agency_id
)
SELECT a.agency_id, a.name
FROM agency a
WHERE a.agency_id NOT IN (SELECT agency_id FROM reachable)
ORDER BY a.agency_id;

-- name: hops_from_lima
-- Menor número de tramos hasta cada agencia. Se guarda el camino en un
-- arreglo para no volver a pasar por una agencia (evita los ciclos).
WITH RECURSIVE trip_path (agency_id, hops, path) AS (
    SELECT dest_agency_id, 1, ARRAY[1, dest_agency_id]
    FROM route
    WHERE origin_agency_id = 1
  UNION ALL
    SELECT r.dest_agency_id, tp.hops + 1, tp.path || r.dest_agency_id
    FROM trip_path tp
      JOIN route r ON r.origin_agency_id = tp.agency_id
    WHERE r.dest_agency_id <> ALL (tp.path)
      AND tp.hops < 4
)
SELECT a.name, min(tp.hops) AS hops
FROM trip_path tp
  JOIN agency a ON a.agency_id = tp.agency_id
GROUP BY a.name
HAVING min(tp.hops) >= 3
ORDER BY hops, a.name;

-- name: cheapest_lima_puno
-- La combinación de rutas más barata (suma de tarifas base) de Lima - La
-- Victoria a Puno - Centro (agency_id 18), con a lo más 4 tramos.
WITH RECURSIVE trip_path (agency_id, hops, fare, path) AS (
    SELECT dest_agency_id, 1, base_fare::numeric, ARRAY[1, dest_agency_id]
    FROM route
    WHERE origin_agency_id = 1
  UNION ALL
    SELECT r.dest_agency_id, tp.hops + 1, tp.fare + r.base_fare, tp.path || r.dest_agency_id
    FROM trip_path tp
      JOIN route r ON r.origin_agency_id = tp.agency_id
    WHERE r.dest_agency_id <> ALL (tp.path)
      AND tp.hops < 4
)
SELECT tp.hops, tp.fare,
       (SELECT string_agg(a.city, ' → ' ORDER BY p.ord)
        FROM unnest(tp.path) WITH ORDINALITY AS p (agency_id, ord)
          JOIN agency a ON a.agency_id = p.agency_id) AS cities
FROM trip_path tp
WHERE tp.agency_id = 18
ORDER BY tp.fare, tp.hops
LIMIT 3;

-- name: cycle_clause
-- Caminos desde Lima - La Victoria con la cláusula CYCLE de PostgreSQL: marca
-- la fila que repite una agencia del camino y no la sigue expandiendo. La
-- parte no recursiva es la propia agencia 1 (0 tramos), para que esa
-- agencia también quede en el camino.
WITH RECURSIVE trip_path (agency_id, hops) AS (
    SELECT 1, 0
  UNION ALL
    SELECT r.dest_agency_id, tp.hops + 1
    FROM trip_path tp
      JOIN route r ON r.origin_agency_id = tp.agency_id
    WHERE tp.hops < 4
) CYCLE agency_id SET is_cycle USING path
SELECT count(*) FILTER (WHERE NOT is_cycle) AS paths,
       count(*) FILTER (WHERE is_cycle) AS cycles,
       count(DISTINCT agency_id) FILTER (WHERE NOT is_cycle) AS agencies
FROM trip_path;

-- name: endless
-- MAL: UNION ALL sin control de ciclos ni de tramos. Como hay rutas de ida
-- y vuelta, la parte recursiva siempre produce filas nuevas y la consulta
-- no termina; aquí la corta el statement_timeout.
SET statement_timeout = '2s';
WITH RECURSIVE trip_path (agency_id, hops) AS (
    SELECT dest_agency_id, 1 FROM route WHERE origin_agency_id = 1
  UNION ALL
    SELECT r.dest_agency_id, tp.hops + 1
    FROM trip_path tp
      JOIN route r ON r.origin_agency_id = tp.agency_id
)
SELECT count(*) FROM trip_path;
RESET statement_timeout;

// 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.
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) {
    queries[parts[i]] = parts[i + 1].trim();
  }
  return queries;
}

-- Semana 5: un trigger que se dispara a sí mismo. NO lo copien.
-- Todo va dentro de una transacción que termina en ROLLBACK, así que no
-- deja nada creado. Para verlo: psql -d chasqui_w05 -f trigger_loop.sql
BEGIN;

-- Cada vez que cambia un envío, "actualiza" su precio... con otro UPDATE
-- sobre la misma tabla, que vuelve a disparar el trigger, y así.
CREATE FUNCTION touch_shipment() RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
  UPDATE shipment SET price = NEW.price WHERE shipment_id = NEW.shipment_id;
  RETURN NULL;
END;
$$;

CREATE TRIGGER touch_shipment_trg
  AFTER UPDATE ON shipment
  FOR EACH ROW
  EXECUTE FUNCTION touch_shipment();

UPDATE shipment SET price = price WHERE shipment_id = 1;

ROLLBACK;

-- Semana 5: triggers sobre shipment.
-- Se cargan con: psql -d chasqui_w05 -f triggers.sql (make lo hace).

-- 1. Log de cambios de estado: una fila por cada envío que cambia de estado.
CREATE TABLE IF NOT EXISTS shipment_status_log (
  log_id      bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  shipment_id integer NOT NULL REFERENCES shipment,
  old_status  text NOT NULL,
  new_status  text NOT NULL,
  changed_at  timestamptz NOT NULL DEFAULT now(),
  changed_by  text NOT NULL DEFAULT current_user
);

CREATE OR REPLACE FUNCTION log_status_change()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
  INSERT INTO shipment_status_log (shipment_id, old_status, new_status)
  VALUES (NEW.shipment_id, OLD.status, NEW.status);
  RETURN NULL;  -- en un trigger AFTER el valor de retorno se ignora
END;
$$;

CREATE OR REPLACE TRIGGER shipment_status_log_trg
  AFTER UPDATE OF status ON shipment
  FOR EACH ROW
  WHEN (OLD.status IS DISTINCT FROM NEW.status)
  EXECUTE FUNCTION log_status_change();

-- 2. BEFORE: al pasar a 'delivered' sin fecha de entrega, se pone la hora
--    actual. El trigger cambia NEW antes de que la fila se guarde.
CREATE OR REPLACE FUNCTION fill_delivered_at()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
  IF NEW.status = 'delivered' AND NEW.delivered_at IS NULL THEN
    NEW.delivered_at := now();
  END IF;
  RETURN NEW;  -- la fila que se va a guardar
END;
$$;

CREATE OR REPLACE TRIGGER shipment_fill_delivered_at_trg
  BEFORE UPDATE OF status ON shipment
  FOR EACH ROW
  EXECUTE FUNCTION fill_delivered_at();

-- 3. FOR EACH STATEMENT: una sola fila por sentencia UPDATE sobre shipment,
--    con cuántas filas cambió. La tabla de transición "changed" tiene las
--    filas nuevas de toda la sentencia.
CREATE TABLE IF NOT EXISTS shipment_update_batch (
  batch_id   bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  run_at     timestamptz NOT NULL DEFAULT now(),
  rows_count integer NOT NULL
);

CREATE OR REPLACE FUNCTION log_update_batch()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
  INSERT INTO shipment_update_batch (rows_count)
  SELECT count(*) FROM changed;
  RETURN NULL;
END;
$$;

CREATE OR REPLACE TRIGGER shipment_update_batch_trg
  AFTER UPDATE ON shipment
  REFERENCING NEW TABLE AS changed
  FOR EACH STATEMENT
  EXECUTE FUNCTION log_update_batch();

// Versión VULNERABLE de findCustomerByDni, solo para ver el ataque en
// injection.mjs y en las pruebas. Arma la consulta pegando el texto del
// usuario: nunca se escribe así.
export async function findCustomerByDniUnsafe(db, dni) {
  return db.query(
    `SELECT customer_id, dni, first_name, last_name, phone, city
     FROM customer
     WHERE dni = '${dni}'`,
  );
}

-- Semana 5: window functions y agregación avanzada.
-- Cada consulta empieza con una línea "-- name: ..."; las pruebas
-- (sql.test.mjs) leen este archivo y corren cada consulta por su nombre.
-- Para verlas todas: psql -d chasqui_w05 -f window.sql

-- name: ties
-- rank, dense_rank y row_number ante empates: las rutas más caras.
SELECT route_id, base_fare,
       rank()       OVER w AS rank,
       dense_rank() OVER w AS dense_rank,
       row_number() OVER w AS row_number
FROM route
WINDOW w AS (ORDER BY base_fare DESC)
ORDER BY base_fare DESC, route_id
LIMIT 10;

-- name: top_customers_by_city
-- Los dos clientes que más gastaron en envíos (como remitentes) en cada
-- ciudad del sur. El ranking se calcula en una CTE y se filtra afuera.
WITH spent AS (
  SELECT c.city, c.customer_id, c.last_name, sum(s.price) AS total,
         rank() OVER (PARTITION BY c.city ORDER BY sum(s.price) DESC) AS pos
  FROM customer c
    JOIN shipment s ON s.sender_id = c.customer_id
  WHERE c.city IN ('Arequipa', 'Cusco', 'Puno', 'Tacna')
  GROUP BY c.city, c.customer_id, c.last_name
)
SELECT city, pos, customer_id, last_name, total
FROM spent
WHERE pos <= 2
ORDER BY city, pos, customer_id;

-- name: running_total
-- Acumulado de ventas de Puno - Centro (18) y Tacna - Centro (20), envío por
-- envío, en la primera semana de febrero. Cada partición empieza de cero.
SELECT origin_agency_id AS agency, shipment_id, created_at, price,
       sum(price) OVER (PARTITION BY origin_agency_id ORDER BY created_at) AS running
FROM shipment
WHERE origin_agency_id IN (18, 20)
  AND created_at >= '2026-02-01' AND created_at < '2026-02-08'
ORDER BY origin_agency_id, created_at;

-- name: event_gaps
-- Tiempo entre un evento y el anterior del mismo envío.
SELECT shipment_id, seq, kind, at,
       at - lag(at) OVER (PARTITION BY shipment_id ORDER BY seq) AS since_previous
FROM shipment_event
WHERE shipment_id = 1
ORDER BY seq;

-- name: moving_average
-- Envíos por día de Lima - La Victoria y el promedio móvil de 7 filas: el
-- día y los 6 anteriores (todos los días tienen envíos). La ventana se
-- calcula en una CTE y se filtra afuera; si el WHERE fuera en la misma
-- consulta, el 1 de marzo no vería los días de febrero.
WITH daily AS (
  SELECT (created_at AT TIME ZONE 'America/Lima')::date AS day, count(*) AS shipments
  FROM shipment
  WHERE origin_agency_id = 1
  GROUP BY 1
), smoothed AS (
  SELECT day, shipments,
         round(avg(shipments) OVER (ORDER BY day ROWS BETWEEN 6 PRECEDING AND CURRENT ROW), 2) AS avg_7
  FROM daily
)
SELECT day, shipments, avg_7
FROM smoothed
WHERE day BETWEEN '2026-03-01' AND '2026-03-07'
ORDER BY day;

-- name: rollup
-- Ingresos por región y ciudad de la agencia de origen, con subtotales por
-- región y el total general (las filas con NULL). Solo regiones del sur.
SELECT a.region, a.city, sum(s.price) AS revenue
FROM shipment s
  JOIN agency a ON a.agency_id = s.origin_agency_id
WHERE a.region IN ('Arequipa', 'Puno', 'Tacna')
GROUP BY ROLLUP (a.region, a.city)
ORDER BY a.region, a.city;

// El error de abrir una transacción con pool.query('BEGIN'): las sentencias
// siguientes pueden ir por otra conexión del pool. Aquí se fuerza el caso
// que en una aplicación ocurre cuando llegan dos pedidos a la vez.
// Solo corre sobre chasqui_w05 (y deja el envío como estaba).
//   PGDATABASE=chasqui_w05 node wrong-transaction.mjs
import pg from 'pg';

const pool = new pg.Pool();
pool.on('connect', (client) => {
  client.on('notice', (msg) => console.log(`  aviso de PostgreSQL: ${msg.message}`));
});

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

try {
  const { rows } = await pool.query('SELECT current_database() AS db');
  if (rows[0].db !== 'chasqui_w05') throw new Error('este script solo corre sobre chasqui_w05');

  const before = await priceOf();
  console.log(`precio del envío ${SHIPMENT} al empezar: ${before}`);

  await pool.query('BEGIN');                 // conexión A; vuelve al pool con la transacción abierta
  const otherRequest = await pool.connect(); // otro pedido se lleva la conexión A
  await pool.query('UPDATE shipment SET price = 0 WHERE shipment_id = $1', [SHIPMENT]); // va por B
  console.log('ROLLBACK...');
  await pool.query('ROLLBACK');              // también por B, donde no hay transacción
  // La conexión A sigue con la transacción abierta: lo que el otro pedido
  // haga ahí queda dentro de ella. La cerramos a mano para dejar todo limpio.
  await otherRequest.query('ROLLBACK');
  otherRequest.release();

  const after = await priceOf();
  console.log(`precio del envío ${SHIPMENT} después del ROLLBACK: ${after}`);

  await pool.query('UPDATE shipment SET price = $2 WHERE shipment_id = $1', [SHIPMENT, before]);
} finally {
  await pool.end();
}

// Pruebas de la Sesión A: consultas desde Node.js, tipos, inyección SQL y
// transacciones. Usan la base de datos chasqui_w05 (make test la crea).
import { test, before, after } from 'node:test';
import assert from 'node:assert/strict';
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import pg from 'pg';
import { findCustomerByDni, routesFrom, addCustomer, CustomerError } from './queries.mjs';
import { findCustomerByDniUnsafe } from './unsafe.mjs';
import { deliverShipment } from './deliver.mjs';

const run = promisify(execFile);
let pool;
before(() => {
  pool = new pg.Pool();
});
after(() => pool.end());

const ATTACKS = ["' OR '1'='1", "'; DELETE FROM shipment_event; --"];

test('la base de datos es chasqui_w05', async () => {
  const { rows } = await pool.query('SELECT current_database() AS db');
  assert.equal(rows[0].db, 'chasqui_w05');
});

test('findCustomerByDni encuentra un DNI conocido', async () => {
  const rows = await findCustomerByDni(pool, '61830975');
  assert.equal(rows.length, 1);
  assert.equal(rows[0].first_name, 'Wilmer');
  assert.equal(rows[0].customer_id, 1);
});

test('findCustomerByDni con un DNI inexistente devuelve []', async () => {
  assert.deepEqual(await findCustomerByDni(pool, '00000000'), []);
});

test('los ataques no le hacen nada a la versión parametrizada', async () => {
  for (const dni of ATTACKS) {
    assert.deepEqual(await findCustomerByDni(pool, dni), []);
  }
  const { rows } = await pool.query('SELECT count(*) AS n FROM shipment_event');
  assert.equal(rows[0].n, '46421');
});

test("la versión vulnerable devuelve todos los clientes con ' OR '1'='1", async () => {
  const result = await findCustomerByDniUnsafe(pool, ATTACKS[0]);
  assert.equal(result.rowCount, 1800);
});

test('la versión vulnerable ejecuta un DELETE inyectado (y el ROLLBACK lo deshace)', async () => {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const results = await findCustomerByDniUnsafe(client, ATTACKS[1]);
    assert.deepEqual(results.map((r) => r.command), ['SELECT', 'DELETE']);
    assert.equal(results[1].rowCount, 46421);
    const { rows } = await client.query('SELECT count(*) AS n FROM shipment_event');
    assert.equal(rows[0].n, '0');
  } finally {
    await client.query('ROLLBACK');
    client.release();
  }
});

test('con parámetros no se pueden mandar dos sentencias', async () => {
  await assert.rejects(
    pool.query('SELECT 1 WHERE $1 = $1; DELETE FROM shipment_event', ['x']),
    { code: '42601', message: 'cannot insert multiple commands into a prepared statement' },
  );
});

test('tipos: integer es number, numeric y bigint son string, timestamptz es Date', async () => {
  const { rows } = await pool.query(
    `SELECT s.shipment_id, s.price, s.created_at, s.delivered_at, count(*) AS events
     FROM shipment s JOIN shipment_event e ON e.shipment_id = s.shipment_id
     WHERE s.shipment_id = $1 GROUP BY s.shipment_id`,
    [1],
  );
  const s = rows[0];
  assert.equal(s.shipment_id, 1);
  assert.equal(s.price, '88.17');
  assert.equal(s.events, '4');
  assert.ok(s.created_at instanceof Date);
  assert.equal(s.created_at.toISOString(), '2026-01-29T00:46:00.000Z');
});

test('un NULL llega como null', async () => {
  const { rows } = await pool.query('SELECT delivered_at FROM shipment WHERE shipment_id = $1', [46]);
  assert.equal(rows[0].delivered_at, null);
});

test('routesFrom ordena con una columna de la lista', async () => {
  const rows = await routesFrom(pool, 'Cusco', 'tarifa', 'desc');
  assert.equal(rows.length, 6);
  assert.equal(rows[0].dest, 'Lima - Los Olivos');
  assert.equal(rows.at(-1).dest, 'Abancay - Centro');
});

test('routesFrom rechaza un orden que no está en la lista', async () => {
  await assert.rejects(routesFrom(pool, 'Cusco', 'base_fare; DROP TABLE route'), RangeError);
  await assert.rejects(routesFrom(pool, 'Cusco', 'tarifa', 'desc; --'), RangeError);
});

test('addCustomer traduce los errores de PostgreSQL', async () => {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const id = await addCustomer(client, {
      id: 5001, dni: '12345678', firstName: 'Rosa', lastName: 'Quispe', city: 'Puno',
    });
    assert.equal(id, 5001);
    await assert.rejects(
      addCustomer(client, { id: 5002, dni: '61830975', firstName: 'X', lastName: 'Y', city: 'Lima' }),
      (err) => err instanceof CustomerError && err.cause.code === '23505'
        && err.message === 'el DNI 61830975 ya está registrado',
    );
  } finally {
    await client.query('ROLLBACK');
    client.release();
  }
  await assert.rejects(
    addCustomer(pool, { id: 5003, dni: '1234', firstName: 'X', lastName: 'Y', city: 'Lima' }),
    (err) => err instanceof CustomerError && err.cause.code === '23514',
  );
  assert.deepEqual(await findCustomerByDni(pool, '12345678'), []);
});

// Los scripts de los apuntes: se corren y se revisa lo que imprimen.
const script = async (file) => (await run('node', [file], { cwd: import.meta.dirname })).stdout;

test('connect.mjs muestra los tipos', async () => {
  const out = await script('connect.mjs');
  assert.match(out, /rowCount: 2/);
  assert.match(out, /price\s+string/);
  assert.match(out, /created_at\s+Date/);
  assert.match(out, /suma con texto: 88\.1764\.60/);
});

test('injection.mjs muestra el ataque y la defensa', async () => {
  const out = await script('injection.mjs');
  assert.match(out, /vulnerable: {4}filas = 1800/);
  assert.match(out, /DELETE 46421/);
  assert.match(out, /eventos tras ROLLBACK: 46421/);
});

test('wrong-transaction.mjs: el ROLLBACK no deshace el UPDATE', async () => {
  const out = await script('wrong-transaction.mjs');
  assert.match(out, /there is no transaction in progress/);
  assert.match(out, /después del ROLLBACK: 0\.00/);
  const { rows } = await pool.query('SELECT price FROM shipment WHERE shipment_id = $1', [46]);
  assert.equal(rows[0].price, '153.50'); // el script lo deja como estaba
});

// Estas cambian datos: van al final, después de los scripts.
test('deliverShipment cambia el envío y agrega el evento', async () => {
  const at = new Date('2026-06-29T15:00:00Z');
  const seq = await deliverShipment(pool, 242, 15, at);
  const { rows } = await pool.query(
    `SELECT s.status, s.delivered_at, e.kind, e.agency_id
     FROM shipment s JOIN shipment_event e ON e.shipment_id = s.shipment_id AND e.seq = $2
     WHERE s.shipment_id = $1`,
    [242, seq],
  );
  assert.equal(rows[0].status, 'delivered');
  assert.equal(rows[0].delivered_at.toISOString(), at.toISOString());
  assert.equal(rows[0].kind, 'delivered');
});

test('si el segundo INSERT falla, deliverShipment no deja el envío a medias', async () => {
  await assert.rejects(deliverShipment(pool, 248, 999), { code: '23503' });
  const { rows } = await pool.query(
    `SELECT status, delivered_at, (SELECT max(seq) FROM shipment_event WHERE shipment_id = $1) AS last
     FROM shipment WHERE shipment_id = $1`,
    [248],
  );
  assert.deepEqual(rows[0], { status: 'in_transit', delivered_at: null, last: 3 });
});

test('deliverShipment no entrega dos veces', async () => {
  await assert.rejects(deliverShipment(pool, 242, 15), /no está en tránsito/);
});

// Pruebas de la Sesión B: funciones, triggers, consultas recursivas y window
// functions. Corren el texto de recursive.sql y window.sql; functions.sql y
// triggers.sql ya están cargados (make los carga). Base de datos chasqui_w05.
import { test, before, after } from 'node:test';
import assert from 'node:assert/strict';
import { readFileSync } from 'node:fs';
import pg from 'pg';
import { namedQueries } from './sqlfile.mjs';

const recursive = namedQueries('./recursive.sql');
const windows = namedQueries('./window.sql');
let pool;
before(() => {
  pool = new pg.Pool();
});
after(() => pool.end());

const all = async (text, values) => (await pool.query(text, values)).rows;

// Corre fn con un client dentro de una transacción que siempre termina en ROLLBACK.
async function inRollback(fn) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    return await fn(client);
  } finally {
    await client.query('ROLLBACK');
    client.release();
  }
}

// ---------------------------------------------------------------- funciones

test('shipment_fare reproduce el precio de los 12 000 envíos', async () => {
  const [row] = await all(
    `SELECT count(*) AS total,
            count(*) FILTER (WHERE s.price = shipment_fare(r.route_id, s.weight_kg, s.declared_value)) AS same
     FROM shipment s
       JOIN route r ON r.origin_agency_id = s.origin_agency_id AND r.dest_agency_id = s.dest_agency_id`,
  );
  assert.deepEqual(row, { total: '12000', same: '12000' });
});

test('shipment_fare y quote_fare', async () => {
  const [row] = await all('SELECT shipment_fare(1, 10) AS a, quote_fare(1, 7, 10, 500) AS b');
  assert.deepEqual(row, { a: '42.00', b: '47.00' });
  const [none] = await all('SELECT shipment_fare(9999, 10) AS fare');
  assert.equal(none.fare, null);
});

test('quote_fare avisa con RAISE EXCEPTION', async () => {
  await assert.rejects(all('SELECT quote_fare(1, 29, 10)'), {
    code: 'P0002', message: 'no hay ruta de la agencia 1 a la agencia 29',
  });
  await assert.rejects(all('SELECT quote_fare(1, 7, -2)'), { code: '23514', message: 'peso inválido: -2' });
});

test('routes_from se usa en el FROM', async () => {
  const rows = await all('SELECT dest, shipments FROM routes_from($1) ORDER BY shipments DESC, dest', ['Cusco']);
  assert.equal(rows.length, 6);
  assert.deepEqual(rows[0], { dest: 'Lima - La Victoria', shipments: '138' });
});

test('CALL refresh_agency_stats llena la tabla de resumen', async () => {
  await pool.query('CALL refresh_agency_stats()');
  const [row] = await all('SELECT count(*) AS n, sum(shipments) AS s FROM agency_stats');
  assert.deepEqual(row, { n: '29', s: '12000' });
  const [lima] = await all('SELECT shipments, revenue FROM agency_stats WHERE agency_id = 1');
  assert.deepEqual(lima, { shipments: 1823, revenue: '116713.45' });
});

test('el COMMIT del procedimiento falla dentro de una transacción abierta', async () => {
  await inRollback(async (client) => {
    await assert.rejects(client.query('CALL refresh_agency_stats()'), {
      code: '2D000', message: 'invalid transaction termination',
    });
  });
});

// ---------------------------------------------------------------- triggers

test('BEFORE llena delivered_at y AFTER deja el cambio en el log', async () => {
  await inRollback(async (client) => {
    const upd = await client.query(
      `UPDATE shipment SET status = 'delivered' WHERE shipment_id = 46
       RETURNING delivered_at, delivered_at = now() AS is_now`,
    );
    assert.equal(upd.rows[0].is_now, true);
    const log = await client.query(
      'SELECT old_status, new_status FROM shipment_status_log WHERE shipment_id = 46',
    );
    assert.deepEqual(log.rows, [{ old_status: 'in_transit', new_status: 'delivered' }]);
  });
});

test('el WHEN del trigger AFTER ignora los UPDATE que no cambian el estado', async () => {
  await inRollback(async (client) => {
    await client.query("UPDATE shipment SET status = status WHERE shipment_id = 2");
    await client.query('UPDATE shipment SET price = price WHERE shipment_id = 3');
    const log = await client.query('SELECT count(*) AS n FROM shipment_status_log WHERE shipment_id IN (2, 3)');
    assert.equal(log.rows[0].n, '0');
  });
});

test('el trigger FOR EACH STATEMENT se dispara una vez por sentencia', async () => {
  await inRollback(async (client) => {
    const before = (await client.query('SELECT coalesce(max(batch_id), 0) AS m FROM shipment_update_batch')).rows[0].m;
    const upd = await client.query(
      "UPDATE shipment SET status = 'returned' WHERE status = 'in_transit' AND origin_agency_id = 1",
    );
    await client.query('UPDATE shipment SET price = price WHERE shipment_id = -1'); // 0 filas
    const rows = (await client.query(
      'SELECT rows_count FROM shipment_update_batch WHERE batch_id > $1 ORDER BY batch_id', [before],
    )).rows;
    assert.deepEqual(rows, [{ rows_count: upd.rowCount }, { rows_count: 0 }]);
    const log = await client.query(
      "SELECT count(*) AS n FROM shipment_status_log WHERE new_status = 'returned'",
    );
    assert.equal(Number(log.rows[0].n), upd.rowCount);
  });
});

test('trigger_loop.sql termina en stack depth limit exceeded', async () => {
  const text = readFileSync(new URL('./trigger_loop.sql', import.meta.url), 'utf8');
  const client = await pool.connect();
  try {
    await assert.rejects(client.query(text), { code: '54001', message: 'stack depth limit exceeded' });
  } finally {
    await client.query('ROLLBACK');
    client.release();
  }
  const [row] = await all("SELECT count(*) AS n FROM pg_trigger WHERE tgname = 'touch_shipment_trg'");
  assert.equal(row.n, '0');
});

// ---------------------------------------------------------------- recursivas

test('desde Lima - La Victoria se llega a 28 agencias; no a Iquitos', async () => {
  assert.deepEqual(await all(recursive.reachable_from_lima), [{ agencies: '28' }]);
  assert.deepEqual(await all(recursive.unreachable_from_lima), [{ agency_id: 29, name: 'Iquitos - Centro' }]);
});

test('las agencias más lejanas en tramos', async () => {
  assert.deepEqual(await all(recursive.hops_from_lima), [
    { name: 'Pucallpa - Centro', hops: 3 },
    { name: 'Puno - Centro', hops: 3 },
    { name: 'Tacna - Centro', hops: 3 },
  ]);
});

test('el camino más barato de Lima a Puno', async () => {
  const rows = await all(recursive.cheapest_lima_puno);
  assert.deepEqual(rows[0], { hops: 3, fare: '66.50', cities: 'Lima → Cusco → Juliaca → Puno' });
});

test('la cláusula CYCLE', async () => {
  assert.deepEqual(await all(recursive.cycle_clause), [{ paths: '1018', cycles: '604', agencies: '28' }]);
});

test('sin control de ciclos la recursión no termina (la corta el timeout)', async () => {
  const client = await pool.connect();
  try {
    await assert.rejects(client.query(recursive.endless), { code: '57014' });
  } finally {
    client.release();
  }
});

// ---------------------------------------------------------------- ventanas

test('rank, dense_rank y row_number ante empates', async () => {
  const rows = await all(windows.ties);
  assert.deepEqual(rows.map((r) => `${r.rank},${r.dense_rank}`), [
    '1,1', '1,1', '1,1', '1,1', '1,1', '1,1', '7,2', '7,2', '9,3', '9,3',
  ]);
  // row_number numera de 1 a 10 sin repetir, en un orden que no se puede predecir.
  assert.deepEqual(rows.map((r) => Number(r.row_number)).sort((a, b) => a - b), [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
});

test('los dos que más gastaron en cada ciudad del sur', async () => {
  const rows = await all(windows.top_customers_by_city);
  assert.equal(rows.length, 8);
  assert.deepEqual(rows[0], {
    city: 'Arequipa', pos: '1', customer_id: 1637, last_name: 'Condori Ramírez', total: '3362.86',
  });
});

test('el acumulado empieza de cero en cada partición', async () => {
  const rows = await all(windows.running_total);
  const puno = rows.filter((r) => r.agency === 18);
  const tacna = rows.filter((r) => r.agency === 20);
  assert.equal(puno.at(-1).running, '308.31');
  assert.equal(tacna[0].running, tacna[0].price);
  assert.equal(tacna.at(-1).running, '448.77');
});

test('lag da el tiempo entre eventos', async () => {
  const rows = await all(windows.event_gaps);
  assert.equal(rows[0].since_previous, null);
  // interval llega como un objeto PostgresInterval
  const gap = rows[2].since_previous;
  assert.deepEqual([gap.days ?? 0, gap.hours, gap.minutes], [0, 4, 30]);
});

test('promedio móvil de 7 días', async () => {
  const rows = await all(windows.moving_average);
  assert.equal(rows.length, 7);
  assert.deepEqual(rows[0], { day: rows[0].day, shipments: '9', avg_7: '9.71' });
});

test('ROLLUP: el total general es la suma de los subtotales', async () => {
  const rows = await all(windows.rollup);
  const total = rows.find((r) => r.region === null);
  assert.equal(total.revenue, '88680.55');
  const subtotals = rows.filter((r) => r.region !== null && r.city === null);
  assert.equal(subtotals.map((r) => Number(r.revenue)).reduce((a, b) => a + b).toFixed(2), '88680.55');
});

test('una window function no puede ir en el WHERE', async () => {
  await assert.rejects(
    all('SELECT route_id FROM route WHERE rank() OVER (ORDER BY base_fare DESC) <= 3'),
    { code: '42P20', message: 'window functions are not allowed in WHERE' },
  );
});

# Semana 5: SQL desde Node.js con pg, funciones, triggers, consultas
# recursivas y window functions. Se usa dentro del contenedor del curso.
#
#   make          crea la base de datos chasqui_w05 con los datos de ejemplo
#                 y le carga las funciones y los triggers de la semana
#   make test     la vuelve a crear y corre las pruebas (*.test.mjs)
#   make clean    borra chasqui_w05

DB = chasqui_w05
PSQL = psql -X -q -v ON_ERROR_STOP=1 -d $(DB)

.PHONY: all test clean

all:
	chasqui-load $(DB)
	$(PSQL) -f functions.sql -f triggers.sql

# Los archivos de prueba cambian datos: se corren uno después del otro.
test: all
	PGDATABASE=$(DB) node --test --test-concurrency=1 --test-reporter=spec

clean:
	dropdb --if-exists $(DB)