Event sourcing with MongoDB: what developers should know

Event sourcing with MongoDB: what developers should know

Event sourcing is a paradigm where every state change in your system is captured as an immutable event and stored permanently. Rather than updating a record in place, you append a new event describing what happened. This approach fundamentally changes how we think about data persistence, and MongoDB's flexible document model and strong write performance make it a practical choice for event store implementations. In this article, we'll explore the patterns and trade-offs that make event sourcing with MongoDB work in production systems.

Why event sourcing matters for your architecture

Traditional CRUD systems maintain a single, current state of an entity. When you update a user's email address, the old value is gone. Event sourcing inverts this: you record that "UserEmailChanged" happened, and the current state is always derived from replaying those events. This gives you a complete audit trail by design, enables temporal queries (what was the state at 3pm yesterday?), and makes debugging production issues much easier because you have a full history of state transitions.

The trade-off is complexity. You now need to maintain two concerns: the event store itself (the source of truth) and projections (read-optimized views built from events). For many teams, this complexity is worth it. We've seen it shine in e-commerce (order state transitions), SaaS (account changes with compliance audits), and financial systems where traceability is non-negotiable.

Append-only events collection

The foundation of event sourcing is a collection that only grows. You never update or delete events. This is surprisingly powerful because it eliminates concurrency issues at the storage layer and makes your event log immutable by design.

In MongoDB, we create a simple events collection with a document structure that captures the essence of what happened:

db.createCollection("events");

db.events.insertOne({
  aggregateId: "user-789",
  aggregateType: "User",
  eventType: "UserCreated",
  version: 1,
  timestamp: new Date("2024-01-15T10:30:00Z"),
  data: {
    email: "alice@example.com",
    name: "Alice Chen",
    plan: "free"
  },
  metadata: {
    userId: "user-789",
    source: "web-api",
    correlationId: "req-abc123"
  }
});

db.events.insertOne({
  aggregateId: "user-789",
  aggregateType: "User",
  eventType: "UserUpgraded",
  version: 2,
  timestamp: new Date("2024-01-16T14:22:00Z"),
  data: {
    plan: "premium",
    billingCycleStart: new Date("2024-01-16")
  },
  metadata: {
    userId: "user-789",
    source: "billing-service",
    correlationId: "req-def456"
  }
});

Notice the structure: every event is identified by the aggregateId (the entity it affects) and a version (a sequential counter for that aggregate). The eventType names the action, and data holds the payload. We also include metadata for tracing and operational context.

The append-only nature means inserts are fast and predictable. You're never contending with updates or deletes. In high-throughput systems, this is a significant advantage. MongoDB's write concerns can enforce durability guarantees; using writeConcern: { w: "majority" } ensures events are replicated before acknowledgment.

Index by aggregate ID and version

To reconstruct the state of an entity, you need to fetch all events for that aggregateId in version order. This is a frequent query pattern, so indexing is critical.

db.events.createIndex({ aggregateId: 1, version: 1 });
db.events.createIndex({ aggregateType: 1, timestamp: 1 });

The compound index on aggregateId and version enables efficient event replay: to load the current state of user-789, you query for all events where aggregateId === "user-789" ordered by version, then apply them sequentially to build state.

async function loadAggregate(aggregateId) {
  const events = await db.collection("events")
    .find({ aggregateId })
    .sort({ version: 1 })
    .toArray();
  
  let state = {};
  for (const event of events) {
    state = applyEvent(state, event);
  }
  return state;
}

function applyEvent(state, event) {
  switch (event.eventType) {
    case "UserCreated":
      return { ...state, email: event.data.email, plan: event.data.plan };
    case "UserUpgraded":
      return { ...state, plan: event.data.plan };
    case "UserEmailChanged":
      return { ...state, email: event.data.newEmail };
    default:
      return state;
  }
}

The secondary index on aggregateType and timestamp helps with analytics and projections. You might want to process all events of a certain type across all aggregates within a time window, and this index supports that efficiently.

In production, event replay can be slow if an aggregate has thousands of events. Many teams implement snapshots: periodic stored states that let you replay only recent events. A simple snapshot document might live in a separate collection:

db.snapshots.insertOne({
  aggregateId: "user-789",
  aggregateType: "User",
  version: 50,
  state: { email: "alice@example.com", plan: "premium", ... },
  timestamp: new Date("2024-01-10T09:00:00Z")
});

// Later, load from snapshot and replay events since version 50
const snapshot = await db.snapshots.findOne({ aggregateId });
let state = snapshot ? snapshot.state : {};
const events = await db.events
  .find({ 
    aggregateId, 
    version: { $gt: snapshot?.version || 0 } 
  })
  .sort({ version: 1 })
  .toArray();
// Apply remaining events...

Unique index prevents conflicting versions

Here's a subtle but crucial detail: you want to guarantee that version numbers for a given aggregate are never duplicated. If two concurrent writes both try to insert version 3 for the same aggregate, only one should succeed. MongoDB's unique indexes enforce this constraint at the database level.

db.events.createIndex(
  { aggregateId: 1, version: 1 },
  { unique: true }
);

This unique index is your concurrency safety net. When a write tries to insert a duplicate (aggregateId, version) pair, MongoDB rejects it with a E11000 duplicate key error. Your application can catch this and retry or return an error to the client.

In practice, you typically assign the next version number by querying the highest existing version, incrementing it, and inserting. But race conditions are possible:

async function appendEvent(aggregateId, eventType, data) {
  // Find the next version
  const lastEvent = await db.collection("events")
    .findOne({ aggregateId }, { sort: { version: -1 } });
  
  const nextVersion = (lastEvent?.version || 0) + 1;
  
  const event = {
    aggregateId,
    aggregateType: "User",
    eventType,
    version: nextVersion,
    timestamp: new Date(),
    data,
    metadata: { source: "api" }
  };
  
  try {
    await db.collection("events").insertOne(event);
    return event;
  } catch (error) {
    if (error.code === 11000) {
      // Version conflict; another process inserted this version first
      // Retry or handle gracefully
      throw new Error("Optimistic lock failed; retry");
    }
    throw error;
  }
}

This pattern is called optimistic locking. You don't lock the aggregate; instead, you let concurrent writers race and rely on the unique index to detect conflicts. If a conflict happens, the client retries. For most systems, conflicts are rare, and this approach is far simpler than pessimistic locking.

Projections build read models

Events are the source of truth, but they're not optimized for queries. You don't want to replay 10,000 events every time a user opens their dashboard. Instead, you build projections: denormalized, read-optimized views of the data.

A projection is typically another MongoDB collection that gets updated whenever events are appended. You can build projections synchronously (during the insert) or asynchronously (a separate process polls the events collection and updates projections).

Synchronous projections are simpler but risk blocking your write if the projection fails. Asynchronous projections decouple write and read paths, making the system more resilient:

// Event store write (fast, append-only)
async function handleUserUpgrade(userId) {
  const event = {
    aggregateId: userId,
    aggregateType: "User",
    eventType: "UserUpgraded",
    version: nextVersion,
    timestamp: new Date(),
    data: { plan: "premium" },
    metadata: { source: "api" },
    _projectionProcessed: false  // Flag for async processing
  };
  await db.collection("events").insertOne(event);
}

// Separate projection worker (runs periodically or continuously)
async function projectEvents() {
  const unprocessed = await db.collection("events")
    .find({ _projectionProcessed: false })
    .sort({ timestamp: 1 })
    .toArray();
  
  for (const event of unprocessed) {
    // Update the read model
    if (event.eventType === "UserCreated") {
      await db.collection("userProjection").updateOne(
        { userId: event.aggregateId },
        {
          $set: {
            email: event.data.email,
            plan: event.data.plan,
            createdAt: event.timestamp
          }
        },
        { upsert: true }
      );
    } else if (event.eventType === "UserUpgraded") {
      await db.collection("userProjection").updateOne(
        { userId: event.aggregateId },
        { $set: { plan: event.data.plan, upgradedAt: event.timestamp } }
      );
    }
    
    // Mark as processed
    await db.collection("events").updateOne(
      { _id: event._id },
      { $set: { _projectionProcessed: true } }
    );
  }
}

The userProjection collection now holds the current state in a query-friendly format. Reads go against projections; writes go against the event store. This separation gives you immutable history plus fast queries.

You can have multiple projections for different use cases. One might be a denormalized user profile for the dashboard. Another might aggregate events by plan type for analytics. Each projection is rebuilt if you detect corruption, because events are always the source of truth.

Common mistakes

Not including version in your index: If you only index by aggregateId, event replay becomes an unordered scan. Always use a compound index on (aggregateId, version).

Forgetting the unique constraint: Without a unique index on version, concurrent writes can silently create duplicates or gaps. The unique constraint is your safety net.

Synchronous projections in high-load scenarios: If projection updates block event writes, a slow projection can tank your write throughput. Decouple them asynchronously.

Storing sensitive data in events: Events are immutable. If you store a password or API key in an event and later realize it's leaked, you can't delete it from history. Use references or encrypted values instead.

Ignoring event schema evolution: As your system evolves, event schemas change. Without versioning or careful migration, old events become unreadable. Plan for this from the start by including a schema version in events or handling multiple event shapes in your apply logic.

Key takeaways

• Event sourcing stores immutable events as the source of truth, giving you a complete audit trail by design.
• Append-only collections are fast and eliminate concurrency contention at the storage layer.
• Compound indexes on (aggregateId, version) and unique constraints on that pair are essential for correctness and replay efficiency.
• Snapshots prevent event replay from becoming prohibitively slow as aggregates accumulate history.
• Projections are denormalized read models that derive from events; they can be rebuilt at any time, making them safe to update and discard.
• Decouple projection updates from event writes to prevent slow projections from blocking high-throughput writes.
• Plan for event schema evolution early; immutable storage means old events stick around forever.