Practical Event Sourcing with CQRS in a Go Service

# The Problem with "Just Update the Row"

We are building a newsletter service. People subscribe with their email, confirm via a double opt-in link, and occasionally unsubscribe. The natural first instinct is a single database table:

CREATE TABLE subscribers (
    id UUID PRIMARY KEY,
    email_address TEXT NOT NULL,
    status TEXT NOT NULL,
    updated_at TIMESTAMP
);

When someone confirms, we run UPDATE subscribers SET status = 'active' WHERE id = ... and move on. It works. It ships. Nobody complains — until someone does.

"When exactly did this person confirm their subscription?" We don't know. We only stored the current status. "There might have been a bug in last Tuesday's deployment — how many people were incorrectly unsubscribed?" We can't tell. The evidence is gone. "Can we undo a batch operation that ran against the wrong segment?" Not without a backup and a lot of luck.

The row tells us where we are. It says nothing about how we got here. Every UPDATE statement destroys the previous state.

What if we flipped the model? Instead of storing the current state and overwriting it on every change, we store every state transition as an immutable fact. The current state becomes something we derive by replaying those facts from the beginning. This is Event Sourcing — and for domains with meaningful state transitions, it changes what questions we can answer.

---

# What is an Event?

An event is something that already happened. It's named in past tense. It's immutable — we can't reject it or undo it, because it describes a fact that already occurred.

This is different from a command. A command is a request: "please subscribe this email address." It might fail — the email might be invalid, the subscriber might already exist. An event is the outcome: "a subscription was initiated." By the time we have an event, the decision has already been made.

Events carry only the data relevant to what changed, not the full entity state. Here's what that looks like in Go:

type DomainEventType string

type DomainEvent interface {
    Type() DomainEventType
}

And a concrete event:

type SubscribeInitiated struct {
    EmailAddress string
}

var SubscribeInitiatedType DomainEventType = "SubscribeInitiated"

func (event SubscribeInitiated) Type() DomainEventType {
    return SubscribeInitiatedType
}

Our subscriber lifecycle has exactly three events: SubscribeInitiated, SubscribeConfirmed, and Unsubscribed. Together, they describe every possible state transition. A subscriber's entire history is a sequence of these events — and that sequence is all we need to reconstruct their current state at any point in time.

---

# Aggregates — Guarding the Rules

Events don't appear out of thin air. If anyone could emit any event at any time, we'd quickly end up with nonsensical histories — a confirmation before a subscription, an unsubscribe on someone who never existed. Something needs to enforce the rules. That's the aggregate.

An aggregate is a consistency boundary. It's the gatekeeper that ensures no operation can leave the domain in an invalid state. In event sourcing, it has two responsibilities:

1. Command handling: validate the incoming request and decide which event to emit — or reject the request entirely.
2. Event application: update internal state when an event is applied, so future decisions are based on the correct current state.

We use Go generics to build a reusable aggregate root that handles the mechanical parts — tracking uncommitted events, managing versions, and routing events to their apply functions:

type EventSourcedAggregateRoot[TAggregate any, TID fmt.Stringer] struct {
    ID                TID
    uncommittedEvents []DomainEvent
    version           int64
    applications      map[DomainEventType]func(*TAggregate, DomainEvent) (*TAggregate, error)
}

Our Subscriber aggregate embeds this and registers which function handles which event type:

type Subscriber struct {
    domain.EventSourcedAggregateRoot[Subscriber, uuid.UUID]
    emailAddress values.EmailAddress
    status       values.SubscriberStatus
}

func NewSubscriber(id uuid.UUID, domainEvents []domain.DomainEvent) (*Subscriber, error) {
    applications := map[domain.DomainEventType]func(*Subscriber, domain.DomainEvent) (*Subscriber, error){
        events.SubscribeInitiatedType: (*Subscriber).applySubscribeInitiated,
        events.SubscribeConfirmedType: (*Subscriber).applySubscribeConfirmed,
        events.UnsubscribedType:       (*Subscriber).applyUnsubscribed,
    }

    subscriber := &Subscriber{
        EventSourcedAggregateRoot: domain.NewEventSourcedAggregateRoot(id, domainEvents, applications),
    }

    for _, domainEvent := range domainEvents {
        subscriber, _ = subscriber.ApplyDomainEvent(subscriber, domainEvent)
    }

    return subscriber, nil
}

Notice what happens in NewSubscriber: we create a blank subscriber and replay all historical events through the apply functions. This is how reconstitution works — there's no separate "load from database columns" code path. The events are the persistence, and the apply functions are the deserialization logic.

Now, command handling. When a request comes in to confirm a subscription, the aggregate checks whether that's a valid transition:

func (subscriber *Subscriber) ConfirmSubscribe() (*Subscriber, error) {
    if subscriber.status == values.SubscriberStatusInitiated {
        return subscriber.EventSourcedAggregateRoot.RecordDomainEvent(
            subscriber,
            &events.SubscribeConfirmed{},
        )
    }
    return subscriber, nil
}

The guard clause (if status == Initiated) makes the business rule explicit and visible. We can only confirm a subscription that was previously initiated — not one that's already active, and not one that was unsubscribed. If the precondition fails, no event is emitted. The aggregate simply returns itself unchanged.

RecordDomainEvent does two things: it applies the event (so the aggregate's internal state updates immediately) and buffers it as uncommitted (so it can be persisted later). The apply function itself is pure state derivation — no I/O, no side effects:

func (subscriber *Subscriber) applySubscribeConfirmed(domainEvent domain.DomainEvent) (*Subscriber, error) {
    subscriber.status = values.SubscriberStatusActive
    return subscriber, nil
}

This separation between deciding what happened (command handling) and computing what it means (event application) is the core of the pattern. The same apply function is used both when recording new events and when replaying historical ones.

---

# The Event Store — Append-Only Persistence

We need somewhere to put events. Our event store is a single SQL table with four columns: bucket (the aggregate ID — which stream this event belongs to), type (the event type string), version (a monotonically increasing sequence number within the stream), and payload (the event data as JSON).

CREATE TABLE event_streams (
    bucket  TEXT    NOT NULL,
    type    TEXT    NOT NULL,
    version BIGINT  NOT NULL,
    payload JSONB   NOT NULL
);

This table is append-only. We never UPDATE or DELETE rows. That's the entire point — the event stream is an immutable log. This constraint is what gives us the audit trail, the ability to answer temporal queries, and the option to rebuild state from scratch at any time.

Loading an aggregate means reading its stream and replaying:

func (s *SQLEventStore) ReadEvents(bucket fmt.Stringer) ([]domain.DomainEvent, error) {
    rows, _ := s.readOnlyContext.
        Select("type", "payload").
        From(goqu.S(s.schema.String()).Table("event_streams")).
        Where(goqu.C("bucket").Eq(bucket.String())).
        Order(goqu.C("version").Asc()).
        Executor().Query()

    var events []domain.DomainEvent
    for rows.Next() {
        var eventType string
        var payload []byte
        rows.Scan(&eventType, &payload)

        event, _ := s.registry.Create(domain.DomainEventType(eventType))
        json.Unmarshal(payload, &event)
        events = append(events, event)
    }
    return events, nil
}

A type registry maps event type strings back to concrete Go structs for deserialization. This keeps the event store generic — it doesn't import any domain packages and works with any aggregate.

Now the critical part: concurrency control. Imagine two HTTP requests arrive simultaneously, both loading the same subscriber at version 3. Both validate their command against the same state. Both try to append an event. Without protection, both would append events based on stale state, creating an inconsistent history.

We solve this with optimistic concurrency — a version check at write time:

func (s *SQLEventStore) AppendEvents(
    bucket fmt.Stringer,
    version int64,
    events ...domain.DomainEvent,
) (int64, error) {
    var currentVersion int64
    s.readOnlyContext.
        Select(goqu.COALESCE(goqu.MAX("version"), 0)).
        From(goqu.S(s.schema.String()).Table("event_streams")).
        Where(goqu.C("bucket").Eq(bucket.String())).
        Executor().ScanVal(&currentVersion)

    if currentVersion > version {
        return 0, fmt.Errorf("version mismatch: current %d > provided %d",
            currentVersion, version)
    }

    for i, event := range events {
        payload, _ := json.Marshal(event)
        s.writeContext.
            Insert(goqu.S(s.schema.String()).Table("event_streams")).
            Rows(goqu.Record{
                "bucket":  bucket.String(),
                "type":    event.Type(),
                "version": currentVersion + int64(i) + 1,
                "payload": payload,
            }).Executor().Exec()
    }

    return currentVersion + int64(len(events)), nil
}

The aggregate remembers what version it was at when loaded. When we try to persist, we compare that against the current version in the database. If someone else wrote in between, the versions won't match, and we reject the write. The loser's request fails, and the caller must reload the aggregate (which now includes the winner's events), re-validate the command against the new state, and retry.

No explicit locks. No blocking. The first writer wins, and conflicts are detected rather than prevented. For most workloads, this is significantly cheaper than pessimistic locking — and for append-only stores, it's the natural fit.

---

# The Read Side Problem

At this point we have a clean write model. Events flow into an append-only stream, aggregates enforce invariants, concurrency is handled. But we've introduced a problem on the read side.

How do we answer: "give me all active subscribers, sorted by sign-up date"?

The naive approach would be: load every subscriber's event stream, replay each one to compute their current status, filter for "active", sort by the timestamp of their SubscribeInitiated event. This is O(subscribers * events-per-subscriber) — absurd for a simple list page.

We can't put an index on an event stream and query it efficiently. The write model is optimized for consistency and history preservation. But reads need flat rows, indexed columns, and efficient range scans.

This is where CQRS comes in: Command Query Responsibility Segregation. We maintain two separate data models:

• The write model is the event stream. It's optimized for appending events and loading aggregate history.
• The read model is one or more denormalized tables, shaped specifically for the queries we need to serve.

Commands flow to the write side — aggregates, business rules, event store. Queries read directly from the denormalized tables and never touch the event store. The two sides have different schemas, different access patterns, and different consistency guarantees.

The question is: how does the read side stay in sync with what the write side produces?

---

# Projectors — Bridging Write and Read

A projector is a component that listens to domain events and maintains a denormalized view. It's the bridge between the event stream and the query-optimized tables.

Our read model for subscribers is deliberately flat — a simple row per subscriber with pre-computed fields:

type SubscriberReadModel struct {
    ID           uuid.UUID `db:"id"`
    EmailAddress string    `db:"email_address"`
    Status       string    `db:"status"`
    CreatedAt    time.Time `db:"created_at"`
    UpdatedAt    time.Time `db:"updated_at"`
}

The projector subscribes to the events it cares about:

func (p *SubscriberProjector) Subscribe(eventBus *messaging.EventBus) {
    eventBus.Subscribe(string(events.SubscribeInitiatedType), p.handleSubscribeInitiated)
    eventBus.Subscribe(string(events.SubscribeConfirmedType), p.handleSubscribeConfirmed)
    eventBus.Subscribe(string(events.UnsubscribedType), p.handleUnsubscribed)
}

Each handler translates an event into a database operation on the read model. When a subscription is initiated, we insert a new row:

func (p *SubscriberProjector) handleSubscribeInitiated(payload []byte) {
    var eventMessage struct {
        AggregateID uuid.UUID
        Event       events.SubscribeInitiated
    }
    json.Unmarshal(payload, &eventMessage)

    readModel := read_models.SubscriberReadModel{
        ID:           eventMessage.AggregateID,
        EmailAddress: eventMessage.Event.EmailAddress,
        Status:       "initiated",
        CreatedAt:    time.Now(),
        UpdatedAt:    time.Now(),
    }

    unitOfWork.SubscriberReadModel().Insert(readModel)
}

When a subscriber confirms, we update the status to "active"; when they unsubscribe, we delete the row entirely. The read model always reflects the latest derived state — but it is derived, not authoritative. The event stream remains the single source of truth.

Here's the real power of this approach: we can add new projectors at any time, for any purpose, without touching the write side. Need a "subscribers grouped by email domain" report six months from now? Write a new projector, replay all historical events through it, and the new table is fully backfilled from day one. The events were already there — we just hadn't looked at them from that angle before. This is something you simply cannot do with destructive updates to a single row.

---

# The Complete Flow

Let's trace a single request end-to-end. A user clicks the confirmation link in their email:

HTTP Request (GET /confirm?token=abc)
  → ConfirmSubscribeCommand
    → Load Subscriber aggregate from event store (replay events)
      → Aggregate checks: status == Initiated? Yes.
        → RecordDomainEvent(SubscribeConfirmed{})
          → Persist event to event_streams table (version check passes)
            → Publish event to EventBus
              → SubscriberProjector receives event
                → UPDATE subscriber_read_model SET status = 'active'

Later, an admin opens the dashboard:

HTTP Request (GET /admin/subscribers)
  → ListSubscribersQuery
    → SELECT * FROM subscriber_read_model WHERE status = 'active'
      → Return results

The write path and read path share nothing except the events flowing between them. The aggregate never queries the read model. The query handler never touches the event store. They are separate systems connected by a stream of facts.

---

# Tradeoffs and When to Use This

We should be honest about what we're paying for this model.

Eventual consistency. After a command succeeds, the read model might not reflect the change immediately. The event needs to travel through the bus, hit the projector, and commit to the read model table. In practice this lag is measured in milliseconds for in-process projectors — but it exists. If a user confirms their subscription and immediately refreshes the page, they might briefly see stale data. For most applications this is acceptable. For some it isn't.

Operational complexity. We now have more components that can fail independently: the event store, the event bus, the projectors, the read model tables. A projector crash means the read model falls behind. We need monitoring, and we need a way to replay events to catch up.

Event schema evolution. Events are immutable. We wrote SubscribeInitiated with an EmailAddress field. If we later need to add a Source field to track where signups come from, we can't change existing events. We either introduce a new event type (SubscribeInitiatedV2), or we use upcasting to transform old events at read time. Both approaches work, but they require discipline.

Aggregate size. Reconstituting an aggregate replays all its events. A subscriber with 3 events is instant. An aggregate with 10,000 events gets noticeably slow. Snapshots (periodically storing the aggregate's computed state alongside the events) solve this, but add another layer of complexity.

When it's worth it: domains with complex state machines where the transitions matter as much as the current state. Systems that need a complete audit trail. Services that require multiple read models from the same underlying data. Situations where "how did we get here" is a question you actually need to answer.

When it's overkill: simple CRUD with no meaningful business rules. Services where the current state is all that matters and history has no value. Low-complexity domains where the additional infrastructure cost isn't justified by the benefits.

Event Sourcing is not about adopting a pattern because it's architecturally pleasing. It's about choosing the right foundation for domains where state transitions are the core of the problem — and where losing the history of those transitions means losing the ability to understand, debug, and evolve the system.