Este artículo describe una arquitectura para propagar cambios de datos desde una base de datos PostgreSQL al frontend en tiempo real, utilizando el sistema de notificación de PostgreSQL y WebSockets con Spring Boot.

1. Configurar el Mecanismo de Notificación en PostgreSQL

Primero, se define una función de notificación y los desencadenadores que la invocan cuando ocurren cambios en una tabla específica.

CREATE OR REPLACE FUNCTION fn_notificar_cambio_tabla() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify('canal_actualizacion_datos', 'Modificación detectada en: ' || TG_TABLE_NAME);
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

Después, se crea un desencadenador asociado a la tabla de interés. Para una tabla llamada inventario:

CREATE TRIGGER trg_inventario_cambios
AFTER INSERT OR UPDATE OR DELETE ON inventario
FOR EACH ROW EXECUTE FUNCTION fn_notificar_cambio_tabla();

2. Integración con Spring Boot

Se configura el endpoint de WebSocket en la aplicación Spring Boot.

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;

@Configuration
@EnableWebSocket
public class WsServerConfig implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(eventosWsHandler(), "/eventos-ws").setAllowedOrigins("*");
    }

    @Bean
    public EventosWsHandler eventosWsHandler() {
        return new EventosWsHandler();
    }
}

3. Gester de Conexiones WebSocket

El gestor mantiene las sesiones activas y permite el envío de mensajse masivos.

import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class EventosWsHandler extends TextWebSocketHandler {
    private final Set<WebSocketSession> sesionesActivas = ConcurrentHashMap.newKeySet();

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        sesionesActivas.add(session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        sesionesActivas.remove(session);
    }

    public void transmitirActualizacion(String payload) {
        sesionesActivas.forEach(sesion -> {
            try {
                if (sesion.isOpen()) {
                    sesion.sendMessage(new TextMessage(payload));
                }
            } catch (Exception e) {
                // Manejo de errores
            }
        });
    }
}

4. Servicio de Escucha de Base de Datos

Este servicio se conecta a PostgreSQL, escucha el cenal de notificación y retransmite los mensajes a los clientes WebSocket.

import org.postgresql.PGConnection;
import org.postgresql.PGNotification;
import org.springframework.stereotype.Service;
import jakarta.annotation.PostConstruct;
import java.sql.*;

@Service
public class EscuchaPgService {

    private final EventosWsHandler eventosWsHandler;
    private Connection connDb;
    private Thread hiloEscucha;

    public EscuchaPgService(EventosWsHandler eventosWsHandler) {
        this.eventosWsHandler = eventosWsHandler;
    }

    @PostConstruct
    public void iniciar() throws SQLException {
        connDb = DriverManager.getConnection("jdbc:postgresql://localhost:5432/mi_base", "usuario", "clave");
        try (Statement stmt = connDb.createStatement()) {
            stmt.execute("LISTEN canal_actualizacion_datos");
        }

        hiloEscucha = new Thread(() -> {
            while (true) {
                try {
                    PGConnection pgConn = connDb.unwrap(PGConnection.class);
                    PGNotification[] notifs = pgConn.getNotifications(1000);
                    if (notifs != null) {
                        for (PGNotification notif : notifs) {
                            String mensaje = "{\"evento\": \"" + notif.getParameter() + "\"}";
                            eventosWsHandler.transmitirActualizacion(mensaje);
                        }
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        });
        hiloEscucha.setDaemon(true);
        hiloEscucha.start();
    }
}

5. Componente Frontend (Vue 3)

Un componente Vue simple que se conecta al WebSocket y muestra los eventos recibidos.

<script setup>
import { ref, onMounted, onUnmounted } from 'vue';

const ultimoEvento = ref('');
let conexionWs = null;

const conectarWs = () => {
    conexionWs = new WebSocket('ws://localhost:8080/eventos-ws');
    
    conexionWs.onmessage = (evento) => {
        const datos = JSON.parse(evento.data);
        ultimoEvento.value = datos.evento;
    };
    
    conexionWs.onerror = (error) => {
        console.error('Error en WS:', error);
    };
};

onMounted(conectarWs);

onUnmounted(() => {
    if (conexionWs && conexionWs.readyState === WebSocket.OPEN) {
        conexionWs.close();
    }
});
</script>

<template>
    <div class="panel-eventos">
        <h3>Monitor de Eventos</h3>
        <p>Último cambio: {{ ultimoEvento }}</p>
    </div>
</template>