Event Sourcing System — Technical Analysis

Azure Service Bus (Queue) + Cosmos DB + .NET 10

---

Section 1 — Full Flow: Step by Step

Phase 1: Application Bootstrap (Program.cs)

Step 1 — Infrastructure setup The app starts by creating a CosmosClient pointed at the local emulator (https://localhost:8081). Because the emulator uses a self-signed certificate, SSL validation is disabled with (_, _, _, _) => true.

Step 2 — Service Bus queue provisioning A ServiceBusAdministrationClient checks if the queue bank-account-events exists in the real Azure namespace testing-gabgreg.servicebus.windows.net. If not, it creates it. Then a ServiceBusSender is created for that queue.

Step 3 — Store initialization CosmosDbEventStore.InitializeAsync() calls CreateDatabaseIfNotExistsAsync("event-sourcing") and then CreateContainerIfNotExistsAsync("events", partitionKey: "/aggregateId"). The snapshot store does the same for the "snapshots" container (partitioned by /id). Both containers are created only if missing — safe to call on every startup.

---

Phase 2: Executing a Command (account.Deposit(10))

Step 4 — Business logic runs in memory BankAccount.Deposit(amount) creates a MoneyDepositedEvent, calls ApplyChanges(event), which calls When(event) to mutate state (Balance += amount, Version++) and adds the event to the private _changes list. Nothing is written to any store yet.

Step 5 — Save is triggered every 5 events repository.SaveAsync(account) is called. It reads account.GetChanges() (the 5 uncommitted events) and calculates expectedVersion = account.Version - changes.Count.

---

Phase 3: Persistence to Cosmos DB

Step 6 — Optimistic concurrency check CosmosDbEventStore.SaveAsync() runs a COUNT query against Cosmos: SELECT VALUE COUNT(1) FROM c WHERE c.aggregateId = @id If the count doesn't match expectedVersion, it throws — meaning another writer already added events. This protects against concurrent writes.

Step 7 — Each event is inserted as a document For each event in the batch, an EventDocument is created:

{
  "id": "<new-guid>",
  "aggregateId": "<account-guid>",
  "version": 3,
  "eventType": "MoneyDepositedEvent",
  "payload": "{\"Amount\":10}",
  "occurredAt": "2026-03-30T..."
}

payload is the event serialized with System.Text.Json. The document itself (its field names) is serialized by the Cosmos SDK using Newtonsoft.Json (hence the [JsonProperty] attributes on EventDocument).

Step 8 — Snapshot check After saving events, the repository reads the current snapshot from Cosmos and checks: account.Version - snapshotVersion >= 50 If true, a snapshot document is upserted (insert or replace) into the snapshots container capturing the current Balance and Version.

---

Phase 4: Publishing to Azure Service Bus

Step 9 — Publish after persistence Only after Cosmos has confirmed the write does _publisher.PublishAsync() run. This is the save-first, publish-after guarantee: if the process crashes before publishing, the events are still safe in Cosmos.

Step 10 — Message construction For each event, a ServiceBusMessage is created wrapping an EventEnvelope:

{
  "EventType": "MoneyDepositedEvent",
  "AggregateId": "<guid>",
  "Version": 1,
  "OccurredAt": "2026-03-30T...",
  "Payload": "{\"Amount\":10}"
}

The message also sets ApplicationProperties["EventType"] = "MoneyDepositedEvent" — a header that would allow consumers to filter messages without deserializing the body.

Step 11 — Batch send All messages are assembled into a ServiceBusMessageBatch and sent in a single call to sender.SendMessagesAsync(batch). The queue now holds the messages waiting for a consumer.

---

Phase 5: Loading (Read Path)

Step 12 — Check for snapshot repository.LoadAsync(id) first calls CosmosDbSnapshotStore.GetAsync(id) — a direct ReadItemAsync by document ID (O(1) point read). If a snapshot exists, account.LoadFromSnapshot(snapshot) sets Balance and Version without replaying any events.

Step 13 — Replay only post-snapshot events CosmosDbEventStore.GetEventsAsync(id, fromVersion: snapshot.Version) runs: SELECT * FROM c WHERE c.aggregateId = @id AND c.version > @fromVersion ORDER BY c.version Only the events after the snapshot are returned and replayed via account.LoadFromHistory(events).

Step 14 — Polymorphic deserialization For each EventDocument, DeserializeEvent() does: Type.GetType("EventSourcingTranning.Events.MoneyDepositedEvent, EventSourcingTranning") It looks up the C# type at runtime by name, then uses System.Text.Json to deserialize the payload string into a concrete MoneyDepositedEvent object.

---

Section 2 — Architecture Diagram

┌─────────────────────────────────────────────────────────────────┐
│                        Program.cs                               │
│  BankAccount.Deposit() / Withdrawn()                            │
│         │                                                       │
│         ▼                                                       │
│  BankAccountRepository.SaveAsync()                              │
│         │                                                       │
│    ┌────┴──────────────────────────────┐                        │
│    │                                   │                        │
│    ▼                                   ▼                        │
│  [1] CosmosDbEventStore          [2] CosmosDbSnapshotStore      │
│      SaveAsync()                      SaveAsync()  (if ≥50)     │
│      INSERT document                  UPSERT document           │
│         │                                                       │
│    ┌────┘                                                       │
│    ▼                                                            │
│  [3] ServiceBusEventPublisher                                   │
│      PublishAsync()                                             │
│      SendMessagesAsync(batch)                                   │
└───────────────────────┬─────────────────────────────────────────┘
                        │
         ┌──────────────┴──────────────┐
         ▼                             ▼
  ┌─────────────┐             ┌────────────────────┐
  │  Cosmos DB  │             │  Azure Service Bus │
  │  (local     │             │  (real Azure)      │
  │  emulator)  │             │                    │
  │             │             │  Queue:            │
  │  Database:  │             │  bank-account-     │
  │  event-     │             │  events            │
  │  sourcing   │             │                    │
  │             │             │ [msg1][msg2][msg3] │
  │  ┌────────┐ │             │ waiting for        │
  │  │events  │ │             │ a consumer         │
  │  │(append)│ │             └────────────────────┘
  │  └────────┘ │
  │  ┌─────────┐│
  │  │snapshot ││
  │  │(upsert) ││
  │  └─────────┘│
  └─────────────┘

LOAD PATH:
  BankAccountRepository.LoadAsync()
       │
       ├──► CosmosDbSnapshotStore.GetAsync()   [point read - fast]
       │         │
       │    snapshot? ──YES──► LoadFromSnapshot() → set Balance + Version
       │         │
       │         NO──────────► start from version 0
       │
       └──► CosmosDbEventStore.GetEventsAsync(fromVersion)
                 │
            replay remaining events
                 │
            BankAccount state restored ✓

---

Section 3 — Component Responsibilities

Component	File	Responsibility
BankAccount	Entities/BankAccount.cs	Domain aggregate. Holds business rules (overdraft check). Applies events to mutate state. Tracks uncommitted changes in _changes.
Event	Entities/Event.cs	Base class for all domain events. Carries a timestamp.
MoneyDepositedEvent / MoneyWithdrawnEvent	Events/	Concrete facts that something happened. Immutable records of intent.
BankAccountRepository	Store/BankAccountRepository.cs	Orchestrator. Coordinates save (persist → snapshot check → publish) and load (snapshot → replay). The only class that knows all three stores.
CosmosDbEventStore	Store/CosmosDbEventStore.cs	Append-only event log in Cosmos. Handles optimistic concurrency, ordered storage, and polymorphic deserialization of events on read.
CosmosDbSnapshotStore	Store/CosmosDbSnapshotStore.cs	Stores one snapshot per aggregate. Uses upsert — always replaces with the latest. Enables fast load by skipping old events.
IEventPublisher	Messaging/IEventPublisher.cs	Interface. Decouples the repository from the concrete broker. Could be swapped for an in-memory publisher in tests.
ServiceBusEventPublisher	Messaging/ServiceBusEventPublisher.cs	Wraps events in EventEnvelope, serializes to JSON, batches them, and sends to Service Bus queue.
EventEnvelope	Messaging/EventEnvelope.cs	Wire format. Adds EventType, AggregateId, Version, OccurredAt metadata to the raw event payload.
BankAccountProjections	Projections/BankAccountProjections.cs	In-memory read model builder. Exists from the earlier version of the project — not currently wired into the main flow.
ConnectionStrings	Config/ConnectionStrings.cs	Holds both connection strings as constants. (Dev-only pattern.)

---

Section 4 — Failure Scenarios

4.1 — Message Processing Failure (Consumer Crashes)

What happens today: The messages sit in the bank-account-events queue. Azure Service Bus has a built-in lock mechanism: when a consumer picks up a message, it becomes invisible to others for a configurable lock duration (default 60 seconds). If the consumer crashes without completing or abandoning, the lock expires and the message becomes visible again. After a configurable max delivery count (default 10), the message moves to the Dead Letter Queue (DLQ). Current gap: This project has no consumer. The messages are published and just accumulate. There is no dead-letter handling or monitoring.

4.2 — Duplicate Messages (At-Least-Once Delivery)

What happens: Service Bus guarantees at-least-once delivery — not exactly-once. If the publisher sends a batch and the acknowledgement is lost (network blip), the SDK retries and the same messages could be sent twice. Current gap: There is no deduplication key set on ServiceBusMessage. The MessageId property (which Service Bus uses for duplicate detection on Standard tier) is not populated. Consumers must be idempotent — meaning processing the same event twice must produce the same result.

4.3 — Communication Issues

Cosmos DB unavailable:

• CreateDatabaseIfNotExistsAsync or CreateItemAsync will throw a CosmosException (503 or timeout).
• There is no retry policy configured on the CosmosClient. The SDK has some built-in retries, but they are conservative.
• The app will crash without a clear user-facing error.

Service Bus unavailable:

• SendMessagesAsync will throw. Because events were already written to Cosmos, the publish step fails but the data is safe. However, downstream consumers will not be notified.
• This creates a silent gap: Cosmos is consistent, Service Bus is not. This is the classic problem the Outbox Pattern solves.

4.4 — Data Inconsistency

The save-then-publish gap: The code does:

1. CosmosDbEventStore.SaveAsync() ← succeeds
2. CosmosDbSnapshotStore.SaveAsync() ← succeeds
3. IEventPublisher.PublishAsync() ← CRASHES HERE Events are durable in Cosmos but the Service Bus message was never sent. Consumers never learn the events happened. The event log and the message queue are now out of sync.

The concurrency check gap: The optimistic concurrency check is a COUNT query followed by individual CreateItemAsync calls — not atomic. In theory, two writers could both pass the count check before either inserts, leading to duplicate version numbers.

Snapshot vs. event store divergence: The snapshot is saved after events but in a separate container and a separate Cosmos operation. If the process crashes between inserting events and upserting the snapshot, the next load will replay more events than expected — which is correct and safe, just slower.

---

Section 5 — Suggestions for Production

#	Issue	Production Fix
1	Secrets in source code	Move connection strings to environment variables, Azure Key Vault, or appsettings.json + User Secrets.
2	No message consumer	Implement a ServiceBusProcessor or Azure Function with ServiceBusTrigger to consume and process messages.
3	Publish gap (no Outbox)	Implement the Transactional Outbox Pattern: store pending messages in Cosmos alongside events, have a background worker poll and publish them reliably.
4	No idempotency	Set message.MessageId = Guid.NewGuid().ToString() and design consumers to be idempotent (check if event was already processed).
5	No retry/resilience	Add Polly for retry policies on Cosmos and Service Bus calls; configure CosmosClientOptions.MaxRetryAttemptsOnRateLimitedRequests.
6	SSL disabled globally	In production use a real certificate; for dev, trust the emulator cert properly instead of disabling all validation.
7	No dead-letter monitoring	Add alerting when messages land in the Dead Letter Queue (Azure Monitor or Application Insights).
8	Non-atomic concurrency	Replace the COUNT + INSERT pattern with Cosmos ETags / optimistic concurrency using ItemRequestOptions.IfMatchEtag, or use a Cosmos stored procedure for atomic append.
9	Upgrade Service Bus tier	Move to Standard tier to unlock Topics + Subscriptions, enabling multiple independent consumers per event type.
10	No structured logging	Replace Console.WriteLine with Serilog or Microsoft.Extensions.Logging with correlation IDs per aggregate operation.

---

Section 6 — System Checklist

STARTUP

• CosmosClient created with Gateway mode (required for emulator)
• SSL validation bypassed for local emulator
• Service Bus queue created if missing (via AdministrationClient)
• Cosmos database + containers created if missing (InitializeAsync)

WRITE PATH

• Command runs on BankAccount in memory (no I/O)
• Event is applied immediately (state mutates + added to _changes)
• Version increments with every event applied
• SaveAsync: optimistic concurrency check (COUNT query)
• SaveAsync: each event inserted as document in Cosmos (append-only)
• SaveAsync: snapshot upserted if version - snapshotVersion >= 50
• SaveAsync: events published to Service Bus queue AFTER Cosmos write
• _changes cleared after successful save

READ PATH

• LoadAsync: point-read snapshot from Cosmos (fast, O(1))
• LoadAsync: if snapshot found, set Balance + Version from snapshot
• LoadAsync: query only events AFTER snapshot.Version
• LoadAsync: if no snapshot, query all events from beginning
• LoadAsync: polymorphic deserialization by event type name
• LoadAsync: replay events via When() to rebuild state

SERVICE BUS

• Each event wrapped in EventEnvelope (type + metadata + payload)
• Messages batched with ServiceBusMessageBatch
• ApplicationProperties["EventType"] set for future filtering
• No consumer implemented
• No MessageId set for deduplication
• No dead-letter handling

MISSING FOR PRODUCTION

• Secrets management
• Retry / resilience policies
• Outbox pattern (publish gap safety)
• Idempotent consumers
• Structured logging
• Dead-letter monitoring

---

Section 7 — 15 Questions to Test Your Understanding

1. Why are events applied to the aggregate in memory first, before anything is saved to Cosmos DB? What would go wrong if you wrote to Cosmos before mutating the aggregate state?
2. BankAccountRepository.SaveAsync() saves to Cosmos first and publishes to Service Bus second. What failure scenario does this ordering protect against, and what failure scenario does it not protect against?
3. The EventDocument class uses [JsonProperty("id")] from Newtonsoft.Json, while the event payload is serialized with System.Text.Json. Why are two different serializers used, and what would break if you used only System.Text.Json for both?
4. The Cosmos partition key for the events container is /aggregateId. What advantage does this give when querying all events for a single bank account? What would happen if you used /id as the partition key instead?
5. The optimistic concurrency check uses a COUNT query followed by individual CreateItemAsync calls. Why is this not fully atomic, and in what scenario could it allow duplicate version numbers?
6. Why does CosmosDbSnapshotStore.SaveAsync() use UpsertItemAsync instead of CreateItemAsync? What would happen if you used CreateItemAsync for snapshots?
7. When loading a BankAccount, the code calls GetEventsAsync(id, snapshot.Version) which queries WHERE c.version > @fromVersion. Why > and not >=? What would happen with >=?
8. The EventEnvelope wraps the event payload with an EventType field. Why can't a consumer just look at the JSON payload alone to figure out what type of event it received?
9. message.ApplicationProperties["EventType"] is set on every Service Bus message. What is this used for, and why would it be especially valuable if the namespace was upgraded to Standard tier with Topics?
10. The snapshot threshold is set to 50. If an account has 149 events and a snapshot at version 50, how many events does LoadAsync need to replay? How many would it replay if there were no snapshot at all?
11. Why does ServiceBusEventPublisher use a ServiceBusMessageBatch instead of sending messages one by one? What problem does TryAddMessage returning false indicate?
12. The DeserializeEvent method uses Type.GetType("EventSourcingTranning.Events.MoneyDepositedEvent, EventSourcingTranning"). What happens at runtime if you rename the MoneyDepositedEvent class or move it to a different namespace, but old documents still reference the old name?
13. The IEventPublisher interface is injected into BankAccountRepository instead of using ServiceBusEventPublisher directly. What are two concrete benefits of this design decision?
14. The Cosmos DB emulator requires ServerCertificateCustomValidationCallback = (_, _, _, _) => true. Why would this be dangerous in a production environment, and what should you use instead?
15. The _changes list is cleared at the end of SaveAsync. What would happen if ClearChanges() was not called after saving, and SaveAsync was called a second time on the same aggregate?

If you wanna help improve this training or even teach me talk to me https://www.gabrielgregori.com/