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

En este proyecto vas a crear un sistema de auditoría que registra todos los cambios (INSERT, UPDATE, DELETE) en tablas seleccionadas, guardando los estados antiguo y nuevo en una tabla de historial. Será práctico, robusto y escalable: usaremos JSONB para flexibilidad, triggers en PL/pgSQL y patrones para particionar/retener datos.

Requisitos previos

• PostgreSQL 12+ (se aprovecha JSONB y particionado nativo).
• Acceso a psql o a una herramienta que ejecute SQL.
• Conocimientos básicos de SQL y triggers.

Estructura del proyecto

audit-project/
├─ sql/
│  ├─ 01_create_schema.sql
│  ├─ 02_create_tables.sql
│  ├─ 03_audit_function_and_triggers.sql
│  └─ 04_queries_and_maintenance.sql
└─ README.md

Objetivos funcionales

• Registrar cada cambio con metadata: operación, usuario, txid, timestamp.
• Almacenar old (antes) y new (después) como JSONB.
• Soportar consultas para recuperar el historial de una fila y reconstruir su estado en un tiempo dado.
• Plan para retencción y particionado para producción.

1) Crear esquema y tablas

Archivo: sql/01_create_schema.sql

-- Crear esquema dedicado
CREATE SCHEMA IF NOT EXISTS app;
CREATE SCHEMA IF NOT EXISTS audit;

Archivo: sql/02_create_tables.sql

-- Tabla de ejemplo: users
CREATE TABLE IF NOT EXISTS app.users (
  id SERIAL PRIMARY KEY,
  email TEXT NOT NULL UNIQUE,
  name TEXT,
  data JSONB DEFAULT '{}'::jsonb,
  created_at TIMESTAMPTZ DEFAULT now(),
  updated_at TIMESTAMPTZ DEFAULT now()
);

-- Tabla de historial (audit trail). Usamos particionado por rango de tiempo en producción.
CREATE TABLE IF NOT EXISTS audit.audit_log (
  id BIGSERIAL PRIMARY KEY,
  schema_name TEXT NOT NULL,
  table_name TEXT NOT NULL,
  operation CHAR(1) NOT NULL, -- I, U, D
  row_id TEXT,                -- clave primaria como texto (convencional)
  data_old JSONB,             -- estado previo
  data_new JSONB,             -- estado posterior
  changed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  changed_by TEXT,           -- rol/usuario que realizó la operación
  tx_id BIGINT                -- transaction id: txid_current()
);

-- Índices útiles
CREATE INDEX IF NOT EXISTS audit_log_idx_table_time ON audit.audit_log(table_name, changed_at DESC);
CREATE INDEX IF NOT EXISTS audit_log_idx_row ON audit.audit_log(table_name, row_id);

Por qué este diseño

• JSONB permite guardar esquemas cambiantes sin bloquear migraciones.
• row_id como texto hace genérico el campo PK (SERIAL, UUID, compuesto, etc.).
• Índices por table & row permiten consultas rápidas por entidad.

2) Función de auditoría y triggers

Archivo: sql/03_audit_function_and_triggers.sql

-- Función que será llamada por triggers
CREATE OR REPLACE FUNCTION audit.audit_if_changes()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $func$
DECLARE
  v_old JSONB;
  v_new JSONB;
  v_row_id TEXT;
  v_user TEXT := current_user; -- o session_user según necesidad
BEGIN
  -- Construir JSONB de OLD/NEW si existen
  IF TG_OP = 'DELETE' THEN
    v_old := to_jsonb(OLD) - 'xmin' - 'xmax';
    v_new := NULL;
    v_row_id := COALESCE(OLD.id::text, NULL);
  ELSIF TG_OP = 'INSERT' THEN
    v_old := NULL;
    v_new := to_jsonb(NEW) - 'xmin' - 'xmax';
    v_row_id := COALESCE(NEW.id::text, NULL);
  ELSIF TG_OP = 'UPDATE' THEN
    v_old := to_jsonb(OLD) - 'xmin' - 'xmax';
    v_new := to_jsonb(NEW) - 'xmin' - 'xmax';
    v_row_id := COALESCE(NEW.id::text, NULL);
    -- Si no hubo cambios reales en contenido, evitar insertar audit
    IF v_old = v_new THEN
      RETURN NEW; -- no registrar
    END IF;
  ELSE
    RETURN NULL; -- no debería pasar
  END IF;

  -- Insertar en la tabla de auditoría
  INSERT INTO audit.audit_log(
    schema_name, table_name, operation, row_id, data_old, data_new, changed_at, changed_by, tx_id
  ) VALUES (
    TG_TABLE_SCHEMA::text,
    TG_TABLE_NAME::text,
    substr(TG_OP,1,1),
    v_row_id,
    v_old,
    v_new,
    now(),
    v_user,
    txid_current()
  );

  -- Retornar la fila para continuar la operación
  IF TG_OP = 'DELETE' THEN
    RETURN OLD;
  ELSE
    RETURN NEW;
  END IF;
END;
$func$;

-- Crear trigger en la tabla de ejemplo
CREATE TRIGGER users_audit_trg
AFTER INSERT OR UPDATE OR DELETE ON app.users
FOR EACH ROW EXECUTE FUNCTION audit.audit_if_changes();

Explicación y decisiones

• Usamos AFTER triggers para garantizar que los cambios se aplicaron (y evitar inconsistencias con rollbacks); sin embargo, ten en cuenta que si la transacción falla, el INSERT en audit también se revertirá (bueno para consistencia pero puede ser costoso).
• to_jsonb(ROW) convierte la fila completa. Restamos xmin/xmax por claridad (opcional).
• Se comprueba si OLD = NEW para evitar ruido en actualizaciones que no cambian contenido real.
• substr(TG_OP,1,1) guarda 'I','U','D' para ahorrar espacio.

3) Consultas útiles

Archivo: sql/04_queries_and_maintenance.sql

-- 1) Ver historial completo de un usuario por id
SELECT *
FROM audit.audit_log
WHERE table_name = 'users' AND row_id = '42'
ORDER BY changed_at DESC;

-- 2) Reconstruir estado de la fila en un punto del tiempo
-- Obtener la última entrada antes de t
WITH last_before AS (
  SELECT data_new, data_old
  FROM audit.audit_log
  WHERE table_name = 'users' AND row_id = '42' AND changed_at <= '2026-02-01T12:00:00Z'
  ORDER BY changed_at DESC
  LIMIT 1
)
SELECT COALESCE(data_new, data_old) AS state_at_time FROM last_before;

-- 3) Encontrar filas que cambiaron en las últimas 24h
SELECT DISTINCT table_name, row_id
FROM audit.audit_log
WHERE changed_at >= now() - interval '24 hours';

4) Mantenimiento y rendimiento

• En producción, particiona audit.audit_log por rango de changed_at (mensual o diario según volumen).
• Considera compresión o almacenamiento en un tablespace más barato para datos antiguos.
• Agrega un job de retención: mover/archivar particiones antiguas a otro sistema (S3, data lake) y DROP PARTITION localmente.
• Si el volumen es muy alto, usar logical decoding / CDC (Debezium) puede ser preferible a triggers por menor overhead y mayor escalabilidad.

Aspectos de seguridad y permisos

• La función se ejecuta con los privilegios del invocador por defecto. Si no quieres que los usuarios con permisos sobre la tabla puedan insertar directamente en audit, crea la función con SECURITY DEFINER y asigna ownership a un rol administración.
• Evita exponer la tabla audit a roles no administrativos.
• Registra changed_by con current_user o con una columna de sesión si tu app pasa el usuario real mediante set_config('app.current_user', ...).

Pruebas rápidas

-- Insertar ejemplo
INSERT INTO app.users(email, name) VALUES ('maria@ejemplo.com','María');
-- Actualizar
UPDATE app.users SET name = 'María A' WHERE email='maria@ejemplo.com';
-- Borrar
DELETE FROM app.users WHERE email='maria@ejemplo.com';

-- Consultar log
SELECT * FROM audit.audit_log WHERE table_name='users' ORDER BY changed_at;

Qué mejorar después

• Soporte para claves primarias compuestas: en vez de asumir columna id, serializa la PK combinada (ej: row_to_json(ROW(OLD.*)) con claves seleccionadas).
• Implementar particionado automático por fecha y una política de retención (pg_cron, cron externo).
• Hacer una vista materializada o espejo en PostgreSQL para consultas analíticas eficientes.

Consejo avanzado: si necesitas auditoría distribuida o con garantía de entrega fuera de la base (por ejemplo replicar cambios a un sistema de búsqueda o a un data lake), evalúa cambiar del enfoque de triggers a CDC (logical decoding) con Debezium o pg_recvlogical; estos enfoques reducen el overhead en escrituras y permiten consumir el stream de cambios desde gestores externos.