Auditoría de cambios en PostgreSQL: historial completo con triggers y JSONB

Proyecto práctico: construimos un sistema de auditoría genérico en PostgreSQL que registra INSERT/UPDATE/DELETE en una tabla audit.audit_log. Usamos triggers genéricos y JSONB para almacenar old y new de cada fila y un objeto con la(s) clave(s) primaria(s). Te dejo los SQL completos, un ejemplo de integración desde Node.js y las razones detrás de cada decisión.

Requisitos previos

• PostgreSQL 12+ (funciona también en 11, pero algunas funciones JSONB son más cómodas en versiones recientes).
• psql o cliente equivalente para ejecutar scripts SQL.
• Node.js 14+ y npm (opcional, para el ejemplo de integración).

Estructura de carpetas

audit_project/
  sql/
    01_schema.sql
    02_functions.sql
    03_triggers.sql
  app/
    package.json
    index.js

Archivo: sql/01_schema.sql

-- Crea esquema y tablas de ejemplo
CREATE SCHEMA IF NOT EXISTS audit;

-- Tabla de ejemplo que queremos auditar
CREATE TABLE IF NOT EXISTS public.users (
  id         BIGSERIAL PRIMARY KEY,
  username   TEXT NOT NULL UNIQUE,
  email      TEXT,
  active     BOOLEAN DEFAULT true,
  updated_at TIMESTAMPTZ DEFAULT now()
);

-- Tabla de auditoría genérica
CREATE TABLE IF NOT EXISTS audit.audit_log (
  id          BIGSERIAL PRIMARY KEY,
  table_name  TEXT NOT NULL,
  operation   TEXT NOT NULL, -- INSERT|UPDATE|DELETE
  primary_key JSONB,        -- {"id": 123}
  old_data    JSONB,        -- snapshot before change (for DELETE/UPDATE)
  new_data    JSONB,        -- snapshot after change (for INSERT/UPDATE)
  changed_by  TEXT,         -- contexto: app user, db role, etc.
  changed_at  TIMESTAMPTZ DEFAULT now()
);

-- Index para consultas por tabla y tiempo
CREATE INDEX IF NOT EXISTS audit_log_table_time_idx ON audit.audit_log(table_name, changed_at DESC);

Archivo: sql/02_functions.sql

-- Función genérica de auditoría. Recibe en TG_ARGV los nombres de las columnas que forman la PK.
CREATE OR REPLACE FUNCTION audit.if_modified_func() RETURNS trigger AS $$
DECLARE
  pk jsonb := '{}';
  i integer;
  colname text;
  src jsonb;
  val jsonb;
BEGIN
  -- Elegir si leer OLD o NEW para extraer valores de PK (DELETE -> OLD, INSERT -> NEW, UPDATE -> NEW)
  IF TG_OP = 'DELETE' THEN
    src := to_jsonb(OLD);
  ELSE
    src := to_jsonb(NEW);
  END IF;

  -- Construir objeto JSON con las columnas PK indicadas en los argumentos del trigger
  FOR i IN 0..TG_NARGS-1 LOOP
    colname := TG_ARGV[i];
    val := src -> colname; -- puede ser null si la columna no existe o es null
    pk := pk || jsonb_build_object(colname, val);
  END LOOP;

  INSERT INTO audit.audit_log(table_name, operation, primary_key, old_data, new_data, changed_by)
  VALUES (
    TG_TABLE_NAME,
    TG_OP,
    pk,
    CASE WHEN TG_OP = 'DELETE' THEN to_jsonb(OLD) ELSE NULL END,
    CASE WHEN TG_OP = 'INSERT' THEN to_jsonb(NEW) WHEN TG_OP = 'UPDATE' THEN to_jsonb(NEW) ELSE NULL END,
    -- leer valor de sesión 'audit.current_user' si existe; NULL en otro caso
    current_setting('audit.current_user', true)
  );

  -- Retornar la fila apropiada
  IF TG_OP = 'DELETE' THEN
    RETURN OLD;
  ELSE
    RETURN NEW;
  END IF;
END;
$$ LANGUAGE plpgsql VOLATILE SECURITY DEFINER;

Archivo: sql/03_triggers.sql

-- Crear trigger en la tabla de ejemplo (pasamos 'id' como columna PK)
DROP TRIGGER IF EXISTS users_audit ON public.users;
CREATE TRIGGER users_audit
  AFTER INSERT OR UPDATE OR DELETE ON public.users
  FOR EACH ROW EXECUTE FUNCTION audit.if_modified_func('id');

Por qué estas decisiones

• Usar JSONB para old_data/new_data te da flexibilidad: no necesitas cambiar el esquema de auditoría al modificar las tablas auditadas.
• Almacenar las claves primarias en un JSON permite reconstruir la fila auditada o buscar fácilmente los eventos de una entidad concreta.
• Función genérica con argumentos TG_ARGV: soporta diferentes tablas y claves primarias sin duplicar lógica.
• Se usa current_setting('audit.current_user', true) para que la aplicación pueda inyectar el usuario/actor en la sesión sin crear columnas extra en cada tabla.

Ejemplo: integración mínima en Node.js

app/index.js y package.json mínimos para ejecutar un par de operaciones y consultar la auditoría.

app/package.json

{
  "name": "audit-example",
  "version": "1.0.0",
  "main": "index.js",
  "dependencies": {
    "pg": "^8.0.0"
  }
}

app/index.js

const { Client } = require("pg");

async function main() {
  const client = new Client({
    connectionString: process.env.DATABASE_URL || "postgresql://postgres:postgres@localhost:5432/postgres"
  });

  await client.connect();

  // Inyecta el contexto de aplicación: quién hace el cambio
  await client.query("SET LOCAL audit.current_user = 'webapp:alice'");

  // INSERT
  const insertRes = await client.query(
    "INSERT INTO public.users (username, email) VALUES ($1, $2) RETURNING id", ["alice", "alice@example.com"]
  );
  const userId = insertRes.rows[0].id;

  // UPDATE
  await client.query("UPDATE public.users SET email = $1, updated_at = now() WHERE id = $2", ["alice@newdomain.com", userId]);

  // DELETE
  await client.query("DELETE FROM public.users WHERE id = $1", [userId]);

  // Consultar auditoría para la tabla users
  const audit = await client.query("SELECT id, operation, primary_key, old_data, new_data, changed_by, changed_at FROM audit.audit_log WHERE table_name = 'users' ORDER BY changed_at");
  console.log(JSON.stringify(audit.rows, null, 2));

  await client.end();
}

main().catch(err => {
  console.error(err);
  process.exit(1);
});

Cómo reconstruir el estado de una fila en un momento concreto

Si tienes los eventos ordenados por changed_at, puedes aplicar las transformaciones (take new_data para INSERT/UPDATE, y old_data para DELETE) para obtener el estado en cualquier punto. Ejemplo para obtener el último new_data antes de una fecha:

SELECT new_data
FROM audit.audit_log
WHERE table_name = 'users'
  AND (primary_key ->> 'id')::bigint = 123
  AND changed_at <= '2026-01-01T12:00:00Z'
ORDER BY changed_at DESC
LIMIT 1;

Consideraciones de rendimiento y operacionales

• Triggers AFTER son síncronos: cada escritura en la tabla genera una inserción adicional en audit.audit_log. En tablas con alto volumen puedes notar latencia.
• Opciones para escala: particionar la tabla de auditoría por rango temporal, hacer compresión de older partitions, o escribir a un topic/log asíncrono (logical decoding, WAL shipping o una cola externa).
• Indice por (table_name, changed_at) y por valores de PK si vas a consultar por entidad individual frecuentemente.
• Para evitar que aplicaciones maliciosas borren la auditoría, restringe permisos: sólo roles de administración deberían poder DELETE/UPDATE en audit.audit_log.

Extensiones y siguientes pasos

• Agregar columna transaction_id o usar txid_current() para agrupar eventos por transacción.
• Calcular y guardar un diff explícito de campos modificados para queries rápidas de «qué cambió». Puedes generar el diff en la función PL/pgSQL comparando to_jsonb(OLD) y to_jsonb(NEW).
• Si necesitas mínima latencia en escritura, considera usar logical decoding o pg_recvlogical para exportar cambios y procesarlos en lote fuera del camino crítico.

Consejo avanzado: usa políticas de seguridad y roles separados para escribir al esquema audit y deshabilita la capacidad de borrar registros de auditoría por parte de aplicaciones normales. Además, si la integridad de la auditoría es crítica, incluye en cada fila un hash firmado (HMAC) calculado a partir de primary_key, old_data, new_data y changed_at para detectar manipulaciones posteriores a la base de datos.