Message Türleri: Command, Event ve Query; Event Notification ve Event-Carried State Transfer

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

Önce Tahmin Et

Bir önceki yazıda onay e-postasını bir queue'ya bıraktık. Ekip bu fikri sevdi ve birkaç hafta içinde Bookstore'un kodunda şu message'lar belirdi:

namespace Bookstore.Before;

public sealed class Order
{
    public Guid Id { get; set; }
    public Guid CustomerId { get; set; }
    public string Status { get; set; } = "Placed";
    public string ShippingCity { get; set; } = "";
    public decimal Total { get; set; }
    public string? CardLast4 { get; set; }
}

public sealed record SendOrderConfirmationEmailEvent(Guid OrderId);
public sealed record OrderUpdated(Order Order);
public sealed record OrderPlacedEvent(Guid OrderId);

Shipping ekibi, sipariş verildiğinde kargo hazırlığına başlamak istiyor. Bu üç message'tan hangisini dinlemeli? OrderUpdated geldiğinde kargo hazırlığı başlamalı mı, yoksa sipariş iptal mi edildi, adres mi değişti? SendOrderConfirmationEmailEvent'i Shipping de dinlemeye başlarsa ne olur? Bir de şu: OrderUpdated'in içindeki Order'ı dinleyen her ekip neyi görüyor?

Message: Zarf ve İçerik

Enterprise Integration Patterns kitabı (EIP: Gregor Hohpe ve Bobby Woolf'un 2003 tarihli, message tabanlı entegrasyon pattern'leri kataloğu) message'ı iki parçalı tarif ediyor: messaging sisteminin kullandığı bir header ve uygulamanın verisini taşıyan bir body. Kitap message'ları amaçlarına göre de ayırıyor: alıcıdan bir şey yapmasını isteyen command message, veri aktaran document message ve bir şeyin olduğunu bildiren event message. Bu üçünü birbirinden ayırmak, bir message'ın adını, sahibini ve kaç alıcısı olacağını belirliyor.

Bookstore'un ilk message'ları bu ayrım düşünülmeden yazılmış. Tek tek bakalım; önce her message'ın zarfına (envelope) koyacağımız bilgileri, sonra içeriklerini konuşalım.

Command, Event ve Query

Command bir emirdir: "ödemeyi al", "onay e-postasını gönder". Adı emir kipindedir (ChargePayment), tek bir mantıksal alıcısı vardır ve gönderen bir sonuç bekler. Alıcı reddedebilir: kart limiti yetmiyorsa ChargePayment başarısız olur. Ödemeyi Payments alır; aynı command'ı iki farklı consumer'ın işlemesi, parayı iki kez çekmek demektir.

Event bir olgudur: "sipariş verildi". Adı geçmiş zamandır (OrderPlaced), sıfır, bir ya da çok alıcısı olabilir ve yayıncı onları tanımaz. Olmuş bir şey reddedilemez; Inventory "ben bu siparişi kabul etmiyorum" diyemez, olsa olsa kendi tarafında bir tepki üretir (OutOfStock). Yarın Notification da OrderPlaced'i dinlemek isterse Ordering'de tek satır değişmez.

Query bir sorudur: "Dune'dan kaç tane var?" Cevabı beklenir. Message ile request/reply kurulabilir (EIP'te Request-Reply pattern'i, cevabın nereye döneceğini söyleyen Return Address ve hangi isteğe ait olduğunu söyleyen Correlation Identifier ile), ama bir önceki yazıda gördüğümüz gibi cevabı aynı HTTP (Hypertext Transfer Protocol) isteğinin içinde bekleyeceksen senkron çağrı daha dürüst bir seçim.

Contract sahipliği satırı pratikte en çok karıştırılanı. Command'ın şeklini, onu işleyecek olan belirler: Payments, ödemeyi almak için neye ihtiyacı olduğunu bilir ve ChargePayment'ı kendi contract'ı olarak yayınlar. Event'in şeklini ise yayıncı belirler: OrderPlaced'in içinde ne olacağına Ordering karar verir; dinleyen her ekip için ayrı bir alan eklemek yerine, olguyu kendi dilinde anlatır. Kodda bu, contract'ların namespace'ine yansıyor:

namespace Bookstore.Messaging;

public interface IMessage
{
    // Contract'ın kalıcı adı: CLR tip adı değişse de aynı kalır
    static abstract string MessageType { get; }
}

public interface ICommand : IMessage;
public interface IEvent : IMessage;

public sealed record MessageEnvelope<T>(
    Guid MessageId,
    string MessageType,
    Guid CorrelationId,
    Guid? CausationId,
    DateTimeOffset SentAt,
    T Body) where T : IMessage
{
    // Bir akışın ilk message'ı: correlation id kendisi
    public static MessageEnvelope<T> Start(T body, TimeProvider clock)
    {
        Guid id = Guid.CreateVersion7(clock.GetUtcNow());
        return new(id, T.MessageType, id, null, clock.GetUtcNow(), body);
    }

    // Bu message'ın sonucu olarak gönderilen yeni message: aynı correlation, causation = bu message
    public MessageEnvelope<TNext> Next<TNext>(TNext body, TimeProvider clock) where TNext : IMessage =>
        new(Guid.CreateVersion7(clock.GetUtcNow()), TNext.MessageType, CorrelationId, MessageId, clock.GetUtcNow(), body);
}

using Bookstore.Messaging;

namespace Bookstore.Ordering.Contracts
{
    public sealed record OrderedBook(Guid BookId, string Title, int Quantity, decimal UnitPrice);
    public sealed record ShippingAddress(string Name, string Street, string City, string PostalCode, string Country);

    // Event: geçmiş zaman, yayıncıya (Ordering) ait, alıcılarını tanımaz
    public sealed record OrderPlaced(
        Guid OrderId,
        Guid CustomerId,
        IReadOnlyList<OrderedBook> Items,
        decimal Total,
        string Currency,
        ShippingAddress ShipTo,
        DateTimeOffset PlacedAt) : IEvent
    {
        public static string MessageType => "bookstore.ordering.order-placed.v1";
    }

    // Event notification: yalnızca "oldu" ve id
    public sealed record OrderPlacedNotification(Guid OrderId) : IEvent
    {
        public static string MessageType => "bookstore.ordering.order-placed-notification.v1";
    }
}

namespace Bookstore.Payments.Contracts
{
    // Command: emir kipi, tek mantıksal alıcı (Payments), contract alıcıya ait
    public sealed record ChargePayment(Guid OrderId, decimal Amount, string Currency) : ICommand
    {
        public static string MessageType => "bookstore.payments.charge-payment.v1";
    }

    public sealed record PaymentCaptured(Guid OrderId, Guid PaymentId, decimal Amount, string Currency) : IEvent
    {
        public static string MessageType => "bookstore.payments.payment-captured.v1";
    }
}

MessageType, C# 11 ile gelen static abstract interface member olarak tanımlı: her contract kendi kalıcı adını taşıyor ve generic kod T.MessageType ile ona reflection'sız ulaşıyor. Neden typeof(T).Name değil? Çünkü CLR (Common Language Runtime) tip adı bir refactoring'de değişebilir; kablodaki ad ise bir contract'tır ve değiştiğinde dinleyen herkes kırılır. Sondaki .v1 da aynı nedenle orada; contract evrimi serinin ilerideki contract ve schema evolution yazısının konusu.

Envelope: Id'ler Neden Üç Tane?

MessageEnvelope<T> beş bilgi taşıyor ve her biri farklı bir soruya cevap veriyor:

• MessageId: bu message'ın kimliği. Aynı message iki kez gelirse bunu fark etmenin anahtarı; serinin idempotent consumer yazısında dedup key olarak kullanılacak.
• CorrelationId: bu message hangi iş akışına ait? Akıştaki bütün message'larda aynı.
• CausationId: bu message'ı hangi message doğurdu? Akışın ağacını kurmanı sağlar.
• SentAt: ne zaman gönderildi; tutarsızlıkları ve gecikmeyi açıklamak için.
• MessageType: body'nin nasıl okunacağı; consumer'ın hangi handler'ı çağıracağını bilmesi için.

Correlation ve causation farkını bir akışta görelim. Sipariş verildi, bu event'e tepki olarak ödeme command'ı gönderildi, ödeme alınınca yeni bir event yayınlandı:

using Bookstore.Messaging;
using Bookstore.Ordering.Contracts;
using Bookstore.Payments.Contracts;
using Bookstore.Shipping;

TimeProvider clock = TimeProvider.System;
OrderPlaced placed = new(
    OrderId: Guid.CreateVersion7(),
    CustomerId: Guid.CreateVersion7(),
    Items: [new(Guid.CreateVersion7(), "Dune", 2, 30m), new(Guid.CreateVersion7(), "Emma", 1, 20m)],
    Total: 80m,
    Currency: "EUR",
    ShipTo: new("Ada", "1 Harbour Street", "Izmir", "35000", "TR"),
    PlacedAt: clock.GetUtcNow());

// Bir akış: OrderPlaced -> ChargePayment -> PaymentCaptured
MessageEnvelope<OrderPlaced> first = MessageEnvelope<OrderPlaced>.Start(placed, clock);
MessageEnvelope<ChargePayment> charge = first.Next(new ChargePayment(placed.OrderId, placed.Total, placed.Currency), clock);
MessageEnvelope<PaymentCaptured> captured = charge.Next(
    new PaymentCaptured(placed.OrderId, Guid.CreateVersion7(), placed.Total, placed.Currency), clock);

static string Short(Guid? id) => id is { } g ? "…" + g.ToString("N")[^6..] : "-";
Console.WriteLine($"{"type",-40} {"id",-8} {"correlation",-12} causation");
Print(first.MessageType, first.MessageId, first.CorrelationId, first.CausationId);
Print(charge.MessageType, charge.MessageId, charge.CorrelationId, charge.CausationId);
Print(captured.MessageType, captured.MessageId, captured.CorrelationId, captured.CausationId);
void Print(string type, Guid id, Guid correlation, Guid? causation) =>
    Console.WriteLine($"{type,-40} {Short(id),-8} {Short(correlation),-12} {Short(causation)}");
Console.WriteLine();

type                                     id       correlation  causation
bookstore.ordering.order-placed.v1       …0ba94d  …0ba94d      -
bookstore.payments.charge-payment.v1     …451509  …0ba94d      …0ba94d
bookstore.payments.payment-captured.v1   …0b37c1  …0ba94d      …451509

Üç message'ın correlation id'si aynı: hepsi aynı siparişin hikâyesi. Causation id ise zinciri kuruyor: ChargePayment'ı OrderPlaced doğurdu, PaymentCaptured'ı ChargePayment. Log'larda correlation id ile filtrelediğinde akışın bütün parçalarını, causation id ile de kimin kimi tetiklediğini görürsün. Bir önceki yazıda "checkout başarılı, worker üç dakika sonra başarısız" senaryosunu birbirine bağlayacak bir id'den bahsetmiştik; o id bu. EventStoreDB gibi event store'lar aynı ikiliyi $correlationId ve $causationId metadata alanları olarak tutar. Tek bir standarda bağlı kalmak istersen CNCF'nin (Cloud Native Computing Foundation) CloudEvents spesifikasyonu da id, source, type ve specversion alanlarını zorunlu tutan ortak bir zarf tanımlıyor; ayrıntısı serinin contract yazısında.

Bir not: correlation id, dağıtık tracing'deki trace id ile aynı şey değil. Trace id teknik bir izdir ve her HTTP isteğinde yenidir; correlation id ise iş akışının kimliğidir, bir saga günler sürse de değişmez. İkisini message'la birlikte taşımak, serinin son yazısındaki gözlemlenebilirlik konusunun başlangıç noktası.

Event Notification mı, Event-Carried State Transfer mı?

Martin Fowler 2017'deki "What do you mean by Event-Driven?" yazısında "event-driven" diye anılan dört farklı şeyi birbirinden ayırıyor: event notification, event-carried state transfer, event sourcing ve CQRS (Command Query Responsibility Segregation). İlk ikisi doğrudan bir event'in içine ne koyacağımızla ilgili.

Event notification'da event incedir: "sipariş verildi, id'si şu". Ayrıntı isteyen alıcı yayıncıyı geri çağırır. Event-carried state transfer'da ise event, alıcının ihtiyaç duyduğu veriyi taşır: alıcı kendi tarafında bir kopya tutar ve yayıncıya sormaz. Shipping'in kargo taslağı için alıcının adına, şehrine ve kitap sayısına ihtiyacı var. İki yolu da yazalım ve Ordering'in API'si (Application Programming Interface) çökmüşken çalıştıralım:

using Bookstore.Ordering.Contracts;

namespace Bookstore.Shipping;

public sealed record ShipmentDraft(Guid OrderId, string RecipientName, string City, int BookCount);

// Ordering'in HTTP API'sini taklit ediyor
public sealed class OrderingApiClient(Dictionary<Guid, OrderPlaced> orders)
{
    public bool IsDown { get; set; }
    public int Calls { get; private set; }

    public Task<OrderPlaced> GetOrderAsync(Guid orderId)
    {
        Calls++;
        if (IsDown) throw new HttpRequestException("Ordering API unavailable");
        return Task.FromResult(orders[orderId]);
    }
}

// Event notification: ayrıntı için yayıncıyı geri çağırıyor
public sealed class ShippingOnNotification(OrderingApiClient ordering, List<ShipmentDraft> drafts)
{
    public async Task HandleAsync(OrderPlacedNotification message)
    {
        OrderPlaced order = await ordering.GetOrderAsync(message.OrderId);
        drafts.Add(new(order.OrderId, order.ShipTo.Name, order.ShipTo.City, order.Items.Sum(i => i.Quantity)));
    }
}

// Event-carried state transfer: ihtiyaç duyduğu veri event'in içinde
public sealed class ShippingOnStateTransfer(List<ShipmentDraft> drafts)
{
    public Task HandleAsync(OrderPlaced message)
    {
        drafts.Add(new(message.OrderId, message.ShipTo.Name, message.ShipTo.City, message.Items.Sum(i => i.Quantity)));
        return Task.CompletedTask;
    }
}

// Ordering API çökmüşken iki Shipping consumer'ı
OrderingApiClient orderingApi = new(new() { [placed.OrderId] = placed }) { IsDown = true };
List<ShipmentDraft> fromNotification = [], fromState = [];
try
{
    await new ShippingOnNotification(orderingApi, fromNotification).HandleAsync(new(placed.OrderId));
}
catch (HttpRequestException ex)
{
    Console.WriteLine($"[notification]   failed: {ex.Message} (API calls: {orderingApi.Calls})");
}
await new ShippingOnStateTransfer(fromState).HandleAsync(placed);
Console.WriteLine($"[state transfer] {fromState[0]}");

[notification]   failed: Ordering API unavailable (API calls: 1)
[state transfer] ShipmentDraft { OrderId = 01a1b834-8ca6-72e7-a3ed-c09ca1d2bcc3, RecipientName = Ada, City = Izmir, BookCount = 3 }

Notification ile çalışan consumer, ilk adımda Ordering'e bağımlı. Bir önceki yazıda kırmaya çalıştığımız temporal coupling, event'in arkasından geri geldi: event asenkron, ama onu işlemek için yapılan çağrı senkron. Üstelik yük de taşınıyor; günde 800 sipariş ve OrderPlaced'i dinleyen üç context (Inventory, Shipping, Notification) varsa, Ordering'in API'si günde 2.400 ek çağrı alır. Kampanya zirvesinde bu çağrılar tam da Ordering'in en meşgul olduğu ana denk gelir.

State transfer ile çalışan consumer, Ordering kapalıyken bile işini yaptı. Fowler'ın saydığı faydalar bunlar: dayanıklılık, daha düşük latency, yayıncıya daha az yük. Bedeli de var: veri kopyalanıyor ve kopyalar bayatlayabilir. Ada siparişten sonra adresini değiştirirse Shipping'in elindeki kopya eskir; Ordering'in bunun için ayrı bir event (ShippingAddressChanged) yayınlaması ve Shipping'in onu da işlemesi gerekir. Ayrıca event büyüdükçe contract'ı da büyür: içine koyduğun her alan, artık geri alması zor bir sözdür.

Pratik bir kural: event'e, olgunun kendisini anlatmak için gereken veriyi koy. OrderPlaced için bu; sipariş satırları, tutar, para birimi ve teslimat adresi, yani "sipariş verildi" cümlesinin kendisi. Müşterinin sadakat puanı ya da kitabın kategori ağacı bu olgunun parçası değil; onlara ihtiyaç duyan, kendi kaynağından alır. Microsoft'un idempotent consumer rehberi de state transfer'in bir yan faydasına dikkat çekiyor: event mutlak sonucu taşıdığında consumer onu bir upsert olarak uygulayabilir ve aynı event'in ikinci kez gelmesi zarar vermez.

Domain Event ile Integration Event

Domain-Driven Design serisinin "Domain Event'ler: Raise, Topla, SaveChanges'te Dispatch Et" yazısında OrderPlaced zaten vardı: aggregate'in içinde raise edilen, aynı process'teki handler'lara dağıtılan bir domain event. Bu yazıdaki Bookstore.Ordering.Contracts.OrderPlaced ise bir integration event: process'ten çıkıp broker'a giden, başka bounded context'lerin okuduğu, sürümlenmiş bir contract. Adları aynı olabilir ama iki ayrı tiptir.

Ayrımın pratik nedeni değişim hızı. Domain event, modelle birlikte serbestçe değişir: OrderItem'a bir alan eklemek, onu kullanan handler'ları derleyicinin yardımıyla düzeltmek demek. Integration event'i ise başka ekipler başka repository'lerde deserialize ediyor; derleyici onları göremez. Bu yüzden domain event'ten integration event'e dönüşümü açıkça yazmak (strongly typed id'leri Guid'e, Money'yi tutar ve para birimine çevirerek) bir maliyet değil, bir sınır. Dönüşümün nerede yapılacağı ve veritabanı commit'iyle birlikte nasıl güvenilir yayınlanacağı serinin outbox yazısının konusu.

Sık Düşülen Tuzaklar

Event kılığında command

SendOrderConfirmationEmailEvent'in adında "Event" yazıyor ama içeriği bir emir: tek bir alıcısı ve beklenen bir sonucu var. Fowler bunu "passive-aggressive command" diye adlandırıyor: gönderen, alıcının bir şey yapmasını bekliyor ama bunu bir olgu gibi duyuruyor. Shipping bu event'i dinlemeye başlarsa e-posta göndermek zorunda mı kalacak? Ya Notification ekibi e-postayı SMS (Short Message Service) ile değiştirmek isterse? Olguyu yayınla (OrderPlaced) ve tepkiyi alıcıya bırak; gerçekten bir emir veriyorsan onu açıkça bir command olarak gönder (SendOrderConfirmation).

Entity'nin tamamını event'e koymak

OrderUpdated(Order Order) kolay görünür: ne lazımsa içinde. Ama Order class'ının her property'si artık bir contract'tır. CardLast4'ü ödeme ekibi oraya debug için koymuştu; şimdi onu dinleyen her context'e, her log'a ve her dead-letter queue'ya gidiyor. Ordering ekibi ShippingCity'yi bir Address value object'ine çevirdiğinde de beş ekibin consumer'ı aynı gün kırılıyor. Event'in şeklini entity'den bağımsız tasarla.

CRUD event'leri

OrderUpdated neyin değiştiğini söylemiyor. Shipping'in her güncellemede siparişin önceki hâlini saklayıp farkı kendisi bulması gerekiyor: iptal mi, adres değişikliği mi, ödeme mi? Niyet kayboldu. OrderCancelled, ShippingAddressChanged, PaymentCaptured gibi iş dilinden gelen event'ler, consumer'ların yalnızca ilgilendikleri olguları dinlemesini sağlar. Domain-Driven Design serisinin "Ubiquitous Language" yazısındaki kural burada da geçerli: event adı, domain expert'in cümlesinde geçebilmeli.

Bir command'ı yayınlamak

ChargePayment'ı bir publish/subscribe topic'ine göndermek, ileride ikinci bir subscriber eklendiğinde parayı iki kez çekmenin kapısını açar. Command, tek bir alıcıya giden bir channel'a gönderilir; channel türleri bir sonraki yazının konusu.

Küçük Bir Deneme

★☆☆ Bookstore'dan beş message adı yaz (her context'ten en az bir tane) ve her birini command, event ya da query olarak sınıflandır. Her biri için contract'ın sahibi kim?

★★☆ OrderPlaced'e bir CustomerEmail alanı eklemek istiyorsun, çünkü Notification ekibi istiyor. Bunu event-carried state transfer'in doğal bir parçası olarak mı görürsün, yoksa yayıncının alıcıya göre şekillenmesi olarak mı? Notification'ın e-posta adresini nereden alması gerekir?

★★★ Next ile üretilmiş message'lar için küçük bir "akış ağacı" yazdırıcısı yaz: birkaç message'ı listede tut, correlation id ile grupla ve causation id'ye göre girintili bir ağaç olarak bas. PaymentCaptured'dan sonra hem OrderShipped hem de bir e-posta command'ı gelirse ağaç nasıl görünüyor?

Message'ın ne olduğunu ve zarfında ne taşıdığını konuştuk. Bir sonraki yazıda onu taşıyan yola bakacağız: point-to-point queue, publish/subscribe topic ve competing consumer'lar, ve aynı kavramların RabbitMQ ile Kafka'da nasıl farklı göründüğü.