Event Sourcing Framework Design and Implementation

Event sourcing is a powerful architectural pattern that captures all changes to an application state as a sequence of events, enabling full audit trails, system recovery, and real-time processing capabilities.

1. Problem Statement & Requirements

Clarifying questions

1. What is the expected scale of events? Will we be dealing with millions of events per second, or more modest throughput in the thousands?
2. What consistency model is required? Should the system support strong consistency, eventual consistency, or a hybrid approach?
3. What are the key entities and their relationships? Will we be modeling a single aggregate or multiple aggregates with cross-aggregate dependencies?
4. What is the read/write ratio? Is this primarily a write-heavy system with reads being less frequent, or is it more balanced?
5. What are the latency sensitivity requirements? Are there strict SLAs for event processing, or can we tolerate longer delays?
6. What authentication and authorization model is in place? Will the system use OAuth2, JWTs, or API keys for access control?
7. Are there any external system integrations? Will events be published to external systems or consumed by other services?

Functional requirements

• Event storage: Persist events in a durable, append-only store with strong ordering guarantees.
• Event replay capability: Allow consumers to replay events from a specific point in time or sequence.
• Aggregate reconstruction: Reconstruct the current state of an aggregate from its event stream.
• Event versioning: Support versioned events to handle schema evolution over time.
• Event filtering and projection: Allow consumers to filter and project events to derive specific views or materialized states.
• Event sourcing for multiple aggregates: Support event sourcing across multiple aggregates with independent lifecycles.

Non-functional requirements

• Latency targets:
  • p50: <10ms for event storage and retrieval
  • p95: <50ms for event storage and retrieval
  • p99: <200ms for event storage and retrieval
• Throughput:
  • 10,000 events/sec for write operations
  • 50,000 events/sec for read operations (with caching)
• Availability:
  • 99.9% uptime for event storage and replay capabilities
• Durability:
  • Events must be persisted to disk with replication across availability zones

Constraints and assumptions

• Assumption: Events are immutable once written.
• Assumption: The system will be used within a single domain or bounded context.
• Constraint: Events must be stored in a consistent, ordered sequence per aggregate.
• Constraint: Event replay capability must not impact performance of ongoing operations.
• Assumption: The system will support eventual consistency for projections and materialized views.

Out of scope

• Cross-domain event synchronization: Handling events across multiple bounded contexts or domains.
• Event schema evolution: Complex migration strategies for event schemas (e.g., field renaming, type changes).
• Event deduplication: Handling duplicate events in a distributed environment (this is handled at the consumer level).
• Event compression: Compression of event payloads for storage or transmission.
• Cross-aggregate transactional consistency: Atomicity across multiple aggregates is not a requirement.
• Real-time analytics: Real-time processing of events for analytics or reporting (this is handled via projections).
• Event encryption: Encryption of event payloads at rest or in transit (handled by infrastructure or external systems).
• Event search: Full-text search capabilities on event content.
• Event archiving: Long-term archival and retrieval of events beyond the standard retention window.

2. API Design

API Style Selection

For an Event Sourcing framework, I'll select gRPC as the primary API style for the following reasons:

1. Strong typing and schema definition: gRPC's Protocol Buffers provide clear, versioned contracts that are essential for event schemas and command structures
2. Efficient binary serialization: Better performance characteristics for high-throughput event systems
3. First-class support for streaming: Critical for real-time event publishing and subscription patterns
4. Built-in error handling: Native support for structured error responses with HTTP-like status codes
5. Service-level observability: First-class support for metrics and tracing integration

API Endpoints

Method	Path	Description	Auth
POST	/eventstore.v1.EventStore/AppendEvents	Append new events to a stream	JWT
GET	/eventstore.v1.EventStore/GetEvents	Retrieve events from a stream	JWT
POST	/eventstore.v1.EventStore/SubscribeEvents	Subscribe to real-time event streams	JWT
GET	/eventstore.v1.EventStore/GetStreamMetadata	Get metadata for a stream	JWT
POST	/eventstore.v1.EventStore/UpdateStreamMetadata	Update stream metadata	JWT
GET	/eventstore.v1.EventStore/GetAllStreams	List all streams with metadata	JWT
POST	/eventstore.v1.EventStore/QueryEvents	Query events using projections	JWT

Request and Response Schemas

Event Append Request

{
  "streamId": "user-123",
  "expectedVersion": 5,
  "events": [
    {
      "eventId": "uuid-1",
      "eventType": "UserCreated",
      "timestamp": "2023-10-15T14:30:00Z",
      "data": {
        "userId": "user-123",
        "name": "John Doe",
        "email": "john@example.com"
      },
      "metadata": {
        "userId": "user-123",
        "source": "UserService"
      }
    }
  ]
}

Event Append Response

{
  "success": true,
  "newVersion": 6,
  "errors": []
}

Event Stream Response

{
  "events": [
    {
      "eventId": "uuid-1",
      "eventType": "UserCreated",
      "timestamp": "2023-10-15T14:30:00Z",
      "data": {},
      "metadata": {}
    }
  ],
  "nextPosition": 100
}

HTTP Status Codes and Error Format

Status Code	Description
200 OK	Successful operation
400 Bad Request	Invalid request parameters or malformed events
401 Unauthorized	Missing or invalid authentication
403 Forbidden	Insufficient permissions
409 Conflict	Version mismatch (concurrent modification)
422 Unprocessable Entity	Validation errors
500 Internal Server Error	Unexpected server error

Standardized error response format:

{
  "error": {
    "code": "CONCURRENT_MODIFICATION",
    "message": "Stream version mismatch. Expected: 5, Actual: 6",
    "details": {
      "streamId": "user-123",
      "expectedVersion": 5,
      "actualVersion": 6
    }
  }
}

API Versioning Strategy

Version v1 will be the initial stable release, with versioning handled through:

• gRPC package naming: eventstore.v1
• HTTP path prefix: /eventstore/v1/
• Semantic versioning in service contracts

Pagination Design

Using cursor-based pagination for the following reasons:

1. Consistent results: Cursor-based pagination is not affected by concurrent writes to the stream
2. Stable ordering: Events are ordered by position, which is immutable once written
3. Better performance: No need to count total records or calculate offsets

Authentication and Authorization

All endpoints require JWT tokens with the following scopes:

• eventstore:read - Read access to event streams
• eventstore:write - Write access to event streams
• eventstore:subscribe - Subscribe to real-time event streams

Rate Limiting Strategy

Endpoint	Limit	Period	Enforcement
AppendEvents	1000 req/min	1 minute	Token bucket algorithm
GetEvents	5000 req/min	1 minute	Fixed window with leaky bucket
SubscribeEvents	100 req/min	1 minute	Token bucket with burst

Idempotency Design

For mutation endpoints, idempotency keys are supported via HTTP header:

X-Idempotency-Key: unique-key-for-this-request

Implementation strategy:

1. Store request hash with idempotency key in a cache (e.g., Redis)
2. If same key exists with same operation, return cached result
3. Cache TTL: 5 minutes to allow for retries within reasonable window

gRPC Service Definition (Proto)

syntax = "proto3";

package eventstore.v1;

import "google/protobuf/timestamp.proto";

service EventStore {
  // Append events to a stream
  rpc AppendEvents(AppendEventsRequest) returns (AppendEventsResponse);
  
  // Get events from a stream
  rpc GetEvents(GetEventsRequest) returns (GetEventsResponse);
  
  // Subscribe to real-time event streams
  rpc SubscribeEvents(SubscribeEventsRequest) returns (stream Event);
  
  // Get stream metadata
  rpc GetStreamMetadata(GetStreamMetadataRequest) returns (GetStreamMetadataResponse);
  
  // Update stream metadata
  rpc UpdateStreamMetadata(UpdateStreamMetadataRequest) returns (UpdateStreamMetadataResponse);
  
  // List all streams
  rpc GetAllStreams(GetAllStreamsRequest) returns (GetAllStreamsResponse);
  
  // Query events using projections
  rpc QueryEvents(QueryEventsRequest) returns (stream Event);
}

message AppendEventsRequest {
  string stream_id = 1;
  optional int64 expected_version = 2;
  repeated Event events = 3;
  map<string, string> metadata = 4;
}

message AppendEventsResponse {
  bool success = 1;
  int64 new_version = 2;
  repeated string errors = 3;
}

message GetEventsRequest {
  string stream_id = 1;
  optional int64 start_position = 2;
  optional int64 end_position = 3;
  optional int64 max_count = 4;
}

message GetEventsResponse {
  repeated Event events = 1;
  optional int64 next_position = 2;
}

message SubscribeEventsRequest {
  repeated string stream_ids = 1;
  optional int64 start_position = 2;
}

message Event {
  string event_id = 1;
  string event_type = 2;
  google.protobuf.Timestamp timestamp = 3;
  bytes data = 4;
  map<string, string> metadata = 5;
  int64 position = 6;
}

message GetStreamMetadataRequest {
  string stream_id = 1;
}

message GetStreamMetadataResponse {
  string stream_id = 1;
  int64 version = 2;
  google.protobuf.Timestamp created_at = 3;
  google.protobuf.Timestamp modified_at = 4;
  map<string, string> metadata = 5;
}

message UpdateStreamMetadataRequest {
  string stream_id = 1;
  map<string, string> metadata = 2;
}

message UpdateStreamMetadataResponse {
  bool success = 1;
}

message GetAllStreamsRequest {
  optional int64 max_count = 1;
  optional string cursor = 2;
}

message GetAllStreamsResponse {
  repeated StreamMetadata streams = 1;
  optional string next_cursor = 2;
}

message QueryEventsRequest {
  string query = 1;
  optional int64 max_count = 2;
}

message StreamMetadata {
  string stream_id = 1;
  int64 version = 2;
  google.protobuf.Timestamp created_at = 3;
  google.protobuf.Timestamp modified_at = 4;
  map<string, string> metadata = 5;
}

3. Data Model Design

Entity-Relationship diagram (Mermaid)

erDiagram
    EVENT_STORE {
        string id PK
        string aggregate_id
        string event_type
        string payload
        integer version
        timestamp created_at
        string created_by
    }

    AGGREGATE {
        string id PK
        string type
        integer version
        timestamp last_updated
    }

    EVENT_STORE ||--o{ AGGREGATE : "contains"

Table / Collection schema definitions with field types, constraints, indices

Event Store Table

CREATE TABLE event_store (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_id VARCHAR(255) NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    version INTEGER NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    created_by VARCHAR(255),
    
    -- Constraints
    CONSTRAINT chk_version_positive CHECK (version > 0),
    CONSTRAINT chk_event_type_not_empty CHECK (LENGTH(event_type) > 0),
    
    -- Indexes
    INDEX idx_aggregate_id (aggregate_id),
    INDEX idx_event_type (event_type),
    INDEX idx_created_at (created_at),
    INDEX idx_aggregate_version (aggregate_id, version)
);

Aggregate State Table

CREATE TABLE aggregate_state (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_id VARCHAR(255) NOT NULL UNIQUE,
    aggregate_type VARCHAR(100) NOT NULL,
    current_version INTEGER NOT NULL DEFAULT 0,
    state JSONB,
    last_updated TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    
    -- Constraints
    CONSTRAINT chk_current_version_non_negative CHECK (current_version >= 0),
    
    -- Indexes
    INDEX idx_aggregate_type (aggregate_type),
    INDEX idx_last_updated (last_updated)
);

Primary key strategy (UUID vs. auto-increment vs. ULID — with justification)

UUID Strategy: Using UUIDs as primary keys for both event_store and aggregate_state tables.

Justification:

• Global uniqueness: UUIDs ensure uniqueness across distributed systems without coordination
• Scalability: No need for centralized ID generation, enabling horizontal scaling
• No hotspots: Unlike auto-increment, UUIDs don't create write hotspots in distributed databases
• Cross-system compatibility: UUIDs can be generated client-side, useful for event sourcing in distributed systems

Index design (composite, partial, covering indices — with justification per index)

1. idx_aggregate_id: CREATE INDEX idx_aggregate_id ON event_store(aggregate_id)

   • Justification: Used for querying all events of a specific aggregate, which is the most frequent operation in event sourcing

2. idx_event_type: CREATE INDEX idx_event_type ON event_store(event_type)

   • Justification: Enables querying all events of a specific type across aggregates (e.g., "all OrderCreated events")

3. idx_created_at: CREATE INDEX idx_created_at ON event_store(created_at)

   • Justification: For time-based queries and audit trails

4. idx_aggregate_version: CREATE INDEX idx_aggregate_version ON event_store(aggregate_id, version)

   • Justification: Composite index for efficient retrieval of events in order by aggregate and version, supporting replay operations

5. idx_aggregate_type: CREATE INDEX idx_aggregate_type ON aggregate_state(aggregate_type)

   • Justification: For querying all aggregates of a specific type for maintenance or analytics

Data partitioning / sharding strategy

Sharding by Aggregate ID Hash:

• Partition key: hash(aggregate_id) % num_partitions
• Justification: Provides even distribution of writes and reads across shards
• Shard size: 100GB per shard (configurable based on storage capacity)
• Shard boundary: Dynamically managed via database sharding tools (e.g., Citus, CockroachDB)

Soft delete vs. hard delete decision with justification

Soft Delete Strategy: Events are not deleted but marked as "soft-deleted" or archived.

Justification:

• Audit and compliance: Regulatory requirements often mandate immutable event logs
• Data integrity: Prevents data loss during replay operations or debugging
• Replay capability: Original events must remain accessible for event replay and state reconstruction

Audit trail design (who, what, when)

Audit Fields in Event Store:

• created_by: User or system identifier that generated the event
• created_at: Timestamp of when event was persisted (UTC)
• version: Sequence number within the aggregate's event stream

Data archival strategy and retention policy

Archival Strategy:

• Hot/Cold Storage: Recent events (last 30 days) in hot storage, older events in cold storage
• Retention Policy:
  • 30 days: Hot storage (fast read/write)
  • 1 year: Cold storage (reduced cost, slower access)
  • 5+ years: Long-term archival (e.g., tape storage or cloud glacier)

Archive Trigger: Events older than 30 days are marked as archived and moved to cold storage using a scheduled job or database partitioning strategy.

4. Class Design & Service Layer

UML Class Diagram

classDiagram
    class EventStore {
        +saveEvents(List~Event~ events)
        +getEvents(String aggregateId, int fromVersion)
        +getEventsByType(String eventType, Instant after, Instant before)
    }
    
    class Event {
        +id: UUID
        +aggregateId: String
        +version: int
        +type: String
        +timestamp: Instant
        +data: Map~String, Object~
    }
    
    class Aggregate {
        +id: String
        +version: int
        +applyEvent(Event event)
        +replayEvents(List~Event~ events)
    }
    
    class EventSourcedAggregate {
        +id: String
        +version: int
        +handleCommand(Command command)
        +applyEvent(Event event)
        +replayEvents(List~Event~ events)
    }
    
    class Command {
        +id: UUID
        +aggregateId: String
        +type: String
        +data: Map~String, Object~
    }
    
    class CommandHandler {
        +handle(Command command)
        +validate(Command command)
    }
    
    class EventPublisher {
        +publish(Event event)
    }
    
    class SnapshotStore {
        +saveSnapshot(Snapshot snapshot)
        +getSnapshot(String aggregateId, Instant before)
    }
    
    class Snapshot {
        +aggregateId: String
        +version: int
        +timestamp: Instant
        +data: Map~String, Object~
    }
    
    class EventSourcingService {
        +processCommand(Command command)
        +getAggregateState(String aggregateId)
        +replayAggregate(String aggregateId)
    }
    
    EventStore --> Event : stores
    EventStore --> SnapshotStore : uses for snapshots
    
    Aggregate <|-- EventSourcedAggregate : implements
    EventSourcedAggregate --> Event : applies
    EventSourcedAggregate --> CommandHandler : delegates
    
    CommandHandler --> Command : handles
    CommandHandler --> Event : produces
    
    EventPublisher --> Event : publishes
    
    EventSourcingService --> EventStore : persists
    EventSourcingService --> SnapshotStore : manages snapshots
    EventSourcingService --> EventPublisher : emits events
    
    class EventSourcedAggregate {
        +id: String
        +version: int
        +handleCommand(Command command)
        +applyEvent(Event event)
        +replayEvents(List~Event~ events)
    }
    
    class CommandHandler {
        +handle(Command command)
        +validate(Command command)
    }

State Machine Diagram for Aggregate

stateDiagram-v2
    [*] --> Created
    Created --> Active : handleCreate()
    Active --> Deleted : handleDelete()
    Deleted --> [*] : handleRestore()
    
    state "Active" {
        [*] --> Processing
        Processing --> Validated : validate()
        Validated --> Committed : commit()
        Committed --> [*] : applyEvent()
    }
    
    state "Deleted" {
        [*] --> SoftDeleted
        SoftDeleted --> Restored : handleRestore()
        Restored --> Active : restoreState()
    }

Service Class Structure and Responsibilities

The Event Sourcing framework consists of several key service classes:

1. EventSourcingService: The main orchestrator that handles command processing, aggregate state management, and event persistence.
2. EventStoreService: Manages the storage and retrieval of events from the event store.
3. SnapshotService: Handles snapshot creation and retrieval for performance optimization.
4. EventPublisherService: Publishes events to message brokers or other consumers.

Interface / Abstract Class Design

public interface EventStore {
    void saveEvents(List<Event> events);
    List<Event> getEvents(String aggregateId, int fromVersion);
    List<Event> getEventsByType(String eventType, Instant after, Instant before);
}

public interface Aggregate {
    String getId();
    int getVersion();
    void applyEvent(Event event);
    void replayEvents(List<Event> events);
}

public interface CommandHandler<T extends Aggregate> {
    List<Event> handle(Command command, T aggregate);
    void validate(Command command);
}

public abstract class EventSourcedAggregate implements Aggregate {
    protected String id;
    protected int version;
    
    public abstract List<Event> applyEvent(Event event);
    public abstract void replayEvents(List<Event> events);
    
    @Override
    public String getId() {
        return id;
    }
    
    @Override
    public int getVersion() {
        return version;
    }
}

Business Logic Encapsulation Patterns

The framework uses the Command Pattern to encapsulate business logic operations:

public class OrderCommandHandler implements CommandHandler<OrderAggregate> {
    @Override
    public List<Event> handle(Command command, OrderAggregate aggregate) {
        switch (command.getType()) {
            case "CREATE_ORDER":
                return handleCreateOrder(command, aggregate);
            case "UPDATE_ORDER":
                return handleUpdateOrder(command, aggregate);
            case "CANCEL_ORDER":
                return handleCancelOrder(command, aggregate);
            default:
                throw new IllegalArgumentException("Unknown command type: " + command.getType());
        }
    }
    
    private List<Event> handleCreateOrder(Command command, OrderAggregate aggregate) {
        validate(command);
        
        // Business logic for creating order
        List<Event> events = new ArrayList<>();
        events.add(new Event(
            UUID.randomUUID(),
            aggregate.getId(),
            aggregate.getVersion() + 1,
            "ORDER_CREATED",
            Instant.now(),
            command.getData()
        ));
        
        return events;
    }
    
    @Override
    public void validate(Command command) {
        if (command.getData() == null || command.getData().isEmpty()) {
            throw new IllegalArgumentException("Command data cannot be empty");
        }
    }
}

Critical-Path Method Implementation

@Service
public class EventSourcingService {
    private final EventStore eventStore;
    private final SnapshotStore snapshotStore;
    private final EventPublisher eventPublisher;
    private final Map<String, CommandHandler> commandHandlers;
    
    public List<Event> processCommand(Command command) {
        // 1. Validate command
        CommandHandler handler = commandHandlers.get(command.getType());
        if (handler == null) {
            throw new IllegalArgumentException("No handler found for command: " + command.getType());
        }
        
        // 2. Load aggregate state
        String aggregateId = command.getAggregateId();
        List<Event> events = new ArrayList<>();
        
        // Use optimistic locking with version check
        int expectedVersion = command.getExpectedVersion();
        List<Event> existingEvents = eventStore.getEvents(aggregateId, 0);
        
        // Validate version consistency
        if (expectedVersion != -1 && existingEvents.size() != expectedVersion) {
            throw new ConcurrencyException("Version mismatch: expected " + 
                expectedVersion + ", got " + existingEvents.size());
        }
        
        // 3. Create new aggregate instance
        EventSourcedAggregate aggregate = createAggregate(aggregateId, existingEvents);
        
        // 4. Handle command and generate events
        List<Event> newEvents = handler.handle(command, aggregate);
        
        // 5. Validate and apply events
        validateEvents(newEvents);
        
        // 6. Persist events in transaction
        try (Transaction tx = TransactionManager.begin()) {
            eventStore.saveEvents(newEvents);
            
            // 7. Publish events
            for (Event event : newEvents) {
                eventPublisher.publish(event);
            }
            
            tx.commit();
        } catch (Exception e) {
            throw new EventStoreException("Failed to persist events", e);
        }
        
        return newEvents;
    }
    
    private EventSourcedAggregate createAggregate(String aggregateId, List<Event> events) {
        // Factory pattern to create appropriate aggregate type
        return new OrderAggregate(aggregateId, events);
    }
    
    private void validateEvents(List<Event> events) {
        // Validate event sequence and business rules
        for (Event event : events) {
            if (event.getAggregateId() == null || event.getType() == null) {
                throw new IllegalArgumentException("Event must have aggregateId and type");
            }
        }
    }
}

Transaction Boundary Design

The transaction boundary encompasses:

1. Reading current aggregate state from event store
2. Validating command and applying business logic
3. Persisting new events to event store
4. Publishing events to message broker

All operations within this boundary are atomic using database transactions.

Service Interface Contracts

public interface EventSourcingService {
    List<Event> processCommand(Command command);
    AggregateState getAggregateState(String aggregateId);
    void replayAggregate(String aggregateId);
    Snapshot getSnapshot(String aggregateId, Instant before);
}

Saga / Orchestration Pattern

For distributed operations, the framework supports saga orchestration:

public class OrderSaga {
    private final List<Step> steps = new ArrayList<>();
    
    public void execute(Command command) {
        try {
            // Execute each step in order
            for (Step step : steps) {
                step.execute();
            }
        } catch (Exception e) {
            // Compensate for failed steps
            compensate();
            throw new SagaExecutionException("Saga failed", e);
        }
    }
    
    private void compensate() {
        // Reverse the steps in reverse order
        for (int i = steps.size() - 1; i >= 0; i--) {
            steps.get(i).compensate();
        }
    }
}

Domain Object / Value Object Design

public class Event {
    private final UUID id;
    private final String aggregateId;
    private final int version;
    private final String type;
    private final Instant timestamp;
    private final Map<String, Object> data;
    
    public Event(UUID id, String aggregateId, int version, String type, 
                 Instant timestamp, Map<String, Object> data) {
        this.id = id;
        this.aggregateId = aggregateId;
        this.version = version;
        this.type = type;
        this.timestamp = timestamp;
        this.data = data != null ? new HashMap<>(data) : Collections.emptyMap();
    }
    
    // Getters and equals/hashCode
}

public class Command {
    private final UUID id;
    private final String aggregateId;
    private final String type;
    private final Map<String, Object> data;
    private final int expectedVersion;
    
    public Command(UUID id, String aggregateId, String type, 
                   Map<String, Object> data, int expectedVersion) {
        this.id = id;
        this.aggregateId = aggregateId;
        this.type = type;
        this.data = data != null ? new HashMap<>(data) : Collections.emptyMap();
        this.expectedVersion = expectedVersion;
    }
    
    // Getters and equals/hashCode
}

public class AggregateState {
    private final String aggregateId;
    private final int version;
    private final Map<String, Object> state;
    
    public AggregateState(String aggregateId, int version, Map<String, Object> state) {
        this.aggregateId = aggregateId;
        this.version = version;
        this.state = new HashMap<>(state);
    }
    
    // Getters and equals/hashCode
}

5. Concurrency & Thread Safety

Concurrency Model

The Event Sourcing framework will utilize a multi-threaded, lock-free approach for handling concurrent event writes and reads. The core concurrency model is built around:

1. Thread-safe event append operations using atomic operations and lock-free data structures
2. Immutable event objects to prevent race conditions during read operations
3. Async processing for event handling and projection updates
4. Read-only transactional consistency through snapshot-based reads

The system will use Java's java.util.concurrent package for thread safety, with a focus on lock-free programming patterns where possible to minimize contention.

Race Condition Prevention Strategy

Race conditions in event sourcing are primarily mitigated through:

1. Event versioning - Each event carries a version number that prevents concurrent writes from overwriting each other
2. Optimistic concurrency control - Events are written only if the expected version matches current state
3. Immutable event objects - Once created, events cannot be modified, eliminating shared mutable state
4. Idempotency keys - For replayable operations, ensuring duplicate events don't cause inconsistencies

Optimistic vs. Pessimistic Locking Selection

Optimistic locking is selected for the following reasons:

• High throughput: Reduces lock contention in high-concurrency scenarios
• Event sourcing nature: Events are append-only and immutable, making pessimistic locking unnecessary
• Better scalability: Lock-free design allows horizontal scaling of event processing
• Event versioning provides safety: Version checks prevent overwrites during concurrent writes

Implementation uses CAS (Compare-And-Swap) operations on version counters to ensure atomicity during event writes.

Deadlock Prevention Strategy

Deadlock prevention in the event sourcing framework includes:

1. Fixed ordering of operations: All event processing follows a consistent order (e.g., version check → append → notify)
2. Timeout-based operation limits: Operations timeout after 5 seconds to prevent indefinite blocking
3. No nested locking: Events are processed in isolation without acquiring multiple locks simultaneously
4. Thread pool size limits: Controlled parallelism to prevent resource exhaustion

Idempotency Implementation for Concurrent Duplicate Requests

Idempotency is implemented at the event level:

public class EventIdempotencyService {
    private final Set<String> processedEventIds = new ConcurrentHashSet<>();
    private final ScheduledExecutorService cleanupScheduler = 
        Executors.newScheduledThreadPool(1);

    public boolean isProcessed(String eventId) {
        return processedEventIds.contains(eventId);
    }

    public boolean markAsProcessed(String eventId) {
        return processedEventIds.add(eventId);
    }

    public void cleanupOldEvents() {
        // Cleanup processed event IDs older than 30 days
        long cutoff = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(30);
        processedEventIds.removeIf(id -> {
            // Assuming event IDs contain timestamp or can be parsed for age
            return extractTimestamp(id) < cutoff;
        });
    }
}

The system uses a distributed cache (Redis) for idempotency tracking in multi-instance deployments, with TTL-based cleanup to prevent memory leaks.

Thread Safety Patterns

1. Immutable Event Objects: Events are designed as immutable value objects to prevent concurrent modification
2. Thread-safe collections: ConcurrentHashMap and CopyOnWriteArrayList for managing event streams
3. Atomic references: For managing current event stream version pointers
4. Synchronized blocks only for critical sections - Only when necessary for maintaining consistency across multiple operations

Concurrency Control in Event Processing

public class ConcurrentEventProcessor {
    private final AtomicLong currentVersion = new AtomicLong(0);
    private final Map<String, EventStream> eventStreams = new ConcurrentHashMap<>();
    
    public boolean appendEvent(String streamId, Event event) {
        // Use optimistic concurrency control
        long expectedVersion = event.getExpectedVersion();
        long currentVersionValue = currentVersion.get();
        
        if (expectedVersion != currentVersionValue) {
            throw new ConcurrentModificationException(
                "Event version mismatch. Expected: " + expectedVersion 
                + ", Current: " + currentVersionValue);
        }
        
        // Atomic append operation
        return eventStreams.computeIfAbsent(streamId, k -> new EventStream())
                          .appendEvent(event);
    }
}

This approach ensures that concurrent event writes are handled safely while maintaining high throughput and avoiding the overhead of traditional locking mechanisms.

6. Sequence Diagrams for Key Flows

sequenceDiagram
    participant Client
    participant EventStore
    participant EventHandler
    participant DomainService

    Client->>EventStore: POST /events
    EventStore->>DomainService: Validate and process event
    DomainService->>EventStore: Store event in event stream
    EventStore->>EventHandler: Publish event to handler queue
    EventHandler->>EventHandler: Process event and update read model
    EventHandler->>EventStore: Acknowledge successful processing

sequenceDiagram
    participant Client
    participant EventStore
    participant ErrorHandler
    participant DomainService

    Client->>EventStore: POST /events
    EventStore->>DomainService: Validate and process event
    DomainService->>EventStore: Store event in event stream
    EventStore->>ErrorHandler: Publish event to error queue on failure
    ErrorHandler->>ErrorHandler: Retry event processing with exponential backoff
    ErrorHandler->>EventStore: Acknowledge error handling completion

sequenceDiagram
    participant Client
    participant EventStore
    participant EventPublisher
    participant ExternalService

    Client->>EventStore: POST /events
    EventStore->>EventStore: Store event in event stream
    EventStore->>EventPublisher: Publish to message broker
    EventPublisher->>ExternalService: Forward event to external system
    ExternalService->>EventPublisher: Acknowledge receipt
    EventPublisher->>EventStore: Update event status to published

7. Caching Design

Cache Strategy Selection

For event sourcing, we'll use cache-aside pattern with a write-through strategy for the event store. This approach provides optimal balance between consistency and performance for read-heavy workloads typical in event-sourced systems.

Cache Key Design

Cache keys will follow a namespace strategy:

{cache_namespace}:{entity_type}:{entity_id}:{version?}

Example key structure:

eventstore:order:12345:latest
eventstore:order:12345:100
eventstore:user:67890:latest

TTL Strategy

Data Type	TTL	Reason
Event Streams	30 days	Recent events frequently accessed, older ones archived
Aggregate State	7 days	Recent state needed for queries, older states can be recomputed
Event Metadata	1 day	Metadata changes less frequently
Cache Control	5 minutes	For configuration and system state updates

Eviction Policy

We'll implement LRU (Least Recently Used) eviction policy with a TTL fallback:

• Primary: LRU-based eviction when cache reaches capacity
• Secondary: TTL-based cleanup for stale entries

Cache Stampede Prevention

To prevent cache stampedes:

1. Probabilistic Early Expiry: Set TTL to 75% of configured value (e.g., 22.5 days instead of 30)
2. Distributed Locking: Use Redis distributed locks for cache warming operations
3. Cache Warm-up Strategy: Pre-warm hot aggregates during low-traffic periods

Distributed Cache Architecture

Using Redis Cluster with:

• 3 master nodes for write availability
• 3 replica nodes per shard for read availability
• Redis Sentinel for automatic failover
• Memory policy set to allkeys-lru with 2GB max memory per node

Cache Invalidation Strategy

Event-driven invalidation with:

1. Immediate invalidation: When events are published to the event store, invalidate related cache entries
2. Batch invalidation: For bulk operations, batch invalidation using Redis pipeline
3. TTL-based cleanup: Fallback to TTL expiration for stale entries

Cache Implementation Example (Java)

@Service
public class EventStoreCache {
    private final RedisTemplate<String, Object> redisTemplate;
    private final ObjectMapper objectMapper;
    
    public EventStoreCache(RedisTemplate<String, Object> redisTemplate, 
                          ObjectMapper objectMapper) {
        this.redisTemplate = redisTemplate;
        this.objectMapper = objectMapper;
    }
    
    public List<Event> getEvents(String aggregateId, String eventType) {
        String cacheKey = buildCacheKey(aggregateId, eventType);
        String cachedValue = (String) redisTemplate.opsForValue().get(cacheKey);
        
        if (cachedValue != null) {
            try {
                return objectMapper.readValue(cachedValue, 
                    new TypeReference<List<Event>>() {});
            } catch (Exception e) {
                // Log and continue with database fetch
                log.warn("Failed to deserialize cached events for {}", aggregateId);
            }
        }
        
        // Cache miss - fetch from database
        List<Event> events = eventRepository.findByAggregateId(aggregateId);
        cacheEvents(cacheKey, events);
        
        return events;
    }
    
    public void invalidateAggregate(String aggregateId) {
        // Invalidate all keys for this aggregate
        Set<String> keys = redisTemplate.keys("eventstore:" + aggregateId + ":*");
        if (keys != null && !keys.isEmpty()) {
            redisTemplate.delete(keys);
        }
    }
    
    private String buildCacheKey(String aggregateId, String eventType) {
        return "eventstore:" + aggregateId + ":" + eventType;
    }
    
    private void cacheEvents(String key, List<Event> events) {
        try {
            String json = objectMapper.writeValueAsString(events);
            redisTemplate.opsForValue().set(key, json, 30, TimeUnit.DAYS);
        } catch (Exception e) {
            log.error("Failed to cache events for key {}", key, e);
        }
    }
}

8. Repository / Data Access Layer

Repository Pattern Implementation

The repository pattern is implemented to abstract data access logic and provide a clean separation between domain objects and persistence concerns. Each aggregate root has its own repository interface with standardized CRUD operations.

public interface EventStoreRepository {
    void saveEvents(List<Event> events);
    List<Event> getEventsForAggregate(String aggregateId, long fromVersion, long toVersion);
    long getLatestVersion(String aggregateId);
    void saveSnapshot(Snapshot snapshot);
    Optional<Snapshot> getSnapshot(String aggregateId, long version);
}

public interface EventRepository {
    void save(Event event);
    List<Event> findByAggregateId(String aggregateId, long fromVersion, long toVersion);
    Optional<Event> findById(String eventId);
    List<Event> findAll();
}

ORM vs Raw Query Decisions

For event sourcing, we use raw SQL queries instead of ORM for better control over query execution and performance optimization. Events are stored in a single table with versioned fields to support efficient querying.

@Repository
public class EventStoreRepositoryImpl implements EventStoreRepository {
    
    private final JdbcTemplate jdbcTemplate;
    
    @Override
    public void saveEvents(List<Event> events) {
        String sql = "INSERT INTO event_store (id, aggregate_id, version, type, data, timestamp) VALUES (?, ?, ?, ?, ?, ?)";
        
        jdbcTemplate.batchUpdate(sql, events, events.size(), (ps, event) -> {
            ps.setString(1, event.getId());
            ps.setString(2, event.getAggregateId());
            ps.setLong(3, event.getVersion());
            ps.setString(4, event.getType());
            ps.setString(5, event.getData());
            ps.setTimestamp(6, Timestamp.valueOf(event.getTimestamp()));
        });
    }
    
    @Override
    public List<Event> getEventsForAggregate(String aggregateId, long fromVersion, long toVersion) {
        String sql = "SELECT * FROM event_store WHERE aggregate_id = ? AND version BETWEEN ? AND ? ORDER BY version";
        
        return jdbcTemplate.query(sql, new Object[]{aggregateId, fromVersion, toVersion}, 
            (rs, rowNum) -> Event.builder()
                .id(rs.getString("id"))
                .aggregateId(rs.getString("aggregate_id"))
                .version(rs.getLong("version"))
                .type(rs.getString("type"))
                .data(rs.getString("data"))
                .timestamp(rs.getTimestamp("timestamp").toLocalDateTime())
                .build());
    }
}

Read vs Write Separation

We implement CQRS pattern for the event store to separate read and write operations:

public interface EventReadRepository {
    List<Event> getEventsByAggregateId(String aggregateId);
    Optional<Event> getLatestEventForAggregate(String aggregateId);
}

public interface EventWriteRepository {
    void saveEvents(List<Event> events);
    void saveSnapshot(Snapshot snapshot);
}

Connection Pool Configuration

Connection pool is configured with appropriate size and timeout settings:

@Configuration
public class DatabaseConfig {
    
    @Bean
    public DataSource dataSource() {
        HikariDataSource dataSource = new HikariDataSource();
        dataSource.setJdbcUrl("jdbc:postgresql://localhost:5432/eventstore");
        dataSource.setUsername("user");
        dataSource.setPassword("password");
        dataSource.setMaximumPoolSize(20);
        dataSource.setConnectionTimeout(30000);
        dataSource.setIdleTimeout(600000);
        dataSource.setMaxLifetime(1800000);
        dataSource.setLeakDetectionThreshold(60000);
        return dataSource;
    }
}

N+1 Query Prevention Strategy

To prevent N+1 queries, we implement eager loading and batch operations:

@Service
public class EventService {
    
    @Autowired
    private EventReadRepository eventReadRepository;
    
    public List<AggregateEventView> getAggregateViews(List<String> aggregateIds) {
        // Batch query to fetch all events for multiple aggregates
        Map<String, List<Event>> eventsByAggregate = eventReadRepository.getEventsForAggregates(aggregateIds);
        
        return aggregateIds.stream()
            .map(aggregateId -> {
                List<Event> events = eventsByAggregate.getOrDefault(aggregateId, Collections.emptyList());
                return new AggregateEventView(aggregateId, events);
            })
            .collect(Collectors.toList());
    }
}

Batch Operation Design

Batch operations are implemented to handle bulk event persistence efficiently:

@Repository
public class BatchEventRepository {
    
    private static final int BATCH_SIZE = 1000;
    
    public void saveEventsInBatches(List<Event> events) {
        for (int i = 0; i < events.size(); i += BATCH_SIZE) {
            int end = Math.min(i + BATCH_SIZE, events.size());
            List<Event> batch = events.subList(i, end);
            
            saveBatch(batch);
        }
    }
    
    private void saveBatch(List<Event> batch) {
        String sql = "INSERT INTO event_store (id, aggregate_id, version, type, data, timestamp) VALUES (?, ?, ?, ?, ?, ?)";
        
        jdbcTemplate.batchUpdate(sql, batch, batch.size(), (ps, event) -> {
            ps.setString(1, event.getId());
            ps.setString(2, event.getAggregateId());
            ps.setLong(3, event.getVersion());
            ps.setString(4, event.getType());
            ps.setString(5, event.getData());
            ps.setTimestamp(6, Timestamp.valueOf(event.getTimestamp()));
        });
    }
}

9. Asynchronous Processing & Messaging

Event-driven boundaries in this feature — what triggers async vs. what stays synchronous, and why

Event sourcing fundamentally embraces asynchronous processing as a core architectural principle. In this framework:

• Synchronous operations: Write operations (e.g., saveEvent(), applyEvent()) that must return immediately to the caller for immediate consistency and user feedback.
• Asynchronous operations: Publishing events to a message broker (e.g., Kafka, RabbitMQ) or internal event bus for downstream consumers to process. This includes:
  • Event persistence to the event store (asynchronous write to DB)
  • Event-driven aggregates (e.g., updating read models, triggering notifications)
  • Audit logging and analytics processing
  • Integration with external systems (e.g., sending order confirmation emails, updating inventory)

The key decision is to keep the core domain logic synchronous but decouple side effects asynchronously. This provides better scalability, resilience, and loose coupling between services.

Message schema design with a sample payload (Avro / Protobuf / JSON Schema — with versioning strategy)

We'll use JSON Schema for simplicity and flexibility, but in production would likely use Avro or Protobuf for better performance and type safety.

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "EventSourcingEvent",
  "version": "1.0.0",
  "type": "object",
  "properties": {
    "eventId": {
      "type": "string",
      "description": "Unique identifier for the event"
    },
    "aggregateId": {
      "type": "string",
      "description": "Identifier of the aggregate this event belongs to"
    },
    "eventType": {
      "type": "string",
      "description": "Type of the event (e.g., 'OrderCreated', 'PaymentProcessed')"
    },
    "timestamp": {
      "type": "string",
      "format": "date-time",
      "description": "ISO 8601 timestamp when the event occurred"
    },
    "payload": {
      "type": "object",
      "description": "Event-specific data"
    },
    "metadata": {
      "type": "object",
      "description": "Additional metadata (e.g., user ID, correlation ID)"
    }
  },
  "required": ["eventId", "aggregateId", "eventType", "timestamp", "payload"]
}

Versioning strategy:

• Use semantic versioning (e.g., v1.0.0, v2.0.0) in the eventType field or as part of a separate versioned event name.
• Schema evolution handled via schema registry (e.g., Confluent Schema Registry) to support backward/forward compatibility.

Dead letter queue strategy and poison message handling

Poison messages are events that repeatedly fail processing. Strategy:

• DLQ: Messages that fail more than N times (e.g., 3) are moved to a dead letter queue.
• Retry logic: Implement exponential backoff with jitter (e.g., 1s, 2s, 4s) before moving to DLQ.
• Manual intervention: DLQ messages are logged with full context and can be replayed manually or fixed by engineers.

At-least-once vs. exactly-once delivery — decision with justification for this feature

At-least-once delivery is chosen because:

• Event sourcing inherently requires that all events are processed to maintain consistency.
• The framework should allow consumers to idempotently handle duplicate events (e.g., via event IDs).
• In systems where order matters, at-least-once provides better correctness guarantees than exactly-once (which can introduce complexity and performance overhead).
• For most downstream consumers (e.g., analytics, notifications), at-least-once is acceptable and more scalable.

Retry policy for consumers (max attempts, backoff formula, non-retryable error classification)

public class RetryPolicy {
    private static final int MAX_RETRIES = 3;
    private static final long INITIAL_DELAY_MS = 1000;
    private static final double BACKOFF_MULTIPLIER = 2.0;
    private static final double JITTER_RANGE = 0.1;

    public static boolean shouldRetry(Exception ex) {
        // Non-retryable errors (e.g., validation, business logic errors)
        if (ex instanceof ValidationException || ex instanceof BusinessLogicException) {
            return false;
        }
        // Retryable errors (e.g., network, DB timeouts)
        return ex instanceof IOException || ex instanceof TimeoutException;
    }

    public static long getDelay(int attempt) {
        double delay = INITIAL_DELAY_MS * Math.pow(BACKOFF_MULTIPLIER, attempt);
        // Add jitter (±10%)
        delay *= (1 - JITTER_RANGE + Math.random() * 2 * JITTER_RANGE);
        return (long) delay;
    }
}

Outbox pattern for reliable event publishing — include or explicitly justify omission

Outbox pattern is included to ensure that events are published reliably and in order.

• Justification: In event sourcing, an event must be persisted to the event store and then published to a message broker. If we publish directly after persisting, we risk losing events if the broker is unavailable.
• Implementation: Events are written to an event_outbox table within the same transaction as the event store. A background job processes this outbox and publishes to the message broker.

Sequence diagram for all async flows (Mermaid)

sequenceDiagram
    participant Client
    participant Service
    participant EventStore
    participant Outbox
    participant MessageBroker

    Client->>Service: SubmitEventRequest
    Service->>EventStore: SaveEventToStore
    Service->>Outbox: InsertIntoOutbox
    Outbox-->>MessageBroker: AsyncPublishToBroker
    MessageBroker->>Consumer: EventDelivery

10. Error Handling Design

Error Classification Hierarchy

Errors in an Event Sourcing framework are categorized into three primary levels:

1. Validation Errors - Input data issues (e.g., missing fields, invalid formats)
2. Domain Errors - Business rule violations (e.g., insufficient funds, invalid state transitions)
3. Infrastructure Errors - System-level failures (e.g., database connection issues, network timeouts)

public class EventSourcingError extends RuntimeException {
    private final ErrorType type;
    private final String errorCode;
    
    public EventSourcingError(ErrorType type, String errorCode, String message) {
        super(message);
        this.type = type;
        this.errorCode = errorCode;
    }
    
    public enum ErrorType {
        VALIDATION, DOMAIN, INFRASTRUCTURE
    }
}

public class ValidationException extends EventSourcingError {
    public ValidationException(String errorCode, String message) {
        super(ErrorType.VALIDATION, errorCode, message);
    }
}

public class DomainException extends EventSourcingError {
    public DomainException(String errorCode, String message) {
        super(ErrorType.DOMAIN, errorCode, message);
    }
}

public class InfrastructureException extends EventSourcingError {
    public InfrastructureException(String errorCode, String message) {
        super(ErrorType.INFRASTRUCTURE, errorCode, message);
    }
}

Error Propagation Strategy

Errors are handled at the service layer and propagated upward to the API layer for proper HTTP status code mapping. For domain-level errors, we maintain specific error codes to support client-side error handling and user feedback.

Retry Strategy with Exponential Backoff

For transient infrastructure errors, we implement a retry mechanism with exponential backoff and jitter to prevent thundering herd issues:

public class RetryableOperation {
    private static final int MAX_RETRIES = 3;
    private static final long INITIAL_DELAY_MS = 1000;
    private static final double JITTER_FACTOR = 0.2;
    
    public <T> T executeWithRetry(Supplier<T> operation) {
        int attempts = 0;
        while (true) {
            try {
                return operation.get();
            } catch (InfrastructureException e) {
                if (attempts >= MAX_RETRIES) {
                    throw e;
                }
                
                long delay = calculateDelay(attempts);
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new InfrastructureException("Retry interrupted", "RETRY_INTERRUPTED");
                }
                
                attempts++;
            }
        }
    }
    
    private long calculateDelay(int attempt) {
        long baseDelay = (long) (INITIAL_DELAY_MS * Math.pow(2, attempt));
        long jitter = (long) (baseDelay * JITTER_FACTOR * (Math.random() - 0.5) * 2);
        return baseDelay + jitter;
    }
}

Circuit Breaker Design

For external dependencies or frequently failing operations, we implement a circuit breaker pattern to prevent cascading failures:

public class CircuitBreaker {
    private final int failureThreshold;
    private final long timeoutMs;
    private final AtomicInteger failureCount = new AtomicInteger(0);
    private final AtomicReference<State> state = new AtomicReference<>(State.CLOSED);
    private final AtomicLong lastFailureTime = new AtomicLong(0);
    
    public enum State {
        CLOSED, OPEN, HALF_OPEN
    }
    
    public <T> T execute(Supplier<T> operation) {
        switch (state.get()) {
            case CLOSED:
                return executeClosed(operation);
            case OPEN:
                return executeOpen(operation);
            case HALF_OPEN:
                return executeHalfOpen(operation);
            default:
                throw new IllegalStateException("Unknown circuit breaker state");
        }
    }
    
    private <T> T executeClosed(Supplier<T> operation) {
        try {
            T result = operation.get();
            failureCount.set(0);
            return result;
        } catch (Exception e) {
            int failures = failureCount.incrementAndGet();
            if (failures >= failureThreshold) {
                state.set(State.OPEN);
                lastFailureTime.set(System.currentTimeMillis());
            }
            throw e;
        }
    }
    
    private <T> T executeOpen(Supplier<T> operation) {
        if (System.currentTimeMillis() - lastFailureTime.get() > timeoutMs) {
            state.set(State.HALF_OPEN);
            return executeHalfOpen(operation);
        }
        throw new InfrastructureException("Circuit breaker is OPEN", "CIRCUIT_OPEN");
    }
    
    private <T> T executeHalfOpen(Supplier<T> operation) {
        try {
            T result = operation.get();
            state.set(State.CLOSED);
            failureCount.set(0);
            return result;
        } catch (Exception e) {
            state.set(State.OPEN);
            lastFailureTime.set(System.currentTimeMillis());
            throw e;
        }
    }
}

Fallback Behavior Design

For operations that can provide degraded functionality, we define fallbacks:

public class EventSourcingFallback {
    public static final String FALLBACK_ERROR_MESSAGE = "Operation failed, using fallback";
    
    public List<Event> getEventsWithFallback(String aggregateId) {
        try {
            return eventStore.getEvents(aggregateId);
        } catch (Exception e) {
            // Log fallback usage
            logger.warn("Using fallback for getEvents on aggregate {}", aggregateId, e);
            
            // Return default or cached state
            return Collections.emptyList();
        }
    }
}

Error Logging Format and Alerting Thresholds

All errors are logged with structured logging including trace_id, span_id, user_id, and error_code:

public class ErrorLogger {
    private static final Logger logger = LoggerFactory.getLogger(ErrorLogger.class);
    
    public void logError(String traceId, String spanId, String userId, 
                        String errorCode, String message, Exception exception) {
        logger.error(
            "TraceId: {}, SpanId: {}, UserId: {}, ErrorCode: {}, Message: {}, Exception: {}",
            traceId, spanId, userId, errorCode, message, exception.getMessage(),
            exception
        );
    }
}

Alerting thresholds:

• Critical: > 100 errors per minute for any error code
• Warning: > 50 errors per minute for validation or domain errors
• Info: > 10 errors per minute for infrastructure errors

Error Handling in Event Processing Pipeline

When processing events, we implement error handling at multiple levels:

public class EventProcessingPipeline {
    private final EventStore eventStore;
    private final EventBus eventBus;
    
    public void processEvent(Event event) {
        try {
            // Validate event format
            validateEvent(event);
            
            // Apply event to aggregate
            Aggregate aggregate = loadAggregate(event.getAggregateId());
            aggregate.applyEvent(event);
            
            // Store event
            eventStore.saveEvent(event);
            
            // Publish to event bus for async processing
            eventBus.publish(event);
            
        } catch (ValidationException e) {
            // Log validation error and potentially send to DLQ
            logger.error("Validation failed for event: {}", event.getId(), e);
            sendToDeadLetterQueue(event, "VALIDATION_ERROR");
            
        } catch (DomainException e) {
            // Log domain error and potentially send to DLQ
            logger.error("Domain rule violated for event: {}", event.getId(), e);
            sendToDeadLetterQueue(event, "DOMAIN_ERROR");
            
        } catch (InfrastructureException e) {
            // Retry logic or escalate to monitoring
            logger.error("Infrastructure error processing event: {}", event.getId(), e);
            throw new RetryableException("Temporary infrastructure issue", e);
        }
    }
    
    private void sendToDeadLetterQueue(Event event, String reason) {
        // Implementation for sending to DLQ
    }
}

11. Security Design

Input Validation and Sanitization Strategy

All external inputs are validated using a combination of:

• Apache Commons Validator for standard validation (email, URL, phone number)
• Custom regex patterns for domain-specific constraints (e.g., event ID format, timestamp validity)
• OWASP Java HTML Sanitizer for any user-provided content that gets rendered in UIs or stored as text

SQL Injection Prevention

• Prepared Statements are used exclusively for all database queries
• Parameterized Queries with named parameters (e.g., SELECT * FROM events WHERE event_id = :eventId) to prevent injection
• No dynamic query construction from user input
• Database user permissions restricted to read/write only on necessary tables

Authentication Middleware Design

• JWT-based authentication with HS256 signing algorithm for stateless session management
• OAuth2 Authorization Code Flow for third-party integrations (e.g., Google, GitHub)
• API Key Authentication for system-to-system communication with rotating keys
• Middleware layer intercepts all requests and validates token or API key before service invocation

Authorization Enforcement

• Role-Based Access Control (RBAC) model with predefined roles:
  • EVENT_ADMIN: Full CRUD access to events and event streams
  • EVENT_READER: Read-only access to events and streams
  • SYSTEM_SERVICE: Internal service access (e.g., event processors, replay services)
• Permission checks performed at service layer using a centralized AuthorizationService that evaluates user roles against resource scope
• Resource-level permissions enforced via composite keys (e.g., event_stream:{streamId}:{userId})

Sensitive Data Encryption at Rest and in Transit

• At-rest encryption: AES-256-GCM using AWS KMS or Azure Key Vault for key management
• In-transit encryption: TLS 1.3 with HSTS header enforcement for all API endpoints
• Event payload encryption: Optional field-level encryption for sensitive fields (e.g., user IDs, PII) using asymmetric keys (RSA-2048 or ECDSA)

PII Handling and Masking Strategy

• PII fields (e.g., user ID, email) are marked with @Sensitive annotation in data models
• Masking policy: Last 4 digits of SSN, masked email (e.g., j***@domain.com)
• Audit logging of PII access with timestamps and user context
• Data retention policy: PII data is deleted after 30 days of event creation or upon user request (GDPR compliance)

Secrets Management

• AWS Secrets Manager or Vault integration for rotating secrets (API keys, database passwords)
• Secret rotation policy: Every 90 days for API keys, every 365 days for database credentials
• Environment-specific secrets stored in encrypted format and injected via secure configuration management tools (e.g., Kubernetes Secrets or Spring Cloud Config)

Audit Logging Design

• Event logging includes:
  • timestamp
  • userId or serviceId
  • actionType (CREATE, UPDATE, DELETE)
  • resourceId (event stream ID or event ID)
  • ipAddress
  • userAgent
  • correlationId (for distributed tracing)
• Log format: Structured JSON with fields like {"timestamp":"2024-01-01T12:00:00Z", "userId":"user_123", "actionType":"CREATE", "resourceId":"stream_abc", "ipAddress":"192.168.1.1"}
• Retention policy: Logs retained for 90 days with daily archiving to S3 or cloud storage
• Compliance: Logs are auditable for GDPR, HIPAA, and SOC2 compliance standards

// Example of sensitive field annotation usage in Java
public class Event {
    @Sensitive
    private String userId;
    
    @Sensitive
    private String email;

    // Getters and setters...
}

12. Testing Strategy

Unit Testing Scope and Coverage Targets

Unit tests will cover 85%+ of business logic and service layer code paths. Each domain object and service method will have dedicated unit tests with:

• Happy path scenarios
• Boundary condition testing (null, empty, max/min values)
• Error condition handling
• State transition validation

Key areas to test:

• Event application logic (state reconstruction from events)
• Command validation and processing
• Aggregate root business rules enforcement
• Event store read/write operations
• Conflict detection and resolution logic

Integration Testing Strategy

Integration tests will validate end-to-end flows using embedded databases (H2/SQLite) and in-memory message brokers (e.g., RabbitMQ in-memory or embedded Kafka). These tests will cover:

1. Event Store Integration: Full round-trip of event persistence and retrieval
2. Aggregate State Reconstruction: End-to-end state rebuild from event stream
3. Command Handling Flow: Command processing through validation → event creation → persistence → state update
4. Event Processing Pipeline: Event consumption and side effect execution

Testcontainers will be used for:

• Database integration (PostgreSQL/MySQL)
• Messaging system integration (Kafka/RabbitMQ)
• External service mocks (e.g., payment gateway, notification service)

Contract Testing

Contract tests will be implemented using Pact for HTTP-based interactions and EventContract testing for event-driven flows. These ensure:

• Service-to-service contract compliance
• Event schema evolution backward compatibility
• Consumer-driven contract validation

Performance Testing Tool and Key Scenarios

Performance testing will use k6 for load testing with the following key scenarios:

1. High Concurrency Command Processing:

   • 500 concurrent users submitting commands
   • Target: p95 latency < 200ms for command processing

2. Event Stream Processing:

   • Simulate 10,000 events per second
   • Target: p99 latency < 500ms for event replay

3. Aggregate Reconstruction Under Load:

   • Reconstruct 10,000 aggregates from event streams
   • Target: p95 reconstruction time < 100ms per aggregate

Test Data Management Strategy

Test data will be managed through:

• Factories: For generating valid test objects with realistic data
• Fixtures: Predefined data sets for specific test scenarios (e.g., "user_with_orders")
• Seeding: Automated data setup for integration tests using database migrations or test data builders

Data generation will use libraries like:

• Faker for realistic synthetic data
• Test Data Builder Pattern for complex object construction
• Database Schema Migrations: For consistent test database state

Test data will be isolated per test environment and cleaned up after each test run to ensure test isolation.

13. Observability Design

Structured Logging Schema

All logs must follow a consistent schema to enable effective search, filtering, and analysis. The logging structure includes:

{
  "timestamp": "2023-10-27T14:30:00.000Z",
  "level": "INFO|WARN|ERROR",
  "service": "event-sourcing-framework",
  "trace_id": "1234567890abcdef1234567890abcdef",
  "span_id": "abcdef1234567890abcdef1234567890",
  "user_id": "user-123",
  "entity_id": "event-stream-456",
  "operation": "handleEvent",
  "event_type": "UserCreated",
  "error_code": "EVENT_STORE_ERROR",
  "message": "Failed to store event in database",
  "error_details": {
    "stack_trace": "...",
    "cause": "Database connection timeout"
  }
}

PII masking rules:

• User IDs are logged as user-123 (masked)
• Email addresses are masked to user***@domain.com
• Credit card numbers are fully redacted

Distributed Tracing Instrumentation

Using OpenTelemetry, we instrument key operations with custom spans:

Span Types and Attributes:

1. Event Handling Span

   • Operation: handleEvent
   • Attributes:
     • event_stream_id: "stream-123"
     • event_type: "UserCreated"
     • event_version: 1
     • processing_time_ms: 150

2. Storage Span

   • Operation: storeEvent
   • Attributes:
     • db_operation: "INSERT"
     • db_table: "event_store"
     • db_duration_ms: 50

3. Projection Span

   • Operation: projectEvent
   • Attributes:
     • projection_name: "UserProjection"
     • projection_duration_ms: 25

Metrics Design

RED Method Metrics:

Metric Name	Type	Labels	Description
event_store_events_processed_total	Counter	event_type, status	Total events processed
event_store_events_processing_time_seconds	Histogram	event_type, operation	Processing duration
event_store_events_errors_total	Counter	error_type, event_type	Event processing errors
event_store_db_query_duration_seconds	Histogram	db_operation, table	Database query times
event_store_projection_errors_total	Counter	projection_name, error_type	Projection processing errors

Additional Service-Level Metrics:

Metric Name	Type	Labels	Description
event_store_active_streams_count	Gauge	-	Current active event streams
event_store_projection_lag_seconds	Gauge	projection_name	Time since last event processed
event_store_memory_usage_bytes	Gauge	-	Memory usage by service

Health Check Endpoint Design

Endpoints:

1. /health - Readiness check

   {
     "status": "healthy",
     "checks": {
       "database": "healthy",
       "message_queue": "healthy",
       "cache": "healthy"
     }
   }

2. /ready - Liveness check

   {
     "status": "ready",
     "services": ["database", "cache", "message-broker"],
     "timestamp": "2023-10-27T14:30:00.000Z"
   }

3. /live - Basic service status

   {
     "status": "alive",
     "uptime_seconds": 3600,
     "version": "v1.2.3"
   }

Alerting Thresholds

Critical Alerts (on-call escalation):

Metric	Threshold	Duration	Alert Severity
event_store_events_errors_total	> 10 per minute	5 min	Critical
event_store_projection_lag_seconds	> 300 seconds	1 min	Critical
event_store_db_query_duration_seconds	> 5 seconds	1 min	Critical
event_store_events_processing_time_seconds	> 2 seconds	1 min	Warning

Warning Alerts:

Metric	Threshold	Duration	Alert Severity
event_store_events_errors_total	> 5 per minute	5 min	Warning
event_store_events_processing_time_seconds	> 1 second	1 min	Warning
event_store_projection_lag_seconds	> 60 seconds	1 min	Warning

Alerting Configuration:

• Slack notifications for critical alerts
• PagerDuty escalation for critical issues
• Email alerts for warning conditions
• Alert grouping by service and metric type

14. Open Questions, Risks & Trade-offs

Decisions requiring product / engineering alignment

1. Event Versioning Strategy - Should we use a single global version for all events or version each event type independently? This affects backward compatibility and migration complexity.
2. Event Schema Evolution - How do we handle breaking changes to event schemas? Do we use a strict schema registry or allow flexible schema evolution with validation?
3. Snapshot Frequency - How often should we create snapshots to balance replay performance vs. storage costs?
4. Event Store Partitioning Strategy - Should we partition by aggregate ID or by time window for better query performance?
5. Replay vs. Recomputation Trade-off - When rebuilding state from events, should we prefer full replay or recomputation from a snapshot?

Known trade-offs and justifications

1. Storage vs. Compute Trade-off - Event sourcing requires storing all events indefinitely, leading to massive storage requirements. We trade compute efficiency for storage cost and auditability.
2. Read Performance vs. Write Performance - Optimizing for fast writes (event persistence) comes at the cost of slower read operations when reconstructing aggregate state.
3. Consistency vs. Availability - Event sourcing with strict ordering can cause availability issues during partitioning, requiring careful CAP theorem balancing.
4. Complexity vs. Simplicity - The framework adds significant complexity for developers but provides powerful audit, debugging, and replay capabilities.
5. Eventual Consistency vs. Strong Consistency - Events may be processed asynchronously, leading to eventual consistency that might not meet all business requirements.

Risks and mitigation strategies

1. Event Store Failure Risk - Single point of failure for all historical data.

   • Mitigation: Implement distributed event stores with replication and automatic failover.

2. Event Replay Performance Degradation - Replaying millions of events can be slow.

   • Mitigation: Implement snapshotting and incremental event processing with proper indexing.

3. Event Schema Evolution Risk - Breaking changes to event schemas can break consumers.

   • Mitigation: Implement robust versioning and migration strategies with schema evolution support.

4. Data Ingestion Bottleneck - High event volume can overwhelm the system.

   • Mitigation: Use async processing, batching, and horizontal scaling of event consumers.

5. Audit Trail Integrity Risk - Events must remain immutable for audit purposes.

   • Mitigation: Implement write-once storage systems and cryptographic signatures for tamper detection.

Technical debt acknowledged

1. Event Schema Maturity - Initial event schemas may not be well-thought-out, requiring future refactoring and migration.
2. Consumer Complexity - Event consumers must handle multiple event types and versions, creating potential for bugs.
3. Snapshot Management - Snapshot creation and maintenance adds operational overhead and complexity.
4. Cross-Service Event Dependencies - Events from one service may not be available when needed by another, creating coupling.
5. Event Ordering Guarantees - Ensuring strict ordering across distributed systems is challenging and may require additional infrastructure.