CQRS y Event Sourcing: Modelos de Lectura y Estado como Historia

// Lado de escritura: comando con validación de invariantes
public record AsignarTarea(long tareaId, String responsable) {}

@Service
class ComandosTarea {
    @Transactional
    public void manejar(AsignarTarea cmd) {
        var tarea = repo.findById(cmd.tareaId()).orElseThrow();
        tarea.asignarA(cmd.responsable());                    // reglas del dominio
        eventos.publishEvent(new TareaAsignada(cmd.tareaId(), cmd.responsable()));
    }
}

// Lado de lectura: tabla desnormalizada pensada para el panel
public record ResumenResponsable(String responsable, int pendientes, int completadasSemana) {}

@Component
class ProyeccionPanel {
    private final JdbcClient jdbc;
    ProyeccionPanel(JdbcClient jdbc) { this.jdbc = jdbc; }

    @TransactionalEventListener
    void on(TareaAsignada e) {
        jdbc.sql("""
            INSERT INTO panel_responsable (responsable, pendientes) VALUES (:r, 1)
            ON CONFLICT (responsable) DO UPDATE SET pendientes = panel_responsable.pendientes + 1
            """).param("r", e.responsable()).update();
    }

    public List<ResumenResponsable> panel() {                // consulta trivial y rápida
        return jdbc.sql("SELECT * FROM panel_responsable ORDER BY pendientes DESC")
            .query(ResumenResponsable.class).list();
    }
}

@Component
class ResumenClienteProyeccion {
    private final JdbcTemplate jdbc;

    ResumenClienteProyeccion(JdbcTemplate jdbc) { this.jdbc = jdbc; }

    @Transactional
    public void aplicar(EventoPedido evento, long posicion) {
        // Idempotencia: si esta posición ya se procesó, no se hace nada
        int avanza = jdbc.update("""
            update proyeccion_estado set posicion = ?
            where nombre = 'resumen_cliente' and posicion < ?""", posicion, posicion);
        if (avanza == 0) return;

        switch (evento) {
            case PedidoConfirmado(var pedidoId, var clienteId, var total, var cuando) -> jdbc.update("""
                insert into resumen_cliente (cliente_id, pedidos, total, ultimo_pedido)
                values (?, 1, ?, ?)
                on conflict (cliente_id) do update set
                    pedidos = resumen_cliente.pedidos + 1,
                    total = resumen_cliente.total + excluded.total,
                    ultimo_pedido = excluded.ultimo_pedido""", clienteId, total, cuando);
            case PedidoCancelado(var pedidoId, var clienteId, var total) -> jdbc.update("""
                update resumen_cliente set pedidos = pedidos - 1, total = total - ?
                where cliente_id = ?""", total, clienteId);
            default -> { }                           // eventos que esta vista no necesita
        }
    }

    // Reconstruir: vaciar la vista y reproducir todos los eventos desde el principio
    public void reconstruir(EventStore store) {
        jdbc.update("truncate resumen_cliente");
        jdbc.update("update proyeccion_estado set posicion = 0 where nombre = 'resumen_cliente'");
        store.todosDesde(0).forEach(e -> aplicar(e.evento(), e.posicion()));
    }
}

sealed interface EventoCuenta permits CuentaAbierta, DineroIngresado, DineroRetirado {}
record CuentaAbierta(String id, String titular) implements EventoCuenta {}
record DineroIngresado(String id, BigDecimal importe) implements EventoCuenta {}
record DineroRetirado(String id, BigDecimal importe) implements EventoCuenta {}

final class Cuenta {
    private String id;
    private BigDecimal saldo = BigDecimal.ZERO;
    private final List<EventoCuenta> pendientes = new ArrayList<>();

    static Cuenta reconstruir(List<EventoCuenta> historial) {
        var cuenta = new Cuenta();
        historial.forEach(cuenta::aplicar);                 // estado = fold(eventos)
        return cuenta;
    }

    void retirar(BigDecimal importe) {
        if (saldo.compareTo(importe) < 0) throw new SaldoInsuficienteException(id);
        registrar(new DineroRetirado(id, importe));         // decide y registra, no muta directamente
    }

    private void registrar(EventoCuenta e) { aplicar(e); pendientes.add(e); }

    private void aplicar(EventoCuenta e) {
        switch (e) {
            case CuentaAbierta a -> id = a.id();
            case DineroIngresado i -> saldo = saldo.add(i.importe());
            case DineroRetirado r -> saldo = saldo.subtract(r.importe());
        }
    }

    List<EventoCuenta> eventosPendientes() { return List.copyOf(pendientes); }
}

// create table eventos (
//     agregado_id  uuid        not null,
//     version      int         not null,
//     tipo         text        not null,
//     contenido    jsonb       not null,
//     ocurrido_en  timestamptz not null,
//     posicion     bigserial,                      -- orden global para las proyecciones
//     primary key (agregado_id, version)
// );

public class EventStoreJdbc implements EventStore {
    private final JdbcTemplate jdbc;
    private final SerializadorEventos serializador;

    public EventStoreJdbc(JdbcTemplate jdbc, SerializadorEventos serializador) {
        this.jdbc = jdbc;
        this.serializador = serializador;
    }

    @Override
    @Transactional
    public void anadir(UUID agregadoId, int versionEsperada, List<EventoPedido> nuevos) {
        int version = versionEsperada;
        try {
            for (EventoPedido e : nuevos) {
                version++;
                jdbc.update("""
                    insert into eventos (agregado_id, version, tipo, contenido, ocurrido_en)
                    values (?, ?, ?, ?::jsonb, now())""",
                    agregadoId, version, e.getClass().getSimpleName(), serializador.aJson(e));
            }
        } catch (DuplicateKeyException conflicto) {
            // Otro comando escribió antes: se deshace todo y el llamador recarga y reintenta
            throw new ConcurrenciaException(agregadoId, versionEsperada);
        }
    }
}

// v1 (guardado durante años): sin moneda
record PedidoCreadoV1(UUID pedidoId, UUID clienteId, BigDecimal total) {}

// v2 (actual): la moneda es obligatoria
record PedidoCreado(UUID pedidoId, UUID clienteId, BigDecimal total, String moneda) implements EventoPedido {}

final class Upcaster {
    // Convierte cualquier versión guardada a la actual antes de aplicarla al agregado
    static EventoPedido actualizar(EventoGuardado guardado, SerializadorEventos json) {
        return switch (guardado.tipo()) {
            case "PedidoCreado" -> guardado.versionEsquema() == 1
                ? desdeV1(json.leer(guardado.contenido(), PedidoCreadoV1.class))
                : json.leer(guardado.contenido(), PedidoCreado.class);
            default -> json.leerEvento(guardado.tipo(), guardado.contenido());
        };
    }

    private static PedidoCreado desdeV1(PedidoCreadoV1 v1) {
        return new PedidoCreado(v1.pedidoId(), v1.clienteId(), v1.total(), "EUR");   // valor acordado con negocio
    }
}