04 - Event Sourcing + CQRS Insieme: Projection, Snapshot e Consistency

Dove event sourcing e CQRS si incontrano: nella proiezione

Hai un sistema che deve poter dimostrare che cosa è successo a ogni ordine, e che deve rispondere a elenchi e ricerche molto più spesso di quanto scriva. Il primo requisito suggerisce di salvare gli eventi; il secondo suggerisce di non leggere mai dagli eventi. La soluzione è usarli insieme: il lato scrittura salva eventi (event sourcing), il lato lettura legge da tabelle costruite a partire da quegli eventi (CQRS). Il punto d'incontro si chiama proiezione, ed è dove nascono i problemi veri: la proiezione è sempre un po' indietro, può ricevere lo stesso evento due volte e deve poter essere buttata e ricostruita.

Ti servono i due articoli che precedono: Event Sourcing per l'event store e la concorrenza ottimistica, CQRS per i due lati. Qui evito di ripetere gli snapshot (sono nell'articolo 02, con il conteggio degli eventi applicati) e mi concentro su ciò che appartiene solo alla combinazione, usando un programma Java 21 unico, in memoria: un ordine, il suo store, due proiezioni e le tre situazioni difficili.

Il percorso di un comando, passo per passo

1. Il client invia un comando, per esempio ConfermaPagamento.
2. Il gestore carica l'aggregato rileggendo i suoi eventi dall'event store.
3. L'aggregato valida il comando e produce uno o più eventi, oppure lo rifiuta.
4. Gli eventi vengono accodati nello store, con la versione attesa per la concorrenza ottimistica.
5. Gli eventi raggiungono le proiezioni, direttamente o tramite un broker.
6. Le proiezioni aggiornano i modelli di lettura.
7. Le query del client leggono dal modello di lettura.

Qui l'ordine è tutto: invertendo i passi la sequenza non funziona. Ma il punto da tenere a mente è dove la catena si spezza. I passi da 1 a 4 sono un'unica operazione sincrona che termina con la risposta al client; dal passo 5 in poi si lavora in modo asincrono, e il client ha già ricevuto la risposta. Tutta la «consistenza eventuale» di questo pattern vive in quella cesura, ed è per questo che il comando deve restituire qualcosa che permetta al client di sapere quando la lettura lo ha raggiunto, come vedremo.

Lato scrittura: eventi, store e aggregato in Java 21

Gli eventi sono record che implementano una interfaccia sealed: il compilatore conosce l'elenco chiuso dei tipi e, quando scrivi uno switch su Event, ti obbliga a trattarli tutti, senza default. È la funzione di pattern matching per switch finalizzata in Java 21 (JEP 441), ed è utile qui perché aggiungere un evento rompe la compilazione in ogni punto che non lo gestisce.

Gli importi sono in centesimi (long), non in BigDecimal, per tenere il listato corto e per non confondere il lettore con questioni di arrotondamento. Lo store tiene due numeri per evento: la position, globale, che dà un ordine unico a tutti gli eventi, e la version, per aggregato, usata per la concorrenza.

    sealed interface Event permits OrdineEffettuato, PagamentoConfermato, OrdineAnnullato {}
    record OrdineEffettuato(String ordineId, String clienteId, long totaleCent) implements Event {}
    record PagamentoConfermato(String ordineId, String transactionId) implements Event {}
    record OrdineAnnullato(String ordineId, String motivo) implements Event {}

    // Un evento salvato: position e' globale (ordine di scrittura), version e' per aggregato.
    record Stored(long position, String ordineId, long version, Event event) {}

    static class EventStore {
        private final List<Stored> log = new ArrayList<>();
        long head() { return log.size(); }

        long versionOf(String id) { return log.stream().filter(s -> s.ordineId().equals(id)).count(); }

        void append(String id, long expectedVersion, List<Event> events) {
            long current = versionOf(id);
            if (current != expectedVersion)
                throw new IllegalStateException("Conflitto su " + id + ": atteso v" + expectedVersion + ", trovato v" + current);
            long v = expectedVersion;
            for (Event e : events) log.add(new Stored(log.size() + 1, id, ++v, e));
        }

        List<Event> load(String id) { return log.stream().filter(s -> s.ordineId().equals(id)).map(Stored::event).toList(); }

        List<Stored> readAfter(long position, int max) {
            return log.stream().filter(s -> s.position() > position).limit(max).toList();
        }
    }

append confronta la versione attesa con quella corrente e rifiuta se differiscono: è la concorrenza ottimistica dell'articolo precedente, ridotta all'osso. In un database vero la verifica e le scritture devono stare nella stessa transazione, altrimenti il controllo può essere superato da una scrittura concorrente. readAfter restituisce gli eventi dopo una posizione: è l'unica operazione che serve a una proiezione, e vedremo perché.

    enum Stato { IN_ATTESA_PAGAMENTO, PAGATO, ANNULLATO }
    record Ordine(Stato stato) {
        static Ordine evolve(Ordine o, Event e) {
            return switch (e) {
                case OrdineEffettuato x -> new Ordine(Stato.IN_ATTESA_PAGAMENTO);
                case PagamentoConfermato x -> new Ordine(Stato.PAGATO);
                case OrdineAnnullato x -> new Ordine(Stato.ANNULLATO);
            };
        }
        static Ordine rehydrate(List<Event> events) {
            Ordine o = null;
            for (Event e : events) o = evolve(o, e);
            return o;
        }
        List<Event> confermaPagamento(String ordineId, String tx) {
            if (stato != Stato.IN_ATTESA_PAGAMENTO) throw new IllegalStateException("Pagamento non accettato in stato " + stato);
            return List.of(new PagamentoConfermato(ordineId, tx));
        }
    }

L'aggregato è due funzioni, come nell'articolo 02. evolve applica un evento e non decide niente; confermaPagamento è la decisione: accetta solo da IN_ATTESA_PAGAMENTO e produce un evento. Non esiste nessun setStato: lo stato si cambia solo producendo un evento che evolve sa leggere. La versione di un aggregato non è un campo che l'aggregato incrementa: la decide lo store, che è l'unico a sapere quanti eventi esistono davvero.

Lato lettura: la proiezione con checkpoint

Una proiezione riceve gli eventi in ordine e li traduce in righe. Le due proprietà che la rendono sicura sono il checkpoint, cioè l'ultima position elaborata, e il fatto di poter ripartire da zero. Il checkpoint la rende idempotente: se riceve un evento con position minore o uguale, lo ignora.

Serve perché quasi tutti i broker garantiscono la consegna almeno una volta, non esattamente una. Il secondo modello, Incassato, non esisteva quando gli ordini sono stati scritti; lo costruisci rileggendo la storia, e non devi cambiare niente nel lato scrittura. È uno dei motivi concreti per cui si accetta il costo del modello.

    interface Projection { void handle(Stored s); }

    // Read model 1: una riga per ordine. Il checkpoint rende il gestore idempotente.
    record Riga(String clienteId, Stato stato, long totaleCent, long versione) {}
    static class OrdiniLista implements Projection {
        final Map<String, Riga> righe = new TreeMap<>();
        long checkpoint = 0;
        int applicati = 0;
        public void handle(Stored s) {
            if (s.position() <= checkpoint) return;            // gia' visto: consegna ripetuta, si ignora
            Riga r = righe.get(s.ordineId());
            switch (s.event()) {
                case OrdineEffettuato e -> righe.put(e.ordineId(), new Riga(e.clienteId(), Stato.IN_ATTESA_PAGAMENTO, e.totaleCent(), s.version()));
                case PagamentoConfermato e -> righe.put(e.ordineId(), new Riga(r.clienteId(), Stato.PAGATO, r.totaleCent(), s.version()));
                case OrdineAnnullato e -> righe.put(e.ordineId(), new Riga(r.clienteId(), Stato.ANNULLATO, r.totaleCent(), s.version()));
            }
            checkpoint = s.position();
            applicati++;
        }
    }

    // Read model 2, nato dopo: incassato per cliente. Nessun evento nuovo, solo un replay.
    static class Incassato implements Projection {
        final Map<String, Long> totaleDi = new HashMap<>();
        final Map<String, String> clienteDi = new HashMap<>();
        final Map<String, Long> perCliente = new TreeMap<>();
        public void handle(Stored s) {
            switch (s.event()) {
                case OrdineEffettuato e -> { clienteDi.put(e.ordineId(), e.clienteId()); totaleDi.put(e.ordineId(), e.totaleCent()); }
                case PagamentoConfermato e -> perCliente.merge(clienteDi.get(e.ordineId()), totaleDi.get(e.ordineId()), Long::sum);
                case OrdineAnnullato e -> {}
            }
        }
    }

    static int aggiorna(EventStore store, OrdiniLista p, int maxEventi) {
        List<Stored> batch = store.readAfter(p.checkpoint, maxEventi);
        batch.forEach(p::handle);
        return batch.size();
    }

In OrdiniLista.handle la prima riga è il controllo di idempotenza; l'ultima aggiorna il checkpoint dopo aver applicato l'evento. In un database vero le due cose vanno nella stessa transazione, altrimenti un crash fra l'update della riga e quello del checkpoint ti lascia con un evento applicato due volte oppure mai. aggiorna legge al massimo maxEventi eventi dopo il checkpoint: nel programma serve a simulare una proiezione in ritardo.

Che cosa succede quando lo si esegue

La parte principale crea sei eventi su tre ordini, poi guarda le tre situazioni difficili: il ritardo, la consegna ripetuta, la ricostruzione. La rilettura di eventi già proiettati e il rebuild sono le prove che contano, perché sono quelle che un test superficiale non copre.

    public static void main(String[] args) {
        EventStore store = new EventStore();
        store.append("o-1", 0, List.of(new OrdineEffettuato("o-1", "cli-A", 4500)));
        store.append("o-2", 0, List.of(new OrdineEffettuato("o-2", "cli-A", 1200)));
        store.append("o-3", 0, List.of(new OrdineEffettuato("o-3", "cli-B", 9900)));
        store.append("o-1", 1, Ordine.rehydrate(store.load("o-1")).confermaPagamento("o-1", "tx-1"));
        store.append("o-3", 1, List.of(new OrdineAnnullato("o-3", "cliente")));
        store.append("o-2", 1, Ordine.rehydrate(store.load("o-2")).confermaPagamento("o-2", "tx-2"));
        System.out.println("eventi nello store: " + store.head());

        OrdiniLista lista = new OrdiniLista();
        System.out.println("lista subito dopo i comandi: " + lista.righe.size() + " righe, checkpoint " + lista.checkpoint);

        // Il comando su o-1 ha prodotto la versione 2. Il client aspetta che la lista la raggiunga.
        int giri = 0;
        while (lista.righe.get("o-1") == null || lista.righe.get("o-1").versione() < 2) { aggiorna(store, lista, 2); giri++; }
        System.out.println("o-1 visibile alla versione 2 dopo " + giri + " aggiornamenti, checkpoint " + lista.checkpoint);

        // Consegna doppia: gli eventi gia' proiettati arrivano una seconda volta e non cambiano nulla.
        int prima = lista.applicati;
        store.readAfter(0, (int) lista.checkpoint).forEach(lista::handle);
        System.out.println("rilettura degli eventi gia' visti: applicati " + prima + " -> " + lista.applicati + " (checkpoint " + lista.checkpoint + ")");
        aggiorna(store, lista, 100);
        System.out.println("lista allineata: " + lista.righe);

        // Un secondo read model costruito oggi sulla storia di ieri.
        Incassato incassato = new Incassato();
        store.readAfter(0, 100).forEach(incassato::handle);
        System.out.println("incassato per cliente: " + incassato.perCliente);

        // Ricostruzione della lista da zero: stesso risultato di quella tenuta in vita.
        OrdiniLista rifatta = new OrdiniLista();
        while (aggiorna(store, rifatta, 100) > 0) {}
        System.out.println("rebuild identico: " + rifatta.righe.equals(lista.righe));

        try { store.append("o-1", 1, List.of(new OrdineAnnullato("o-1", "doppio click"))); }
        catch (IllegalStateException e) { System.out.println("concorrenza: " + e.getMessage()); }
        try { Ordine.rehydrate(store.load("o-3")).confermaPagamento("o-3", "tx-3"); }
        catch (IllegalStateException e) { System.out.println("regola: " + e.getMessage()); }
    }
}

// Output (esecuzione dell'intero file):
// eventi nello store: 6
// lista subito dopo i comandi: 0 righe, checkpoint 0
// o-1 visibile alla versione 2 dopo 2 aggiornamenti, checkpoint 4
// rilettura degli eventi gia' visti: applicati 4 -> 4 (checkpoint 4)
// lista allineata: {o-1=Riga[clienteId=cli-A, stato=PAGATO, totaleCent=4500, versione=2], o-2=Riga[clienteId=cli-A, stato=PAGATO, totaleCent=1200, versione=2], o-3=Riga[clienteId=cli-B, stato=ANNULLATO, totaleCent=9900, versione=2]}
// incassato per cliente: {cli-A=5700}
// rebuild identico: true
// concorrenza: Conflitto su o-1: atteso v1, trovato v2
// regola: Pagamento non accettato in stato ANNULLATO

La prima riga di uscita mostra il problema: sei eventi nello store, zero righe nella lista. Il comando su o-1 ha prodotto la versione 2 e il client la aspetta: il ciclo while chiama aggiorna due volte, a due eventi per volta, e il checkpoint arriva a 4. Nota che o-1 compare alla versione 2 quando il checkpoint è 4, non 6: la lista non è allineata a tutto lo store, solo a ciò che serve a quel client.

Nella rilettura degli eventi già visti, applicati resta 4, quindi la consegna ripetuta non ha effetto. Il rebuild dà una lista identica a quella tenuta aggiornata: se non fosse così, la proiezione avrebbe una dipendenza nascosta dal tempo o dall'ordine di arrivo, un bug. Le ultime due righe sono i rifiuti: la concorrenza (due comandi sulla stessa versione, il secondo perde) e la regola (non si paga un ordine annullato). Nessuno dei due ha scritto nulla.

Tre strategie per la finestra di inconsistenza

Dopo un comando riuscito la lista non è aggiornata. Il client ha tre strade, che si possono combinare:

• Restituire la versione nella risposta al comando, e far attendere la lettura fino a quella versione, come nel ciclo sopra.
• Interrogare con tentativi la lettura finché non mostra la modifica, con un limite di tempo e di tentativi.
• Proiettare in modo sincrono i dati critici, aggiornando il modello di lettura nella stessa transazione della scrittura degli eventi.

Le prime due sono la stessa idea, e la prima è preferibile perché non indovina un tempo: sa che cosa aspetta. La terza elimina la finestra solo per quel modello, ma lega lato lettura e lato scrittura nello stesso database e nella stessa transazione, cioè annulla il vantaggio di averli separati. Usala per una sola vista critica (per esempio la ricevuta mostrata subito dopo il pagamento), non come regola. Un'alternativa che non tocca il backend è mostrare subito il risultato atteso nell'interfaccia e riconciliarlo quando la lettura è pronta, che è il tema dell'articolo sulle strategie di consistenza eventuale.

Errori che costano caro

Il primo è modificare un evento già scritto: l'event store è la fonte di verità e riscriverlo rompe il replay di ogni proiezione e l'audit. Se lo schema cambia, scrivi una versione nuova dell'evento e adatta quelli vecchi in lettura (upcasting); il versionamento è trattato negli eventi di dominio. Il secondo è far scrivere a un comando più aggregati nella stessa transazione: il confine di consistenza è l'aggregato, e per coordinarne più d'uno serve una saga.

Il terzo è pubblicare gli eventi sul broker dopo averli scritti senza garanzie: se il processo muore in mezzo, le proiezioni non sapranno mai dell'evento, e il rimedio è l'outbox. Il quarto è la proiezione non idempotente: funziona in test e produce righe duplicate la prima volta che il broker riconsegna.

Che cosa sai fare adesso

Sai spiegare dove il pattern combinato diventa asincrono, sai scrivere una proiezione con checkpoint che regge le consegne ripetute e il rebuild, e sai scegliere come gestire la finestra di inconsistenza invece di ignorarla. Il passo successivo è la coordinazione fra servizi, con la saga.