Proyecto práctico: Historial y auditoría de cambios en PostgreSQL

En este tutorial vas a construir un sistema de auditoría que registra cada cambio (INSERT/UPDATE/DELETE) en una tabla principal usando funciones y triggers en PostgreSQL. Además guardaremos la versión completa de la fila en jsonb, campos cambiados, el usuario que hizo el cambio y un transaction id para poder restaurar versiones pasadas.

Requisitos previos

• PostgreSQL >= 11 (funciones PL/pgSQL, jsonb y txid_current() usados)
• Acceso a psql o alguna herramienta de administración (pgAdmin, DBeaver)
• Permisos para crear esquemas, tablas, funciones y triggers

Estructura del proyecto

sql-audit-project/
├─ sql/
│  ├─ 01_schema.sql           -- tabla principal
│  ├─ 02_audit_table.sql      -- tabla de historial
│  ├─ 03_audit_functions.sql  -- funciones PL/pgSQL
│  ├─ 04_triggers.sql         -- triggers que llaman a las funciones
│  ├─ 05_sample_data.sql      -- datos de ejemplo
│  └─ 06_restore_examples.sql -- ejemplos para restaurar versiones
└─ README.md

Por qué este enfoque

• Usar una tabla de historial separada mantiene la tabla principal ligera y permite mantener múltiples versiones por fila.
• Guardar la fila como jsonb permite flexibilidad ante cambios de esquema y facilita reconstruir versiones antiguas.
• Incluir transaction id (txid_current()) y timestamp permite correlacionar cambios y recuperar estados consistentes.
• Registrar "changed_fields" ayuda para auditoría rápida y minimiza cantidad de datos a inspeccionar.

Código: 01_schema.sql (tabla principal)

-- 01_schema.sql
CREATE SCHEMA IF NOT EXISTS app;

CREATE TABLE IF NOT EXISTS app.products (
    id BIGSERIAL PRIMARY KEY,
    sku TEXT NOT NULL UNIQUE,
    name TEXT NOT NULL,
    description TEXT,
    price NUMERIC(10,2) NOT NULL DEFAULT 0,
    available BOOLEAN NOT NULL DEFAULT TRUE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- trigger para mantener updated_at
CREATE OR REPLACE FUNCTION app.set_updated_at()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
  NEW.updated_at = now();
  RETURN NEW;
END;
$$;

DROP TRIGGER IF EXISTS set_updated_at_products ON app.products;
CREATE TRIGGER set_updated_at_products
BEFORE UPDATE ON app.products
FOR EACH ROW EXECUTE FUNCTION app.set_updated_at();

Código: 02_audit_table.sql (tabla de historial)

-- 02_audit_table.sql
CREATE TABLE IF NOT EXISTS app.audit_history (
    id BIGSERIAL PRIMARY KEY,
    table_name TEXT NOT NULL,
    record_id BIGINT,                -- id de la fila en la tabla principal
    op CHAR(1) NOT NULL,             -- 'I' insert, 'U' update, 'D' delete
    changed_by TEXT,                 -- puede venir de current_user o variable de sesión
    changed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    txid BIGINT NOT NULL,            -- txid_current()
    data jsonb,                      -- estado completo de la fila (NEW para I/U, OLD para D)
    changed_fields text[],           -- lista de campos cambiados (para UPDATE)
    reason TEXT                       -- campo opcional para notas
);

CREATE INDEX IF NOT EXISTS idx_audit_table_record ON app.audit_history(table_name, record_id);
CREATE INDEX IF NOT EXISTS idx_audit_txid ON app.audit_history(txid);
CREATE INDEX IF NOT EXISTS idx_audit_changed_at ON app.audit_history(changed_at);

Código: 03_audit_functions.sql (funciones de auditoría)

-- 03_audit_functions.sql
-- Función util para obtener usuario de aplicación (se puede poner en variable de sesión)
CREATE OR REPLACE FUNCTION app.get_audit_user()
RETURNS TEXT LANGUAGE sql AS $$
SELECT current_setting('app.current_user', true)  -- devuelve NULL si no está seteado
       COALESCE(current_setting('app.current_user', true), current_user);
$$;

-- Función de trigger genérica
CREATE OR REPLACE FUNCTION app.audit_if_needed()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
DECLARE
  v_op CHAR(1) := substr(TG_OP,1,1); -- 'I', 'U', 'D'
  v_record jsonb;
  v_changed_fields text[] := ARRAY[]::text[];
  v_user TEXT := app.get_audit_user();
BEGIN
  IF v_op = 'I' THEN
    v_record := to_jsonb(NEW);
  ELSIF v_op = 'D' THEN
    v_record := to_jsonb(OLD);
  ELSIF v_op = 'U' THEN
    v_record := to_jsonb(NEW);
    -- detectar campos cambiados comparando OLD y NEW
    PERFORM 1; -- placeholder
    -- Construimos lista simple de campos cambiados
    FOR i IN SELECT jsonb_object_keys(to_jsonb(OLD)) LOOP
      IF (to_jsonb(OLD)->>i) IS DISTINCT FROM (to_jsonb(NEW)->>i) THEN
        v_changed_fields := array_append(v_changed_fields, i);
      END IF;
    END LOOP;
  END IF;

  INSERT INTO app.audit_history(
    table_name, record_id, op, changed_by, changed_at, txid, data, changed_fields
  ) VALUES (
    TG_TABLE_NAME,
    COALESCE((CASE WHEN v_op = 'I' THEN NEW.id WHEN v_op = 'U' THEN NEW.id ELSE OLD.id END), NULL),
    v_op,
    v_user,
    now(),
    txid_current(),
    v_record,
    CASE WHEN array_length(v_changed_fields,1) IS NULL THEN NULL ELSE v_changed_fields END
  );

  IF TG_OP = 'DELETE' THEN
    RETURN OLD;
  ELSE
    RETURN NEW;
  END IF;
END;
$$;

Código: 04_triggers.sql (adjuntar triggers)

-- 04_triggers.sql
DROP TRIGGER IF EXISTS audit_products ON app.products;
CREATE TRIGGER audit_products
AFTER INSERT OR UPDATE OR DELETE ON app.products
FOR EACH ROW EXECUTE FUNCTION app.audit_if_needed();

Código: 05_sample_data.sql (ejemplos de uso)

-- 05_sample_data.sql
-- Opcional: establecer variable de sesión para detectar 'usuario de la app'
SET LOCAL app.current_user = 'alice@example.com';

INSERT INTO app.products (sku, name, description, price) VALUES
('SKU-001','Cafetera','Cafetera express',129.90),
('SKU-002','Tostadora','Tostadora 4 ranuras',49.50);

-- Update por otro usuario
SET LOCAL app.current_user = 'bob@example.com';
UPDATE app.products SET price = 119.90 WHERE sku = 'SKU-001';

-- Delete
SET LOCAL app.current_user = 'alice@example.com';
DELETE FROM app.products WHERE sku = 'SKU-002';

-- Ver registros de auditoría
SELECT * FROM app.audit_history ORDER BY changed_at;

Código: 06_restore_examples.sql (restaurar versión antigua)

-- 06_restore_examples.sql
-- Ejemplo: reconstruir última versión antes de un update/delete
-- 1) Buscar versiones del record
SELECT id, record_id, op, changed_at, txid, data->>'name' AS name
FROM app.audit_history
WHERE table_name = 'products'
  AND record_id = 1
ORDER BY changed_at DESC;

-- 2) Restaurar completamente a un estado guardado (data es jsonb)
-- Supongamos que audit_history.id = 3 contiene la versión deseada
WITH v AS (
  SELECT data FROM app.audit_history WHERE id = 3
)
-- Para restaurar: actualizamos o insertamos según necesidad
INSERT INTO app.products (id, sku, name, description, price, available, created_at, updated_at)
SELECT (data->>'id')::bigint,
       data->>'sku',
       data->>'name',
       data->>'description',
       (data->>'price')::numeric,
       (data->>'available')::boolean,
       (data->>'created_at')::timestamptz,
       now()
FROM v
ON CONFLICT (id) DO UPDATE SET
  sku = EXCLUDED.sku,
  name = EXCLUDED.name,
  description = EXCLUDED.description,
  price = EXCLUDED.price,
  available = EXCLUDED.available,
  updated_at = now();

-- Important: el trigger de auditoría registrará este reintegro como una operación

Explicación técnica y decisiones

• Usamos txid_current() para poder agrupar cambios por transacción; útil si tu app hace múltiples cambios en distintas filas dentro de una misma operación lógica.
• Almacenar data como jsonb permite capturar la fila completa sin depender de columnas fijas; facilita restaurar aunque la tabla cambie (con precauciones).
• changed_fields se calcula usando jsonb_object_keys sobre OLD y comparando valores textuales. Es simple y funciona para la mayoría de tipos; para tipos complejos o NULLs hay que ajustar la comparación.
• Usamos triggers AFTER para que la inserción en audit_history no vea filas inconsistentes; la función de auditoría es INSERT en la tabla de audit y luego retorna NEW/OLD según corresponda.
• El usuario auditado se toma de la setting 'app.current_user' si está presente; así tu aplicación puede setearla por conexión (SET LOCAL) para identificar al usuario real en el log. Si no hay una variable, usamos current_user.

Consideraciones de rendimiento y mantenimiento

• La tabla de audit_history crece rápido: planifica retention. Opciones: particionar por rango de fecha (MONTH), archivar a otra base, o eliminar registros antiguos.
• Indexa por table_name y record_id para búsquedas rápidas y por txid/changing_at para análisis por transacción.
• Si tu carga es alta, considera enviar eventos de auditoría a una cola (Kafka) en lugar de escribir sincrónicamente, o usar logical decoding para replicar cambios fuera de la base de datos.
• Ten en cuenta el tamaño de jsonb: campos BLOB/bytea pueden inflar el historial; decide si excluirlos o almacenar referencias.

Seguridad y permisos

• Restringe el acceso directo a app.audit_history: solo personal autorizado o servicios deben poder leerla. Contiene información sensible.
• Si la auditoría es obligatoria por cumplimiento, evita que los roles puedan borrar registros de audit_history; crea políticas para la retención controlada.
• Si tu política requiere inmutabilidad, considera escribir el historial en una tabla con políticas que impidan UPDATE/DELETE, o exportar regularmente a almacenamientos inmutables.

Pruebas rápidas

1. Ejecuta los scripts en orden: 01_schema.sql, 02_audit_table.sql, 03_audit_functions.sql, 04_triggers.sql.
2. Inserta y modifica filas con different app.current_user y verifica app.audit_history.
3. Prueba restaurar una versión antigua usando 06_restore_examples.sql.

Siguiente paso: si necesitas escalabilidad a producción, implementa particionado de app.audit_history por rango de fecha y un proceso batch que archive particiones antiguas a un S3/Cold storage. Como consejo avanzado, considera integrar logical decoding (pgoutput) para replicar cambios a un sistema de auditoría externo sin afectar el path de escritura de la base de datos.

Advertencia: las restauraciones masivas pueden disparar triggers y volver a registrar auditoría; si quieres evitar que la restauración genere entradas, implementa una bandera de sesión que la función audit_if_needed lea y condicione la inserción.