Event Store ile Çalışmak: Marten ve PostgreSQL'de Append, Optimistic Concurrency ve Stream'ler

Sürüm notu: Bu yazıdaki kodlar C# 14 ve .NET 10 ile derlenip denenmiştir.

Önce Tahmin Et

Okul açılışı kampanyasının ilk sabahı. Depoda tek bir Dune nüshası kaldı ve Ada ile Alan aynı saniyede sipariş veriyor. İki istek iki ayrı thread'de, iki ayrı Marten session'ında işleniyor. İkisi de stok kaleminin stream'ini okuyor, ikisi de "1 nüsha müsait" görüyor, ikisi de 05. yazıdaki StockDecider'dan aynı kararı alıyor: StockReserved.

StockItemId dune = await NewStockItem(quantity: 1);

// Ada ve Alan son Dune nüshası için aynı anda sipariş veriyor
await using IDocumentSession ada = store.LightweightSession();
await using IDocumentSession alan = store.LightweightSession();
IEventStream<StockItem> adaStream = await ada.Events.FetchForWriting<StockItem>(dune.Value);
IEventStream<StockItem> alanStream = await alan.Events.FetchForWriting<StockItem>(dune.Value);
Console.WriteLine($"Both read: version={adaStream.StartingVersion} available={adaStream.Aggregate!.State.Available}");

adaStream.AppendMany(StockDecider.Decide(
    new ReserveStock(dune, new OrderId(Guid.CreateVersion7()), 1, now), adaStream.Aggregate.State));
OrderId alanOrder = new(Guid.CreateVersion7());
alanStream.AppendMany(StockDecider.Decide(
    new ReserveStock(dune, alanOrder, 1, now), alanStream.Aggregate!.State));

await ada.SaveChangesAsync();
Console.WriteLine("Ada:  saved");
try
{
    await alan.SaveChangesAsync();
    Console.WriteLine("Alan: saved");
}
catch (ConcurrencyException ex)
{
    Console.WriteLine($"Alan: {ex.GetType().Name}");
    Console.WriteLine($"      {ex.Message}");
}

// Yeniden dene: aynı command, güncel state ile yeni bir karar
IReadOnlyList<StockEvent> retry = await handler.HandleAsync(new ReserveStock(dune, alanOrder, 1, now));
Console.WriteLine(retry.Single() is OutOfStock o
    ? $"Alan retry: OutOfStock (requested {o.Requested}, available {o.Available})"
    : $"Alan retry: {retry.Single().GetType().Name}");

İki SaveChangesAsync çağrısından sonra ne olur? Üç seçenek var: iki rezervasyon da yazılır ve aynı nüshayı iki kişiye satmış oluruz; ikinci kayıt bir exception ile reddedilir; ya da ikinci istek birincinin bitmesini bekler ve sonra yazılır. Cevabı vermeden önce bir event store'un bu soruya neden zaten bir cevabı olması gerektiğine bakalım.

Bir Event Store'dan Ne İstiyoruz?

04. yazıda event store'u "event'leri stream'ler hâlinde, sırayla ve değiştirilemez şekilde saklayan veritabanı" diye tanımladık ve onu bir message broker'dan ayırdık. Uygulama açısından bu tanım dört somut yeteneğe dönüşüyor:

• Bir stream'e event eklemek (append) ve bunu yalnızca stream beklediğin version'dayken yapmak (expected version ile optimistic concurrency).
• Bir stream'i baştan sona, yazıldığı sırayla okumak; istersen belli bir version'a ya da zamana kadar.
• Bütün stream'leri kapsayan global bir sırayla okumak; projection'lar ve subscription'lar bununla beslenir (07. yazı).
• Bir transaction'da yazılan event'lerin ya hepsinin ya hiçbirinin görünmesi.

İlk madde aggregate'in consistency boundary'sini korur. Domain-Driven Design (DDD) serisinin "Aggregate Tasarımı: Consistency Boundary, Küçük Aggregate'ler ve Id ile Referans" yazısındaki kural burada da geçerli: bir command tek bir aggregate'i, yani tek bir stream'i değiştirir ve o stream'in version'ı, iki yazarın birbirinin kararını ezmesini engelleyen tek bilgidir. Klasik ilişkisel modelde bunun karşılığı Entity Framework Core (EF Core) serisinin "EF Core'da Concurrency: Token, rowversion ve DbUpdateConcurrencyException" yazısındaki concurrency token'dı; event store'da token, stream'in son version numarasıdır.

Marten'i Kurmak: PostgreSQL Üzerinde Event Store

Bu serideki event store Marten. Marten, PostgreSQL'i hem document database hem event store olarak kullanan bir .NET library'si; NuGet'teki güncel sürüm 9.58.0 (5 Kasım 2026), lisansı MIT (Massachusetts Institute of Technology) License, yani kısıtlaması çok az olan bir açık kaynak lisansı, hedef framework'leri net9.0 ve net10.0. Marten 9 Mayıs 2026'da çıktı ve birkaç varsayılanı değiştirdi: serializer artık System.Text.Json, event'ler varsayılan olarak "Quick" append modunda yazılıyor, runtime'da kod üretimi kaldırıldı. Marten 8 artık yalnızca kritik düzeltmeleri alıyor ve Kasım 2026'da destek dışı kalıyor; yeni bir projeye 9 ile başla.

Denemek için bir PostgreSQL yeter. Kendi makinende geçici bir container açıp işin bitince silebilirsin:

docker run -d --name bookstore-pg -e POSTGRES_PASSWORD=postgres -p 5432:5432 postgres:17
dotnet add package Marten --version 9.58.0

Store'u kurmak birkaç satır. Örnekler ASP.NET Core yerine console uygulamasında çalışsın diye DocumentStore.For kullanıyorum; bir web uygulamasında aynı ayarlar builder.Services.AddMarten(...) içine girer.

string connectionString = Environment.GetEnvironmentVariable("BOOKSTORE_PG")
    ?? "Host=localhost;Port=5432;Database=bookstore;Username=postgres;Password=postgres";

await using DocumentStore store = DocumentStore.For(opts =>
{
    opts.Connection(connectionString);
    opts.DatabaseSchemaName = "inventory";
    opts.Projections.Add(new StockItemProjection(), ProjectionLifecycle.Live);
});

StockItemProjection aşağıda geliyor; ProjectionLifecycle.Live "bu aggregate'i hiçbir yerde saklama, her okumada event'lerden yeniden kur" demek. İlk SaveChangesAsync'te Marten şemayı kendisi oluşturur. Event tarafındaki iki ana tablo mt_streams (stream başına bir satır: id, tip, son version) ve mt_events:

SELECT column_name, data_type FROM information_schema.columns
WHERE table_schema = 'inventory' AND table_name = 'mt_events' ORDER BY ordinal_position;

  column_name   |        data_type
----------------+--------------------------
 seq_id         | bigint
 id             | uuid
 stream_id      | uuid
 version        | bigint
 data           | jsonb
 bdata          | bytea
 type           | character varying
 timestamp      | timestamp with time zone
 tenant_id      | character varying
 mt_dotnet_type | character varying
 is_archived    | boolean

SELECT indexname, indexdef FROM pg_indexes WHERE schemaname = 'inventory' AND tablename = 'mt_events';

 pkey_mt_events_seq_id           | CREATE UNIQUE INDEX pkey_mt_events_seq_id ON inventory.mt_events USING btree (seq_id)
 pk_mt_events_stream_and_version | CREATE UNIQUE INDEX pk_mt_events_stream_and_version ON inventory.mt_events USING btree (stream_id, version)

İki satıra dikkat: seq_id bütün store'daki global sıra, (stream_id, version) üzerindeki unique index ise optimistic concurrency'nin veritabanındaki son savunma hattı. Aynı stream'e aynı version'la iki event yazılamaz. bdata binary serialization, tenant_id multi-tenancy, is_archived de 08. yazıdaki arşivleme için.

StartStream, Append ve Okuma

Bir stok kalemi açalım, beş nüsha girelim ve iki nüsha rezerve edelim:

StockItemId dune = new StockItemId(Guid.CreateVersion7());
await using (IDocumentSession session = store.LightweightSession())
{
    // Yeni stream: tipi ve id'si belli, ilk event'lerle birlikte
    session.Events.StartStream<StockItem>(dune.Value,
        new StockItemRegistered(dune, new BookId(Guid.CreateVersion7()), now),
        new StockReceived(dune, 5, now));
    await session.SaveChangesAsync();
}

await using (IDocumentSession session = store.LightweightSession())
{
    // Var olan stream'e ekleme (version kontrolü yok)
    session.Events.Append(dune.Value, new StockReserved(dune, new OrderId(Guid.CreateVersion7()), 2, now.AddMinutes(5)));
    await session.SaveChangesAsync();
}

await using IQuerySession query = store.QuerySession();
foreach (IEvent e in await query.Events.FetchStreamAsync(dune.Value))
    Console.WriteLine($"v{e.Version} seq={e.Sequence} {e.EventTypeName}");

StockItem? current = await query.Events.AggregateStreamAsync<StockItem>(dune.Value);
StockItem? atVersion2 = await query.Events.AggregateStreamAsync<StockItem>(dune.Value, version: 2);
Console.WriteLine($"Current: OnHand={current!.State.OnHand} Available={current.State.Available}");
Console.WriteLine($"At v2:   OnHand={atVersion2!.State.OnHand} Available={atVersion2.State.Available}");

v1 seq=1 stock_item_registered
v2 seq=2 stock_received
v3 seq=3 stock_reserved
Current: OnHand=5 Available=3
At v2:   OnHand=5 Available=5

StartStream<StockItem> yeni bir stream açar ve stream'in tipini mt_streams'e yazar; aynı id ile ikinci kez çağırırsan ExistingStreamIdCollisionException alırsın. Append ise stream yoksa onu da açar ve version kontrolü yapmaz; bu yüzden command handler'larda tek başına kullanılmaz. FetchStreamAsync event'leri version sırasıyla döndürür, AggregateStreamAsync onları projection'dan geçirerek state'i kurar. version: 2 parametresi tarihte geriye gitmenin en ucuz yolu: "rezervasyondan önce stok neydi?" sorusunun cevabı ayrı bir audit tablosu gerektirmeden geliyor.

Veritabanında event'ler JSON (JavaScript Object Notation) olarak duruyor:

 seq_id | version |      type      | data
--------+---------+----------------+------------------------------------------------------------------------------
      2 |       2 | stock_received | {"At": "2026-09-01T09:00:00+00:00", "Quantity": 5, "StockItemId": {"Value": "01a1b850-8783-7b96-9a18-c1bd6bde6e3a"}}
      3 |       3 | stock_reserved | {"At": "2026-09-01T09:05:00+00:00", "OrderId": {"Value": "01a1b850-884d-71dc-bc23-325d6e7b7510"}, "Quantity": 2, "StockItemId": {"Value": "01a1b850-8783-7b96-9a18-c1bd6bde6e3a"}}

type kolonu event'in Marten'deki adı (alias), mt_dotnet_type ise .NET tipinin tam adı. Okurken Marten önce alias'a, bulamazsa .NET tip adına bakıyor. Bu ayrıntı, tuzaklar bölümünde Türkçe makinelerde can yakan bir hataya dönüşecek.

FetchForWriting: Decider'ı Event Store'a Bağlamak

05. yazıdaki decider saf fonksiyonlardan oluşuyordu: Decide(command, state) event üretir, Evolve(state, event) state'i ilerletir. Event store'a bağlarken bu saflığı bozmak istemiyoruz. Marten'in tek bir isteği var: sakladığı ya da kurduğu her document'in bir Id'si olmalı. StockState'e id eklemek yerine onu ince bir zarfa koydum:

using JasperFx.Events;
using Marten;
using Marten.Events.Aggregation;

namespace Bookstore.Inventory;

// Marten her document'te Id ister; decider'ın state'ine dokunmadan ince bir zarf
public sealed record StockItem(Guid Id, StockState State);

// Decider'ın Evolve'unu Marten'e tanıtan adaptör
public sealed class StockItemProjection : SingleStreamProjection<StockItem, Guid>
{
    public override StockItem Evolve(StockItem? snapshot, Guid id, IEvent e) =>
        e.Data is StockEvent stockEvent
            ? new StockItem(id, StockDecider.Evolve(snapshot?.State ?? StockDecider.InitialState, stockEvent))
            : snapshot ?? new StockItem(id, StockDecider.InitialState);
}

public sealed class StockCommandHandler(IDocumentStore store)
{
    public async Task<IReadOnlyList<StockEvent>> HandleAsync(StockCommand command, CancellationToken ct = default)
    {
        await using IDocumentSession session = store.LightweightSession();

        // Stream'i ve o anki version'ını birlikte okur
        IEventStream<StockItem> stream = await session.Events.FetchForWriting<StockItem>(command.StockItemId.Value, ct);
        StockState state = stream.Aggregate?.State ?? StockDecider.InitialState;

        IReadOnlyList<StockEvent> events = StockDecider.Decide(command, state);
        stream.AppendMany(events);

        // Okuduğumuz version'dan sonra başka biri yazdıysa burada ConcurrencyException
        await session.SaveChangesAsync(ct);
        return events;
    }
}

SingleStreamProjection<TDoc, TId>'nin Evolve override'ı, Marten 8'den beri decider terminolojisiyle aynı adı taşıyor; içinde yalnızca bizim StockDecider.Evolve'u çağırıyoruz. 05. yazıdaki StockCommandHandler bellek içi store'dan stream'i okuyor, fold ediyor ve beklenen version'la append ediyordu; Marten'li hâlinde bu üç işin ikisini FetchForWriting üstleniyor, decider'a ise hiç dokunmadık. Adımlar her command için aynı: FetchForWriting ile stream'i ve o anki version'ı oku, Decide ile karar ver, AppendMany ile ekle ve SaveChangesAsync ile kaydet. Marten'in dokümantasyonu yeni kod için FetchForWriting'i açıkça öneriyor, çünkü aggregate'i ister Live, ister Inline ya da Async projection ile tutuyor ol, aynı kod çalışıyor. 08. yazıda Live'dan snapshot'a geçerken handler'a dokunmayacağız.

Bir not: Marten 9'da UseIdentityMapForAggregates varsayılan olarak açık. Bu ayar, Inline projection'larda FetchForWriting'in getirdiği instance'ı yeniden kullanır ve yalnızca handler o instance'ı değiştirmiyorsa güvenlidir. Decider'da state immutable record olduğu için bu sorun bize hiç uğramıyor; aggregate'i yerinde değiştiren klasik class'larda değişiklik iki kez uygulanabiliyor.

Optimistic Concurrency: Aynı Stream'e İki Yazar

Şimdi açılıştaki soruya dönelim. Gerçek koşunun çıktısı:

Both read: version=2 available=1
Ada:  saved
Alan: EventStreamUnexpectedMaxEventIdException
      Unexpected starting version number for event stream '01a1b850-89db-7833-889d-d27079230926', expected 2 but was -1
Alan retry: OutOfStock (requested 1, available 0)

İkinci seçenek: Alan'ın kaydı reddedildi. İki session da stream'i version 2'de okudu. Ada'nın event'i version 3 olarak yazıldı; Alan'ın session'ı da "version 2'den sonrasını ben yazıyorum" diyerek geldi ve kabul edilmedi. EventStreamUnexpectedMaxEventIdException, JasperFx'in ConcurrencyException'ından türüyor, bu yüzden handler'da tek bir catch yetiyor. Mesajdaki "but was -1" garip görünebilir: Marten 9'un varsayılanı olan Quick append modunda version kontrolü sunucudaki bir PostgreSQL fonksiyonunda yapılıyor ve gerçek version mesaja taşınmıyor. Aynı senaryoyu EventAppendMode.Rich ile çalıştırınca mesaj "expected 2 but was 3" oluyor.

Retry'ın nasıl yapıldığı da önemli. Alan'ın isteğini aynı event'lerle tekrar göndermek yanlış olurdu, çünkü o event'ler artık bayat bir state'e göre üretildi. Doğrusu command'ı baştan çalıştırmak: stream'i yeniden oku, yeniden karar ver. Bu sefer decider stokta nüsha olmadığını gördü ve 05. yazıdaki kural gereği exception yerine OutOfStock event'i üretti. Bu retry'ı genelde bir kez ya da birkaç kez, kısa bir bekleme ile yapmak yeterli; aynı stream'de sürekli çakışma yaşıyorsan sorun retry'da değil, stream'in tasarımındadır (aşağıda tek global stream tuzağı).

Bazen beklenen version'ı sunucu değil istemci bilir. Depo görevlisi sayım düzeltme ekranını açtığında stok v2'deydi; o ekrandayken bir sipariş rezervasyon yaptı. Görevlinin "3 nüsha eksik" düzeltmesi, artık görmediği bir state'e göre verilmiş bir karar. Ekranın gönderdiği version'ı FetchForWriting'e verirsen kontrol okuma anında yapılır. Bir web API'sinde (Application Programming Interface) bu version genelde HTTP (Hypertext Transfer Protocol) yanıtında ETag (entity tag) olarak gider ve bir sonraki istekte If-Match header'ı ile geri gelir:

StockItemId dune = await NewStockItem(quantity: 10);

// Depo görevlisi sayım ekranını v2'deyken açtı; bu arada bir rezervasyon geldi
long versionOnScreen = 2;
await handler.HandleAsync(new ReserveStock(dune, new OrderId(Guid.CreateVersion7()), 1, now));

await using IDocumentSession session = store.LightweightSession();
try
{
    IEventStream<StockItem> stream = await session.Events.FetchForWriting<StockItem>(dune.Value, versionOnScreen);
    stream.AppendOne(new StockAdjusted(dune, -3, AdjustmentReason.CountCorrection, now));
    await session.SaveChangesAsync();
}
catch (ConcurrencyException ex)
{
    Console.WriteLine($"{ex.GetType().Name}: {ex.Message}");
}

ConcurrencyException: Expected the existing version to be 2, but was 3

Bu sefer exception SaveChangesAsync'te değil, FetchForWriting çağrısında geldi; ekrana "Bu kayıt sen bakarken değişti, yenile" demek için en doğru yer. Marten bunlara ek olarak FetchForExclusiveWriting ile stream'e PostgreSQL row lock'u almayı da sunuyor; bu ikinci istek bekler seçeneği. Çakışmanın çok sık ve retry'ın pahalı olduğu nadir durumlar dışında optimistic yaklaşım daha iyi ölçeklenir. Eski Append(streamId, expectedVersion, events) overload'ı da hâlâ var ama Marten 9'da yalnızca Rich modda çalışıyor; dokümantasyon onun yerine de FetchForWriting'i öneriyor.

Stream Id Stratejileri

Marten'de stream id ya Guid ya da string olur (opts.Events.StreamIdentity = StreamIdentity.AsString). Seçim, aggregate'e nasıl ulaştığına bağlı:

Bookstore'da bir kitap için tek StockItem var, bu yüzden Guid v7 yeterli ve 09. yazıda ISBN'in benzersizliğini ayrıca nasıl koruyacağımızı göreceğiz. Doğal anahtar kullanacaksan anahtarın gerçekten değişmez olduğundan emin ol; stream id sonradan değiştirilemez. Bir de opts.Events.UseMandatoryStreamTypeDeclaration = true ayarını bil: tipsiz StartStream çağrılarını ve var olmayan stream'e Append'i reddeder; yazım hatası yüzünden sessizce yeni bir stream açılmasının önüne geçer ve Marten'in projection rebuild optimizasyonları için de gereklidir.

Alternatifler: KurrentDB ve Kendi Tablon

Event store için tek seçenek Marten değil. En bilineni KurrentDB: Greg Young'ın başlattığı EventStoreDB, Mart 2025'teki 25.0 sürümüyle bu adı aldı. Kendi sunucusu olan, event sourcing için yazılmış bir veritabanı: stream, expected version, subscription ve projection birinci sınıf kavramlar. .NET client'ı KurrentDB.Client (güncel 1.6.1, Apache 2.0). Sunucunun lisansı ise farklı: kaynak kodu açık olan ama OSI (Open Source Initiative) onaylı olmayan Kurrent License v1. Eylül 2026'da çıkan 26.2 ile tek node'lu kurulum, connector'lar ve kimlik doğrulama seçenekleri ücretsiz; cluster, read-only replica, arşivleme ve encryption at rest için lisans anahtarı gerekiyor. Ekip zaten PostgreSQL işletiyorsa Marten yeni bir altyapı parçası eklemiyor; event'leri ve read model'leri aynı transaction'da yazabilmek de 07. yazıda göreceğimiz Inline projection'ı mümkün kılıyor.

Üçüncü yol kendi tablonu yazmak. Çekirdeği gerçekten küçük:

await using NpgsqlDataSource dataSource = NpgsqlDataSource.Create(connectionString);
await using (NpgsqlCommand ddl = dataSource.CreateCommand("""
    CREATE TABLE IF NOT EXISTS stock_events (
        global_position bigint GENERATED ALWAYS AS IDENTITY,
        stream_id uuid NOT NULL,
        version integer NOT NULL,
        event_type text NOT NULL,
        data jsonb NOT NULL,
        recorded_at timestamptz NOT NULL DEFAULT now(),
        PRIMARY KEY (stream_id, version)
    )
    """))
    await ddl.ExecuteNonQueryAsync();

Guid streamId = Guid.CreateVersion7();
await Append(dataSource, streamId, expectedVersion: 0, "stock_received", """{"quantity": 1}""");

// İki istek de v1'i okudu ve v2'yi yazmaya çalışıyor
await Append(dataSource, streamId, expectedVersion: 1, "stock_reserved", """{"quantity": 1}""");
try
{
    await Append(dataSource, streamId, expectedVersion: 1, "stock_reserved", """{"quantity": 1}""");
}
catch (PostgresException ex) when (ex.SqlState == PostgresErrorCodes.UniqueViolation)
{
    Console.WriteLine($"Conflict: {ex.SqlState} on {ex.ConstraintName}");
}

static async Task Append(NpgsqlDataSource dataSource, Guid streamId, int expectedVersion, string type, string json)
{
    await using NpgsqlCommand insert = dataSource.CreateCommand("""
        INSERT INTO stock_events (stream_id, version, event_type, data)
        VALUES ($1, $2, $3, $4::jsonb)
        """);
    insert.Parameters.Add(new() { Value = streamId });
    insert.Parameters.Add(new() { Value = expectedVersion + 1 });
    insert.Parameters.Add(new() { Value = type });
    insert.Parameters.Add(new() { Value = json });
    await insert.ExecuteNonQueryAsync();
    Console.WriteLine($"Appended v{expectedVersion + 1} {type}");
}

Appended v1 stock_received
Appended v2 stock_reserved
Conflict: 23505 on stock_events_pkey

Optimistic concurrency'yi primary key veriyor: aynı (stream_id, version) ikinci kez yazılamaz ve PostgreSQL 23505 (unique_violation) döner. Zor kısım sonra başlıyor. global_position bir identity kolonu; değerleri transaction başında alınır ama commit sırası farklı olabilir. 105 numaralı event 104'ten önce commit edilirse, "son okuduğum pozisyon 105" diyen bir projection 104'ü sonsuza kadar kaçırır. Marten'in async daemon'u bu boşlukları "high water mark" mekanizmasıyla ve tombstone event'lerle yönetiyor; kendi tablonda bunu yeniden icat etmen gerekir. Üstüne projection çalıştırıcısı, checkpoint, rebuild, snapshot, event versioning ve serialization gelir. Kendi tablon şu durumda makul: tek bir küçük aggregate türü, projection'ı aynı transaction'da güncellenen bir iki tablo, async okuyucu yok ve ekibin bir library'ye bağımlılık eklemek istemediği net bir sebep var. Async projection ihtiyacı ortaya çıktığı gün hazır bir event store'a geçmek daha ucuz.

Sık Düşülen Tuzaklar

Tek global stream

"Bütün stok hareketleri tek bir inventory stream'inde dursun, sırası kolay okunur" fikri ilk bakışta düzenli görünür. Ama optimistic concurrency stream başına çalışır: tek stream, bütün sistemde aynı anda yalnızca bir yazarın başarılı olabileceği anlamına gelir. Kampanyada dakikada 120 sipariş geldiğinde açılıştaki çakışma istisna olmaktan çıkar, kural olur. Stream, consistency boundary'nin kendisi; aggregate kadar küçük olmalı. Global sırayı okumak için zaten seq_id var.

Devasa stream'ler

Live aggregate her okumada stream'in tamamını okur. Aynı makinede ölçtüm (ısınma okumasından sonra ikinci okuma):

    10 events -> OnHand=    9, rehydrated in 0 ms
  1000 events -> OnHand=  999, rehydrated in 7 ms
 20000 events -> OnHand=19999, rehydrated in 113 ms

Yirmi bin event, her command'da 113 ms demek. Çok popüler bir kitabın stok kalemi yıllar içinde bu sayıya ulaşabilir. Çözüm ya snapshot ya da stream'i dönemlere bölmek; ikisi de 08. yazının konusu. Burada önemli olan, stream uzunluğunu tasarım anında düşünmek: "bu stream bir yılda kaç event alır?" sorusunu sor.

Event'e domain object koymak

Event'in içine başka bir modelin entity'sini olduğu gibi koymak kolaydır:

// Kötü örnek: event, Catalog'un Book entity'sini olduğu gibi taşıyor
public sealed record Book(Guid Id, string Isbn, string Title, string Author, decimal ListPrice, string Currency, string Description);
public sealed record StockReceivedWithBook(Book Book, int Quantity, DateTimeOffset At);

{"At": "2026-09-01T09:00:00+00:00", "Book": {"Id": "01a1b850-978a-7183-94a5-3e90c497fb2b", "Isbn": "978-0441172719", "Title": "Dune", "Author": "Frank Herbert", "Currency": "EUR", "ListPrice": 14.99, "Description": "Set on the desert planet Arrakis..."}, "Quantity": 5}

Event artık Catalog'un Book class'ının o günkü hâline bağlı. Catalog Description'ı kaldırdığında ya da ListPrice'ı Money'ye çevirdiğinde, yıllar önce yazılmış event'leri deserialize ederken o eski şekli hatırlaman gerekir. Stok defteri fiyatla ilgilenmiyor bile; event yalnızca kararın gerektirdiği ve kendi context'ine ait değerleri taşımalı (BookId, Quantity). Kişisel veri taşıyan bir entity'yi event'e gömmenin bedeli daha da ağır; 09. yazıda GDPR (General Data Protection Regulation) konusunda buna döneceğiz.

Türkçe makinede event adı: stock_ıtem_registered

Bu serinin örneklerini çalıştırırken gerçekten karşılaştığım bir hata. Marten event alias'ını .NET tip adından türetiyor ve JasperFx'in ToTableAlias method'u bunun için culture'a duyarlı ToLower() kullanıyor. Makinenin culture'ı tr-TR ise büyük I noktasız ı'ya dönüşüyor:

foreach (string culture in (string[])["en-US", "tr-TR"])
{
    CultureInfo.CurrentCulture = new CultureInfo(culture);
    await using DocumentStore cultureStore = DocumentStore.For(opts =>
    {
        opts.Connection(connectionString);
        opts.DatabaseSchemaName = "inventory_culture";
    });
    await using IDocumentSession session = cultureStore.LightweightSession();
    StockItemId id = new StockItemId(Guid.CreateVersion7());
    session.Events.StartStream<StockItem>(id.Value, new StockItemRegistered(id, new BookId(Guid.CreateVersion7()), now));
    await session.SaveChangesAsync();
}

await using NpgsqlDataSource dataSource = NpgsqlDataSource.Create(connectionString);
await using NpgsqlCommand command = dataSource.CreateCommand(
    "SELECT type, count(*) FROM inventory_culture.mt_events GROUP BY type ORDER BY type");
await using NpgsqlDataReader reader = await command.ExecuteReaderAsync();
while (await reader.ReadAsync())
    Console.WriteLine($"{reader.GetString(0)}: {reader.GetInt64(1)}");

stock_item_registered: 1
stock_ıtem_registered: 1

Aynı event tipi veritabanında iki farklı adla duruyor. Marten'in kendisi bunu tolere ediyor: alias'ı tanımayınca mt_dotnet_type'taki .NET tip adına düşüyor; async daemon'u hem en-US hem tr-TR ile çalıştırdım, iki durumda da iki event'i de işledi. Sorun iki yerde çıkıyor. Birincisi, type kolonuna bakan her SQL (Structured Query Language) sorgusu, rapor ve araç event'lerin yarısını görmez. İkincisi daha sinsi: 08. yazıda event class'larını başka bir namespace'e taşıyacağız; o gün ı'lı alias'la yazılmış event'ler ne alias'la ne de artık var olmayan eski .NET tip adıyla bulunabiliyor. Çözüm iki adımlı: event adlarını opts.Events.MapEventType<StockItemRegistered>("stock_item_registered") ile açıkça ver (08. yazıda bunun versioning için de doğru olduğunu göreceğiz) ve sunucu process'lerini InvariantGlobalization ya da sabit bir culture ile çalıştır.

Küçük Bir Deneme

★☆☆ Açılıştaki yarışı EventAppendMode.Rich ile tekrar çalıştır ve exception mesajını karşılaştır. Sonra FetchForWriting yerine FetchForExclusiveWriting kullan: Alan'ın session'ı hata almak yerine ne yapıyor?

★★☆ Retry'ı StockCommandHandler'ın içine taşı: ConcurrencyException gelirse command'ı en fazla üç kez, her seferinde stream'i yeniden okuyarak çalıştır. Aynı stream'e on paralel ReserveStock gönderip kaç retry yaşandığını say.

★★★ Kendi tablonla yazdığın event store'a, global_position'a göre okuyan basit bir projection ekle. İki transaction'ı elle aç, ikincisini önce commit et ve projection'ın bir event'i kaçırdığını göster. Sonra bunu nasıl önlerdin?

Bu yazıda event'leri güvenle yazdık ve tek bir stream'den state kurduk. Ama "bu ay hangi kitaplardan kaç nüsha çıktı?" ya da "kargoya verilmeyi bekleyen rezervasyonlar hangileri?" gibi soruların cevabı tek bir stream'de yok. Bir sonraki yazıda Inline, Async ve Live projection'larla event'lerden okuma için tasarlanmış read model'ler üreteceğiz.