CQRS ve Event Sourcing — Olanı Yaz, Okumayı Ayır

Bankadaki eski hesap cüzdanlarını düşün: kimse bakiyeyi silip yenisini yazmazdı. Her para girişi ve çıkışı yeni bir satırdı; bakiye, satırların toplamıydı.

Yazan model, okuyan model

Para çekerken merak edilen, kuralın buna izin verip vermediğidir. Hesap ekranı ise son on işlemi ve bakiyeyi ister. İkisini tek tabloya sığdırmak çoğu zaman iki tarafı da zorlar.

CQRS?Terimin tanımını açCQRSCommand Query Responsibility Segregation: değiştiren işler (komutlar) ile okuyan işlerin (sorgular) ayrı modellerle yapılması. Okuma tarafı yazmanın biraz gerisinden gelebilir.Sözlükte gör → bu iki işi ayırır. Komutlar yazma modeline gider ve kuralları korur; sorgular ekranın sorusuna göre şekillenmiş bir okuma modelinden?Terimin tanımını açread modelCQRS'te yalnızca okumak için tutulan, ekranın sorusuna göre şekillenmiş veri. Yazma tarafındaki değişikliklerden beslenir; silinip yeniden kurulabilir.Sözlükte gör → cevaplanır.

1
// Write side: checks the rules, records the change, returns the new version.
2
long version = accounts.handle(new Withdraw(accountId, new BigDecimal("80")));
3

4
// Read side: a flat table shaped for the screen, filled from the changes a moment later.
5
AccountSummary summary = summaries.findById(accountId);   // balance + last 10 lines

Bedeli gecikmedir: okuma modeli yazmanın biraz gerisinden gelir. Kullanıcı “kaydedildi” görüp sayfayı yenilediğinde eski bakiyeyle karşılaşabilir. Bu, okuma replikalarındaki read-your-writes sorununun aynısıdır; çözümleri de benzer.

Satır satır: bakiyeyi olaylardan kurmak

Event sourcing?Terimin tanımını açevent sourcingBir nesnenin son hâlini değil, başına gelen olayları saklamak. Durum, olaylar baştan sırayla uygulanarak kurulur; olaylar yalnızca sona eklenir.Sözlükte gör → kullanan bir sistemde hesabın son hâli hiçbir yere yazılmaz. Saklanan şey olaylardır: MoneyDeposited, MoneyWithdrawn. Olaylar yalnızca sona eklenir; hiçbiri değiştirilmez.

Hesabı yüklemek, olayları baştan sırayla uygulamak demektir. Kuralları bilen nesne, zengin domain modeli dersindeki aggregate’in ta kendisi; yalnızca durumunu olaylardan kurar.

Olaylar binlere çıkınca her komutta baştan okumak yorar. Snapshot, arada bir bakiyenin fotoğrafını çeker; yükleme son fotoğraftan başlar ve yalnızca sonrasını okur. Fotoğraf kaybolsa da olaylar durduğu için yeniden çekilir.

Olay deposu aynı zamanda bir outbox gibi çalışır: yazılan tek şey olay olduğu için dual write oluşmaz. Olayları yayınlayan bir relay ve idempotent tüketiciler yine gerekir.

Eski olaylar, yeni kod

Üç yıl önce hesaplar tek para birimliydi ve MoneyDeposited olayında currency alanı yoktu. Bugünkü kod bu alanı zorunlu sayıyor.

Olayı depoda düzeltmek cazip gelir, ama olay olmuş bir gerçektir ve onu okumuş servisler var. Çözüm upcasting?Terimin tanımını açupcastingDepoda eski biçimde duran bir olayı, okunurken yeni biçime çevirmek. Olayın kendisi değişmez; yalnızca kod onu yeni sürüm gibi görür.Sözlükte gör →: eski olay okunurken yeni biçime çevrilir (currency = "TRY"), depodaki olay aynen kalır.

Yeni olaylar hep son sürümle yazılır. Bir alanın anlamı değişecekse eski olayı zorlamak yerine yeni bir olay tipi açmak daha güvenlidir.

Kendin gör

Üstte iki bakiye var: olaylardan hesaplanan gerçek bakiye ve okuma modelinin gösterdiği. Altta olay deposu.

1. “100 TL yatır”a bas, hemen ardından “Bakiyeyi oku”. Okuma modeli eski bakiyeyi gösteriyor. Oynat’a bas ve yeni olayın “yolda”dan “okuma modelinde”ye geçmesini izle.
2. Okumayı “Kendi yazdığım sürümü bekle” yap. Yatır, oku, sonra zamanı ilerlet: okuma bekler ve taze döner.
3. Snapshot’ı aç ve bir komut ver. Günlükte kaç olay okunduğuna bak: sekiz yerine üç.
4. Eski sürüm olayları aç ve okuma modelini baştan kur. Kurulum #2’de kırılıyor. Upcaster’ı da açıp yeniden dene.

Tuzaklar

İkisini tek paket sanmak. CQRS olağan tablolarla da yapılır; event sourcing ayrı bir okuma modeli olmadan da. Birini seçmek ötekini zorunlu kılmaz.

Gerek yokken kurmak. Ekle, düzenle, listele ekranı için olay deposu, projeksiyon ve upcaster ağır bir bedeldir. Yalnızca “kim, ne zaman değiştirdi” soruluyorsa bir değişiklik geçmişi tablosu yeter.

Okuma modeline bakarak karar vermek. Para çekme kuralı gecikmeli bakiyeye bakarsa yanlış karar verir. Kurallar yazma tarafında, olaylardan yüklenen hâlle çalışır.

Projeksiyonu idempotent yazmamak. Olaylar en az bir kez gelir. Projeksiyon, uyguladığı son olayın sıra numarasını tutarak tekrarı tanır.

1
CREATE TABLE account_events (
2
    account_id     VARCHAR(34)  NOT NULL,
3
    seq            BIGINT       NOT NULL,     -- the aggregate's version after this event
4
    event_type     VARCHAR(100) NOT NULL,
5
    schema_version INT          NOT NULL,
6
    payload        JSONB        NOT NULL,
7
    recorded_at    TIMESTAMPTZ  NOT NULL DEFAULT now(),
8
    PRIMARY KEY (account_id, seq)             -- two writers expecting the same version: one fails
9
);
10

11
-- Read side: shaped for the account screen, rebuildable from account_events at any time.
12
CREATE TABLE account_balance (
13
    account_id VARCHAR(34)   PRIMARY KEY,
14
    balance    NUMERIC(19,2) NOT NULL,
15
    last_seq   BIGINT        NOT NULL
16
);

1
public sealed interface AccountEvent permits AccountOpened, MoneyDeposited, MoneyWithdrawn {}
2

3
public record AccountOpened(String accountId) implements AccountEvent {}
4

5
public record MoneyDeposited(BigDecimal amount, Currency currency) implements AccountEvent {}
6

7
public record MoneyWithdrawn(BigDecimal amount, Currency currency) implements AccountEvent {}

1
// v1 events were written before accounts had a currency: {"amount": 100}.
2
// They stay that way in the store; only what the code sees is translated.
3
final class MoneyDepositedUpcaster implements Upcaster {
4

5
    @Override
6
    public boolean canUpcast(StoredEvent event) {
7
        return event.type().equals("MoneyDeposited") && event.schemaVersion() == 1;
8
    }
9

10
    @Override
11
    public StoredEvent upcast(StoredEvent event) {
12
        ObjectNode payload = event.payload().deepCopy();
13
        payload.put("currency", "TRY");          // every v1 account was a TRY account
14
        return event.withPayload(payload).withSchemaVersion(2);
15
    }
16
}

1
@Component
2
class BalanceProjection {
3

4
    private final JdbcTemplate jdbc;
5

6
    BalanceProjection(JdbcTemplate jdbc) {
7
        this.jdbc = jdbc;
8
    }
9

10
    @KafkaListener(topics = "account-events", groupId = "balance-projection")
11
    @Transactional
12
    public void on(@Payload MoneyDeposited event, @Header("account-id") String accountId,
13
                   @Header("seq") long seq) {
14
        // Apply only the next event of this account.
15
        // 0 rows updated means a duplicate, or a gap the retry will fill once the earlier event lands.
16
        jdbc.update("""
17
                UPDATE account_balance
18
                   SET balance = balance + ?, last_seq = ?
19
                 WHERE account_id = ? AND last_seq = ?
20
                """, event.amount(), seq, accountId, seq - 1);
21
    }
22
}

Kendini sına

Önce hızlı bir ısınma: puan yok, kayıt yok. Sonra asıl sorular.