CQRS Framework Design and Implementation

The CQRS pattern separates read and write operations into distinct models, enabling scalable systems with clear separation of concerns. This document presents a comprehensive low-level design for implementing a robust CQRS framework with event sourcing capabilities and distributed consistency patterns.

1. Problem Statement & Requirements

Clarifying questions

1. What scale and throughput expectations are we targeting?

   • Is this for a high-volume system (e.g., millions of operations per second) or more moderate scale?
   • Are there specific latency targets (e.g., p99 < 100ms)?

2. What consistency models are required for reads vs. writes?

   • Do we need strong consistency for all reads, or eventual consistency is acceptable?
   • Are there specific use cases where read-after-write consistency is critical?

3. What are the key entities in the system and their relationships?

   • Is this a document-centric system (e.g., orders, users) or more transactional?
   • Are there any complex domain relationships (e.g., order items, customer profiles)?

4. What is the read/write ratio?

   • Is this a write-heavy system (e.g., logs, events) or read-heavy (e.g., analytics)?
   • Are there any real-time dashboards or reporting that drive read volume?

5. What latency sensitivity is expected for different operations?

   • Are there any real-time or near-real-time processing requirements (e.g., live notifications)?
   • Is the system expected to support microsecond-level response times for some operations?

6. What authentication and authorization model is expected?

   • Will the system support OAuth2, JWTs, or API keys?
   • Is there a need for role-based access control (RBAC) or attribute-based access control (ABAC)?

7. Are there any external integrations or third-party dependencies?

   • Will the system integrate with other services (e.g., payment gateways, notification systems)?
   • Are there any compliance or audit requirements that need to be considered?

Functional requirements

• Support for Command and Query responsibilities separately (CQRS pattern)
• Ability to define commands, queries, and their respective handlers
• Support for asynchronous command execution and event publishing
• Support for read models that can be updated from write model events
• Integration with a message broker (e.g., Kafka, RabbitMQ) for event-driven architecture
• Support for Saga orchestration for distributed transactions (if needed)
• Ability to define and enforce command validation rules
• Support for idempotent command execution

Non-functional requirements

• Latency Targets:

  • p50: <50ms
  • p95: <100ms
  • p99: <200ms

• Throughput:

  • Minimum 10,000 commands/second for write operations
  • Minimum 50,000 queries/second for read operations

• Availability:

  • 99.9% uptime (5 minutes of downtime/month)

• Scalability:

  • Horizontal scaling support for command and query handlers
  • Ability to scale read models independently

Constraints and assumptions

• The system will be built using Java (Spring Boot) with a focus on microservices architecture
• Message broker is assumed to be Kafka or RabbitMQ (implementation details abstracted)
• The system assumes a single database for write model and read models can be denormalized or backed by separate systems
• All commands will be processed synchronously, with asynchronous events published for read model updates
• The system assumes a single region deployment (multi-region support is out of scope)

Out of scope

• Full database sharding or horizontal partitioning strategies
• Multi-region deployment and cross-region replication
• Support for multiple message brokers or switching between them
• Real-time analytics or streaming data processing (e.g., Flink, Spark)
• UI or frontend integration (focus is on backend infrastructure)
• Support for cross-service transactions beyond the scope of a single service
• Complex business logic validation (e.g., fraud detection, rule engine)
• Support for legacy systems or integration with older protocols (e.g., SOAP)

2. API Design

API Style Selection

For a CQRS framework, I recommend gRPC as the primary API style for the following reasons:

1. Strong typing: gRPC's Protocol Buffers provide compile-time type safety, which is crucial for a framework that needs to handle diverse command/query types
2. Performance: Binary serialization is more efficient than JSON for high-throughput systems
3. Built-in streaming support: Enables real-time event publishing and subscription patterns essential for CQRS
4. Service contracts: Clear, versioned service definitions that can be shared between framework components and consumers
5. Language agnostic: Supports multiple backend languages (Java, Go, Python, etc.) for framework extensibility

API Endpoints

Method	Path	Description	Auth	Rate Limit
POST	/cqrs.v1.CommandService/ExecuteCommand	Execute a command operation	JWT/OAuth2	1000 req/min
GET	/cqrs.v1.QueryService/GetProjection	Retrieve a projection view	JWT/OAuth2	5000 req/min
GET	/cqrs.v1.EventService/SubscribeEvents	Subscribe to event stream	JWT/OAuth2	100 req/min
POST	/cqrs.v1.CommandService/ExecuteBatch	Execute multiple commands in batch	JWT/OAuth2	500 req/min
GET	/cqrs.v1.ProjectionService/GetProjectionState	Get current projection state	JWT/OAuth2	1000 req/min

Request and Response Schemas

Command Execution Request (Protobuf):

message CommandRequest {
  string command_id = 1;
  string command_type = 2;
  bytes payload = 3;
  map<string, string> headers = 4;
  google.protobuf.Timestamp created_at = 5;
}

message CommandResponse {
  string command_id = 1;
  bool success = 2;
  string error_code = 3;
  string error_message = 4;
  google.protobuf.Timestamp processed_at = 5;
}

Query Request (Protobuf):

message QueryRequest {
  string query_id = 1;
  string projection_name = 2;
  string query_filter = 3;
  int32 page_size = 4;
  string cursor = 5;
  map<string, string> headers = 6;
}

message QueryResponse {
  string query_id = 1;
  repeated ProjectionData results = 2;
  string next_cursor = 3;
  int32 total_count = 4;
  google.protobuf.Timestamp query_time = 5;
}

message ProjectionData {
  string projection_id = 1;
  string data = 2;
  google.protobuf.Timestamp updated_at = 3;
}

Event Stream Response (Server-side streaming):

message EventStreamRequest {
  string subscription_id = 1;
  repeated string event_types = 2;
  google.protobuf.Timestamp since = 3;
  map<string, string> headers = 4;
}

message EventStreamResponse {
  string event_id = 1;
  string event_type = 2;
  bytes payload = 3;
  google.protobuf.Timestamp timestamp = 4;
  string source = 5;
}

HTTP Status Codes and Error Format

Standard HTTP Status Codes:

• 200 OK - Successful command execution
• 201 Created - New resource created (e.g., new projection)
• 400 Bad Request - Invalid command/query format
• 401 Unauthorized - Missing or invalid authentication
• 403 Forbidden - Insufficient permissions
• 429 Too Many Requests - Rate limit exceeded
• 500 Internal Server Error - Unexpected system error

Standardized Error Response Format:

{
  "error": {
    "code": "COMMAND_VALIDATION_ERROR",
    "message": "Command payload validation failed",
    "details": {
      "field": "command_payload",
      "reason": "Missing required field: customerId"
    },
    "timestamp": "2023-10-15T14:30:00Z",
    "request_id": "req-12345"
  }
}

API Versioning Strategy

Versioned using semantic versioning in the service namespace:

• cqrs.v1 - Initial stable version
• cqrs.v2 - Major breaking changes (e.g., new event schema)
• cqrs.v3 - Minor feature additions

Versioning handled through:

1. Protocol buffer package names
2. Service method signatures (backward compatible)
3. Client-side version negotiation

Pagination Design

Using cursor-based pagination with the following design:

message QueryRequest {
  string query_id = 1;
  string projection_name = 2;
  string cursor = 3; // Base64 encoded cursor token
  int32 page_size = 4; // Default: 50, Max: 1000
}

Justification:

• Cursor-based pagination is more stable and consistent than offset-based
• Handles data changes during pagination gracefully
• Provides better performance for large datasets
• Supports consistent ordering with stable sort keys

Authentication and Authorization

Authentication:

• JWT Bearer tokens for service-to-service communication
• OAuth2 for external client authentication
• API key support for read-only operations

Authorization:

@PreAuthorize("hasPermission(#command, 'EXECUTE')")
public CommandResponse executeCommand(CommandRequest request) {
    // Implementation checks permissions based on command type and user role
}

Rate Limiting

Per Endpoint Rate Limits:

• Command Execution: 1000 req/min (token bucket algorithm)
• Query Operations: 5000 req/min (sliding window)
• Event Subscriptions: 100 req/min (fixed window)

Implementation Strategy:

@RateLimiter(key = "command_execution", limit = 1000, windowSeconds = 60)
public CommandResponse executeCommand(CommandRequest request) {
    // Rate-limited method
}

Idempotency Design

Idempotency Key Strategy:

@Idempotent(key = "command_id")
public CommandResponse executeCommand(CommandRequest request) {
    // Idempotent command execution
}

Implementation:

• Use command_id as the idempotency key
• Store command execution results in a cache with TTL of 24 hours
• For duplicate requests, return cached result instead of reprocessing
• Support for idempotency key header in HTTP requests

API Versioning Example

// v1 version
syntax = "proto3";
package cqrs.v1;

service CommandService {
  rpc ExecuteCommand(CommandRequest) returns (CommandResponse);
}

// v2 version with new features
syntax = "proto3";
package cqrs.v2;

service CommandService {
  rpc ExecuteCommand(CommandRequest) returns (CommandResponse);
  rpc ExecuteBatch(ExecuteBatchRequest) returns (ExecuteBatchResponse);
}

Security Considerations

1. Transport security: All APIs must use TLS 1.3+
2. Authentication: JWT tokens with proper audience validation
3. Authorization: Role-based access control for command execution
4. Input validation: Protocol buffer validation prevents malformed requests
5. Rate limiting: Prevents abuse and DDoS attacks
6. Idempotency: Protects against duplicate command execution

3. Data Model Design

Entity-Relationship diagram (Mermaid)

erDiagram
    COMMAND {
        string id PK
        string type
        json payload
        timestamp created_at
        timestamp processed_at
        string status
    }

    EVENT {
        string id PK
        string type
        json payload
        timestamp created_at
        string source
    }

    COMMAND_EVENT {
        string command_id FK
        string event_id FK
        timestamp created_at
    }

    COMMAND ||--o{ COMMAND_EVENT : "triggers"
    EVENT ||--o{ COMMAND_EVENT : "is_triggered_by"

    READ_MODEL {
        string id PK
        string type
        json data
        timestamp updated_at
    }

    READ_MODEL ||--o{ COMMAND_EVENT : "consumed_by"

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

Commands Table

CREATE TABLE commands (
    id UUID PRIMARY KEY,
    type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    processed_at TIMESTAMP WITH TIME ZONE,
    status VARCHAR(50) NOT NULL DEFAULT 'pending',
    
    CONSTRAINT chk_status CHECK (status IN ('pending', 'processing', 'completed', 'failed'))
);

CREATE INDEX idx_commands_type ON commands(type);
CREATE INDEX idx_commands_status ON commands(status);
CREATE INDEX idx_commands_created_at ON commands(created_at);

Events Table

CREATE TABLE events (
    id UUID PRIMARY KEY,
    type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    source VARCHAR(255) NOT NULL,
    
    CONSTRAINT chk_event_type CHECK (type ~ '^[a-zA-Z0-9_]+$')
);

CREATE INDEX idx_events_type ON events(type);
CREATE INDEX idx_events_created_at ON events(created_at);

Command-Event Mapping Table

CREATE TABLE command_events (
    command_id UUID REFERENCES commands(id) ON DELETE CASCADE,
    event_id UUID REFERENCES events(id) ON DELETE CASCADE,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    
    PRIMARY KEY (command_id, event_id)
);

CREATE INDEX idx_command_events_event_id ON command_events(event_id);

Read Models Table

CREATE TABLE read_models (
    id UUID PRIMARY KEY,
    type VARCHAR(255) NOT NULL,
    data JSONB NOT NULL,
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    
    CONSTRAINT chk_read_model_type CHECK (type ~ '^[a-zA-Z0-9_]+$')
);

CREATE INDEX idx_read_models_type ON read_models(type);
CREATE INDEX idx_read_models_updated_at ON read_models(updated_at);

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

UUID Strategy Selected

• Justification:
  • Distributed system compatibility (no coordination needed for ID generation)
  • No contention in high-concurrency environments
  • Better audit trail and traceability across systems
  • Support for eventual consistency patterns
  • Avoids the need for centralized ID generation services

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

Index Name	Fields	Type	Justification
idx_commands_type	type	Standard	For command type filtering
idx_commands_status	status	Standard	For status-based queries and state management
idx_commands_created_at	created_at	Standard	For time-range queries and command aging
idx_events_type	type	Standard	For event type filtering
idx_events_created_at	created_at	Standard	For time-range queries and event aging
idx_command_events_event_id	event_id	Standard	For reverse lookup from events to commands
idx_read_models_type	type	Standard	For read model type filtering
idx_read_models_updated_at	updated_at	Standard	For read model freshness queries

Data partitioning / sharding strategy

Sharding by Command Type

• Partition commands by their type field using consistent hashing
• Events and read models are not sharded as they're typically smaller in volume
• Each command type can be handled by a dedicated partition to allow for:
  • Different processing strategies per command type
  • Isolation of high-volume command types
  • Independent scaling of processing units

Soft delete vs. hard delete decision with justification

Soft Delete Strategy

• Commands and Events are soft-deleted using a status field rather than physical deletion
• Read models are hard-deleted when no longer needed (they're derived data)
• Justification:
  • Auditability and compliance requirements
  • Traceability of system evolution through command-event chains
  • Support for replaying events for reprocessing or debugging
  • Avoids complex referential integrity issues in distributed systems

Audit trail design (who, what, when)

CREATE TABLE audit_trail (
    id UUID PRIMARY KEY,
    operation_type VARCHAR(50) NOT NULL,
    entity_type VARCHAR(100) NOT NULL,
    entity_id UUID,
    user_id UUID,
    details JSONB,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_audit_trail_entity_type ON audit_trail(entity_type);
CREATE INDEX idx_audit_trail_created_at ON audit_trail(created_at);

Data archival strategy and retention policy

Archival Strategy

• Commands and Events older than 30 days are archived to cold storage (e.g., S3)
• Read models are retained based on business requirements (typically 90 days)
• Archival is handled by a scheduled job that:
  • Copies data to archive storage
  • Updates status in database to indicate archival
  • Removes archived records from primary tables after confirmation

-- Archival status tracking
ALTER TABLE commands ADD COLUMN archived BOOLEAN DEFAULT FALSE;
ALTER TABLE events ADD COLUMN archived BOOLEAN DEFAULT FALSE;

CREATE INDEX idx_commands_archived ON commands(archived);
CREATE INDEX idx_events_archived ON events(archived);

4. Class Design & Service Layer

UML Class Diagram

classDiagram
    class Command {
        <<interface>>
        execute()
    }
    
    class Query {
        <<interface>>
        execute()
    }
    
    class CommandHandler {
        <<abstract>>
        handle(Command)
    }
    
    class QueryHandler {
        <<abstract>>
        handle(Query)
    }
    
    class CQRSFramework {
        +handleCommand(Command)
        +handleQuery(Query)
        +registerCommandHandler(String, CommandHandler)
        +registerQueryHandler(String, QueryHandler)
    }
    
    class CommandBus {
        +send(Command)
    }
    
    class QueryBus {
        +send(Query)
    }
    
    class EventStore {
        +save(Event)
        +getEvents(String)
    }
    
    class EventBus {
        +publish(Event)
        +subscribe(EventHandler)
    }
    
    class EventHandler {
        <<interface>>
        handle(Event)
    }
    
    class AggregateRoot {
        <<abstract>>
        +getId()
        +apply(Event)
        +getVersion()
    }
    
    class Event {
        <<abstract>>
        +getTimestamp()
        +getAggregateId()
    }
    
    class CommandHandlerRegistry {
        +register(String, CommandHandler)
        +getHandler(String)
    }
    
    class QueryHandlerRegistry {
        +register(String, QueryHandler)
        +getHandler(String)
    }
    
    Command <|-- CreateOrderCommand
    Command <|-- UpdateOrderCommand
    
    Query <|-- GetOrderQuery
    
    CommandHandler <|-- CreateOrderCommandHandler
    CommandHandler <|-- UpdateOrderCommandHandler
    
    QueryHandler <|-- GetOrderQueryHandler
    
    CQRSFramework --> CommandBus
    CQRSFramework --> QueryBus
    CQRSFramework --> CommandHandlerRegistry
    CQRSFramework --> QueryHandlerRegistry
    CQRSFramework --> EventStore
    CQRSFramework --> EventBus
    
    CommandBus --> CommandHandlerRegistry
    QueryBus --> QueryHandlerRegistry
    
    CommandHandlerRegistry --> CommandHandler
    QueryHandlerRegistry --> QueryHandler
    
    AggregateRoot <|-- Order
    Event <|-- OrderCreatedEvent
    Event <|-- OrderUpdatedEvent
    
    EventBus --> EventHandler
    EventHandler <|-- OrderEventHandler
    
    CommandHandler --> AggregateRoot
    QueryHandler --> AggregateRoot

State Machine Diagram for Order

stateDiagram-v2
    [*] --> Created
    Created --> Confirmed : confirm()
    Confirmed --> Shipped : ship()
    Confirmed --> Cancelled : cancel()
    Shipped --> Delivered : deliver()
    Shipped --> Cancelled : cancel()
    Cancelled --> [*]
    Delivered --> [*]

Service Class Structure and Responsibilities

The CQRS framework consists of several key service classes:

1. CQRSFramework: Main orchestrator that manages command/query handling and routing
2. CommandBus: Routes commands to appropriate handlers
3. QueryBus: Routes queries to appropriate handlers
4. EventStore: Persists and retrieves domain events
5. EventBus: Publishes events to subscribers
6. CommandHandlerRegistry: Maintains mapping of command types to handlers
7. QueryHandlerRegistry: Maintains mapping of query types to handlers

Interface / Abstract Class Design

// Command interface
public interface Command {
    String getId();
    String getType();
}

// Query interface
public interface Query {
    String getId();
    String getType();
}

// Command handler abstract class
public abstract class CommandHandler {
    public abstract void handle(Command command);
    
    protected <T extends Command> T validateAndCast(Command command, Class<T> type) {
        if (!type.isInstance(command)) {
            throw new IllegalArgumentException("Invalid command type");
        }
        return type.cast(command);
    }
}

// Query handler abstract class
public abstract class QueryHandler {
    public abstract Object handle(Query query);
    
    protected <T extends Query> T validateAndCast(Query query, Class<T> type) {
        if (!type.isInstance(query)) {
            throw  new IllegalArgumentException("Invalid query type");
        }
        return type.cast(query);
    }
}

// Event interface
public interface Event {
    String getAggregateId();
    long getTimestamp();
    String getType();
}

Business Logic Encapsulation Patterns

We implement the Command Pattern to encapsulate business logic operations:

// Concrete command implementation
public class CreateOrderCommand implements Command {
    private final String id;
    private final String customerId;
    private final List<OrderItem> items;
    
    public CreateOrderCommand(String id, String customerId, List<OrderItem> items) {
        this.id = id;
        this.customerId = customerId;
        this.items = items;
    }
    
    @Override
    public String getId() {
        return id;
    }
    
    @Override
    public String getType() {
        return "CreateOrder";
    }
    
    // Getters...
}

// Command handler using Strategy pattern for validation
public class CreateOrderCommandHandler extends CommandHandler {
    private final OrderValidator validator;
    private final EventPublisher eventPublisher;
    
    public CreateOrderCommandHandler(OrderValidator validator, 
                                     EventPublisher eventPublisher) {
        this.validator = validator;
        this.eventPublisher = eventPublisher;
    }
    
    @Override
    public void handle(Command command) {
        CreateOrderCommand createCmd = validateAndCast(command, CreateOrderCommand.class);
        
        // Validate command using strategy
        validator.validate(createCmd);
        
        // Business logic - create aggregate
        Order order = new Order(createCmd.getId(), createCmd.getCustomerId());
        order.setItems(createCmd.getItems());
        
        // Apply domain events
        OrderCreatedEvent event = new OrderCreatedEvent(
            order.getId(), 
            order.getCustomerId(), 
            order.getItems(),
            System.currentTimeMillis()
        );
        
        // Publish event
        eventPublisher.publish(event);
    }
}

Critical-Path Method Implementation

public class OrderService {
    private final EventStore eventStore;
    private final EventBus eventBus;
    private final OrderRepository orderRepository;
    
    public void processCreateOrder(CreateOrderCommand command) {
        // Transaction boundary - start transaction
        try (Transaction tx = transactionManager.begin()) {
            // 1. Validate command
            validateCreateOrderCommand(command);
            
            // 2. Check for idempotency (if needed)
            String commandId = command.getId();
            if (commandId != null && orderRepository.exists(commandId)) {
                // Idempotent operation - return early or throw exception
                return;
            }
            
            // 3. Create order aggregate with optimistic locking
            Order order = new Order(command.getId(), command.getCustomerId());
            order.setItems(command.getItems());
            
            // 4. Apply domain events
            OrderCreatedEvent event = new OrderCreatedEvent(
                order.getId(),
                order.getCustomerId(),
                order.getItems(),
                System.currentTimeMillis()
            );
            
            // 5. Store events in event store (with versioning)
            eventStore.save(event);
            
            // 6. Update aggregate state in repository
            orderRepository.save(order);
            
            // 7. Publish events to event bus (async)
            eventBus.publish(event);
            
            // 8. Commit transaction
            tx.commit();
            
        } catch (Exception e) {
            // Transaction rollback handled by try-with-resources
            throw new OrderProcessingException("Failed to process order creation", e);
        }
    }
    
    private void validateCreateOrderCommand(CreateOrderCommand command) {
        if (command.getCustomerId() == null || command.getCustomerId().isEmpty()) {
            throw new ValidationException("Customer ID is required");
        }
        
        if (command.getItems() == null || command.getItems().isEmpty()) {
            throw new ValidationException("Order must have at least one item");
        }
    }
}

Transaction Boundary Design

Transaction boundaries in CQRS are more granular than traditional monolithic systems:

1. Command Processing: Each command is processed in a single transaction
2. Event Storage: Events are stored atomically with the aggregate state update
3. Idempotency Handling: Commands can be idempotent with proper tracking
4. Cross-Service Transactions: Distributed transactions use Saga pattern for coordination

Service Interface Contracts

public interface OrderService {
    void createOrder(CreateOrderCommand command);
    Order getOrder(String orderId);
    void updateOrder(UpdateOrderCommand command);
    List<Order> getOrdersByCustomer(String customerId);
}

Saga / Orchestration Pattern

For distributed operations, we implement a saga pattern:

public class OrderSaga {
    private final SagaManager sagaManager;
    
    public void handleCreateOrder(CreateOrderCommand command) {
        String sagaId = UUID.randomUUID().toString();
        
        // Start saga
        sagaManager.startSaga(sagaId, "OrderCreation");
        
        try {
            // Step 1: Create order in database
            sagaManager.executeStep(sagaId, "CreateOrder", 
                () -> orderRepository.save(order));
            
            // Step 2: Reserve inventory (external service call)
            sagaManager.executeStep(sagaId, "ReserveInventory",
                () -> inventoryService.reserveItems(order.getItems()));
            
            // Step 3: Charge customer (external service call)
            sagaManager.executeStep(sagaId, "ChargeCustomer",
                () -> paymentService.charge(order.getCustomerId(), order.getTotal()));
            
            // Commit saga
            sagaManager.completeSaga(sagaId);
            
        } catch (Exception e) {
            // Compensate for failed steps
            sagaManager.compensate(sagaId, e);
        }
    }
}

Domain Object / Value Object Design

// Value Objects
public class Money {
    private final BigDecimal amount;
    private final String currency;
    
    public Money(BigDecimal amount, String currency) {
        this.amount = amount;
        this.currency = currency;
    }
    
    // Getters and validation methods...
}

public class OrderItem {
    private final String productId;
    private final int quantity;
    private final Money price;
    
    // Constructor, getters...
}

// Aggregate Root
public class Order {
    private final String id;
    private final String customerId;
    private List<OrderItem> items;
    private OrderStatus status;
    private long version;
    
    public Order(String id, String customerId) {
        this.id = id;
        this.customerId = customerId;
        this.status = OrderStatus.CREATED;
        this.version = 0;
    }
    
    public void setItems(List<OrderItem> items) {
        this.items = items;
    }
    
    public void apply(OrderCreatedEvent event) {
        // Apply event to aggregate state
        this.status = OrderStatus.CONFIRMED;
        this.version++;
    }
    
    // Getters...
}

5. Concurrency & Thread Safety

Concurrency Model

The CQRS framework will leverage a multi-threaded approach using Java's java.util.concurrent package and thread-safe constructs. The system will utilize a fixed thread pool for processing command and query operations, with dedicated threads for each operation type to ensure isolation between write and read workloads. For asynchronous processing, the framework will support reactive streams using Project Reactor or RxJava to handle backpressure and non-blocking operations efficiently.

Race Condition Prevention Strategy

Race conditions in CQRS are mitigated through:

1. Command Processing Isolation: Each command is processed in a single-threaded context with proper locking to prevent concurrent modifications to the same aggregate
2. Eventual Consistency Patterns: Read models are updated asynchronously, ensuring that read operations don't block write operations
3. Optimistic Locking: For aggregates with versioned state, optimistic locking prevents concurrent modifications to the same entity
4. Idempotency: Commands are designed to be idempotent to handle duplicate processing scenarios gracefully

Optimistic vs. Pessimistic Locking Selection

Optimistic Locking is selected for the CQRS framework due to:

• Higher throughput in read-heavy systems (typical in CQRS)
• Reduced contention on database resources
• Better scalability for distributed systems
• Natural fit with eventual consistency model

Implementation approach:

// Example optimistic locking in aggregate root
public class Order {
    private Long version;
    
    public void updateStatus(OrderStatus newStatus, Long expectedVersion) {
        if (!expectedVersion.equals(this.version)) {
            throw new OptimisticLockException("Order has been modified concurrently");
        }
        this.status = newStatus;
        this.version++;
    }
}

Deadlock Prevention Strategy

Deadlock prevention is achieved through:

1. Lock Ordering: All commands acquire locks in a consistent order (e.g., by entity ID)
2. Timeout Configuration: Commands have timeout limits to prevent indefinite blocking
3. Hierarchical Locking: Use of lock hierarchies to avoid circular dependencies
4. Transaction Isolation Levels: Read committed or repeatable read isolation levels to prevent phantom reads

Idempotency Implementation for Concurrent Duplicate Requests

Idempotency is implemented at the command level using a command deduplication mechanism:

@Service
public class CommandDeduplicationService {
    private final Set<String> processedCommands = new ConcurrentHashSet<>();
    
    public boolean isProcessed(String commandId) {
        return processedCommands.contains(commandId);
    }
    
    public void markAsProcessed(String commandId) {
        processedCommands.add(commandId);
    }
    
    public <T> T executeIfNotProcessed(String commandId, Supplier<T> operation) {
        if (isProcessed(commandId)) {
            return null; // or return cached result
        }
        
        T result = operation.get();
        markAsProcessed(commandId);
        return result;
    }
}

The system uses a combination of:

• Command ID tracking: Unique identifiers for each command
• In-memory cache with TTL: Temporary storage of processed commands (5 minutes)
• Distributed cache integration: For multi-instance deployments
• Idempotency key validation: Commands with same key are treated as duplicates

Thread Safety in Event Handling

Event handlers in the CQRS framework are designed to be thread-safe:

1. Immutable Event Objects: Events are immutable value objects
2. Stateless Handlers: Event handlers do not maintain state between invocations
3. Thread-Safe Storage: Read model updates are synchronized using atomic operations or thread-safe collections
4. Async Processing: Events are processed asynchronously to prevent blocking the command processing thread

Concurrency Control for Read Models

Read models are updated through a concurrent update mechanism:

@Component
public class ReadModelUpdater {
    private final Map<String, AtomicReference<ReadModel>> readModels = new ConcurrentHashMap<>();
    
    public void updateReadModel(String modelId, ReadModel updatedModel) {
        readModels.computeIfAbsent(modelId, k -> new AtomicReference<>())
                  .set(updatedModel);
    }
    
    public ReadModel getReadModel(String modelId) {
        AtomicReference<ReadModel> ref = readModels.get(modelId);
        return ref != null ? ref.get() : null;
    }
}

Performance Considerations

1. Thread Pool Configuration: Fixed-size thread pools with appropriate sizing based on workload characteristics
2. Lock Granularity: Fine-grained locking to minimize contention while maintaining consistency
3. Batch Processing: Commands and events are batched where appropriate to reduce thread overhead
4. Memory Management: Efficient use of memory to prevent GC pressure during high-concurrency scenarios

Scalability Implications

The concurrency model is designed to scale horizontally:

1. Partitioned Command Processing: Commands can be distributed across multiple instances
2. Shared State Management: Use of external coordination services (Redis, ZooKeeper) for distributed locking
3. Asynchronous Event Processing: Events are processed in parallel without blocking command execution
4. Resource Pooling: Efficient resource utilization through connection pooling and thread pool management

This approach ensures that the CQRS framework can handle high-concurrency workloads while maintaining data consistency and system responsiveness. The design balances performance requirements with the need for predictable behavior in distributed environments.

6. Sequence Diagrams for Key Flows

Happy Path Sequence Diagram

sequenceDiagram
    participant Client
    participant CommandHandler
    participant EventBus
    participant WriteModelRepo
    participant QueryModelRepo
    participant EventProcessor

    Client->>CommandHandler: Submit CreateOrderCommand
    CommandHandler->>WriteModelRepo: persist(Order)
    WriteModelRepo-->>CommandHandler: OrderSaved
    CommandHandler->>EventBus: publish(OrderCreatedEvent)
    EventBus->>EventProcessor: dispatch OrderCreatedEvent
    EventProcessor->>QueryModelRepo: update(OrderView)
    QueryModelRepo-->>EventProcessor: OrderViewUpdated
    CommandHandler-->>Client: return OrderCreatedResponse

Error / Edge Case Sequence Diagram

sequenceDiagram
    participant Client
    participant CommandHandler
    participant EventBus
    participant WriteModelRepo
    participant QueryModelRepo
    participant EventProcessor
    participant ErrorHandler

    Client->>CommandHandler: Submit CreateOrderCommand
    CommandHandler->>WriteModelRepo: persist(Order)
    WriteModelRepo-->>CommandHandler: DatabaseError
    CommandHandler->>ErrorHandler: handle(DatabaseError)
    ErrorHandler->>EventBus: publish(OrderCreationFailedEvent)
    EventBus->>EventProcessor: dispatch OrderCreationFailedEvent
    EventProcessor->>QueryModelRepo: update(OrderFailedView)
    QueryModelRepo-->>EventProcessor: OrderFailedViewUpdated
    CommandHandler-->>Client: return OrderCreationFailedResponse

Async Flow Sequence Diagram

sequenceDiagram
    participant Client
    participant CommandHandler
    participant EventBus
    participant EventProcessor
    participant NotificationService

    Client->>CommandHandler: Submit CreateOrderCommand
    CommandHandler->>WriteModelRepo: persist(Order)
    WriteModelRepo-->>CommandHandler: OrderSaved
    CommandHandler->>EventBus: publish(OrderCreatedEvent)
    EventBus->>EventProcessor: dispatch OrderCreatedEvent
    EventProcessor->>NotificationService: sendOrderConfirmation()
    NotificationService-->>EventProcessor: confirmationSent
    EventProcessor->>QueryModelRepo: update(OrderView)
    QueryModelRepo-->>EventProcessor: OrderViewUpdated
    CommandHandler-->>Client: return OrderCreatedResponse

7. Caching Design

Cache Strategy Selection

For a CQRS framework, we'll use cache-aside pattern as the primary caching strategy. This approach gives us fine-grained control over cache operations and allows for better handling of eventual consistency between read and write models.

Cache Key Design

Cache keys will follow a structured namespace pattern:

{cacheNamespace}:{entityType}:{entityId}:{version}

Example key structure for a user read model:

readmodel:user:12345:v1

TTL Configuration

Data Type	TTL	Reason
Read Models (User, Order)	5 minutes	Frequent reads with acceptable staleness
Configuration Data	1 hour	Infrequently updated, high consistency requirement
Session/Token Cache	30 minutes	Short-lived tokens with refresh capability
Search Indexes	15 minutes	Real-time search results with moderate freshness needs

Cache Eviction Policy

We'll implement a hybrid approach combining:

1. LRU (Least Recently Used) - Primary eviction policy for most data types
2. TTL-based - Secondary policy to handle stale entries in high-traffic scenarios
3. Size-based - When cache reaches capacity, evict least recently used items

Cache Stampede Prevention

To prevent thundering herds:

1. Probabilistic Early Expiry: Set cache TTL to 75% of configured TTL, allowing for early refresh
2. Distributed Locks: Use Redis distributed locks during cache rebuild operations
3. Cache Warm-up Strategy: Pre-warm frequently accessed read models during low-traffic periods

Distributed Cache Architecture

We'll use Redis Cluster with:

• 3 master nodes for data distribution
• 3 replica nodes per master for high availability
• Automatic failover with Redis Sentinel monitoring
• Sharding based on entity type and ID hash

Cache Invalidation Strategy

We'll implement event-driven invalidation:

1. Write Model Updates: When a write model is updated, publish cache invalidation events
2. Read Model Invalidation: Consumers listen for these events and invalidate corresponding read models
3. Explicit Purge: Manual cache clearing for administrative operations

// Cache invalidation event handler example
@Component
public class ReadModelCacheInvalidator {
    
    private final RedisTemplate<String, Object> redisTemplate;
    private final ObjectMapper objectMapper;
    
    @EventListener
    public void handleWriteModelUpdate(WriteModelUpdatedEvent event) {
        String cacheKey = buildCacheKey(event.getEntityType(), event.getEntityId());
        
        // Delete from cache
        redisTemplate.delete(cacheKey);
        
        // Optionally publish read model invalidation event for downstream consumers
        publishReadModelInvalidationEvent(event.getEntityType(), event.getEntityId());
    }
    
    private String buildCacheKey(String entityType, String entityId) {
        return String.format("readmodel:%s:%s:v1", entityType, entityId);
    }
    
    private void publishReadModelInvalidationEvent(String entityType, String entityId) {
        // Publish to message queue for async invalidation
        ReadModelInvalidatedEvent invalidationEvent = new ReadModelInvalidatedEvent(
            entityType, entityId, System.currentTimeMillis());
        
        // Publish via message broker (e.g., Kafka)
        messagePublisher.publish("readmodel.invalidated", invalidationEvent);
    }
}

Cache Hit/Miss Metrics

We'll track cache performance with:

• Hit Ratio: Cache hits / (cache hits + misses) - target > 85%
• Miss Rate: Cache misses / total requests - target < 15%
• Eviction Rate: Evicted items / total items - target < 5%

Cache Read-Through Pattern

// Read-through cache implementation for CQRS read models
@Component
public class ReadModelCache {
    
    private final RedisTemplate<String, Object> redisTemplate;
    private final ObjectMapper objectMapper;
    private final ReadModelRepository readModelRepository;
    
    public <T> T getOrCreate(String cacheKey, Class<T> type, Supplier<T> loader) {
        // Try cache first
        String cachedValue = redisTemplate.opsForValue().get(cacheKey);
        
        if (cachedValue != null) {
            try {
                return objectMapper.readValue(cachedValue, type);
            } catch (IOException e) {
                // Log error but continue with database fetch
                log.warn("Failed to deserialize cached value for key: {}", cacheKey, e);
            }
        }
        
        // Cache miss - fetch from database
        T result = loader.get();
        
        // Store in cache with TTL
        if (result != null) {
            try {
                String jsonValue = objectMapper.writeValueAsString(result);
                redisTemplate.opsForValue().set(cacheKey, jsonValue, 5, TimeUnit.MINUTES);
            } catch (IOException e) {
                log.error("Failed to serialize object for caching", e);
            }
        }
        
        return result;
    }
}

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 that defines the operations required for data access.

public interface OrderRepository {
    Order save(Order order);
    Optional<Order> findById(UUID orderId);
    List<Order> findByCustomerId(UUID customerId);
    void delete(UUID orderId);
}

public interface OrderViewRepository {
    OrderView save(OrderView orderView);
    Optional<OrderView> findById(UUID orderId);
    List<OrderView> findByCustomerId(UUID customerId);
}

ORM vs. Raw Query Decisions

For write operations (Command side), raw SQL queries are preferred to maintain fine-grained control over transaction boundaries and ensure optimal performance for complex operations. For read operations (Query side), we use an ORM like Hibernate to leverage object-relational mapping and simplify complex joins.

@Repository
public class OrderRepositoryImpl implements OrderRepository {
    
    @PersistenceContext
    private EntityManager entityManager;
    
    @Override
    public Order save(Order order) {
        if (order.getId() == null) {
            entityManager.persist(order);
        } else {
            entityManager.merge(order);
        }
        return order;
    }
    
    @Override
    public Optional<Order> findById(UUID orderId) {
        return Optional.ofNullable(entityManager.find(Order.class, orderId));
    }
    
    // Raw SQL for performance-critical operations
    @Transactional
    public void updateOrderStatus(UUID orderId, OrderStatus status) {
        String sql = "UPDATE orders SET status = :status WHERE id = :id";
        entityManager.createNativeQuery(sql)
            .setParameter("status", status.name())
            .setParameter("id", orderId)
            .executeUpdate();
    }
}

Read vs. Write Separation (CQRS Pattern)

The CQRS pattern is implemented by separating read and write models into distinct repositories:

// Write Model - Command Side
@Repository
public class OrderCommandRepository {
    private final EntityManager entityManager;
    
    public Order save(Order order) {
        entityManager.persist(order);
        return order;
    }
}

// Read Model - Query Side  
@Repository
public class OrderQueryRepository {
    private final EntityManager entityManager;
    
    public List<OrderView> findOrdersByCustomerId(UUID customerId) {
        String jpql = "SELECT new com.example.OrderView(o.id, o.customerId, o.status) " +
                     "FROM Order o WHERE o.customerId = :customerId";
        return entityManager.createQuery(jpql, OrderView.class)
                           .setParameter("customerId", customerId)
                           .getResultList();
    }
}

Connection Pool Configuration

Connection pools are configured with appropriate sizing based on expected concurrent operations:

# application.yml
spring:
  datasource:
    hikari:
      maximum-pool-size: 20
      minimum-idle: 5
      connection-timeout: 30000
      idle-timeout: 600000
      max-lifetime: 1800000
      leak-detection-threshold: 60000

N+1 Query Prevention Strategy

To prevent N+1 query problems, eager loading and batch fetching are used where appropriate:

@Entity
public class Order {
    @Id
    private UUID id;
    
    @ManyToOne(fetch = FetchType.LAZY)
    @JoinColumn(name = "customer_id")
    private Customer customer;
    
    @OneToMany(mappedBy = "order", fetch = FetchType.EAGER)
    private List<OrderItem> items;
    
    // Avoid N+1 by using JOIN FETCH in JPQL
    @Query("SELECT o FROM Order o JOIN FETCH o.items WHERE o.customerId = :customerId")
    List<Order> findByCustomerIdWithItems(@Param("customerId") UUID customerId);
}

Batch Operation Design

Batch operations are implemented to handle bulk data processing efficiently:

@Repository
public class OrderBatchRepository {
    
    @PersistenceContext
    private EntityManager entityManager;
    
    public void batchSaveOrders(List<Order> orders) {
        int batchSize = 1000;
        for (int i = 0; i < orders.size(); i += batchSize) {
            List<Order> batch = orders.subList(i, Math.min(i + batchSize, orders.size()));
            
            // Use bulk insert for better performance
            String sql = "INSERT INTO orders (id, customer_id, status, created_at) VALUES ";
            // ... build bulk insert query with parameters
            
            entityManager.createNativeQuery(sql).executeUpdate();
            
            // Clear persistence context to avoid memory issues
            entityManager.flush();
            entityManager.clear();
        }
    }
}

Read-Write Splitting Strategy

Separate data sources are configured for read and write operations:

@Configuration
public class DataSourceConfig {
    
    @Bean
    @Primary
    @ConfigurationProperties(prefix = "spring.datasource.write")
    public DataSource writeDataSource() {
        return DataSourceBuilder.create().build();
    }
    
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.read")
    public DataSource readDataSource() {
        return DataSourceBuilder.create().build();
    }
    
    @Bean
    public DataSource routingDataSource() {
        RoutingDataSource routingDataSource = new RoutingDataSource();
        Map<Object, Object> dataSourceMap = new HashMap<>();
        dataSourceMap.put("write", writeDataSource());
        dataSourceMap.put("read", readDataSource());
        routingDataSource.setTargetDataSources(dataSourceMap);
        return routingDataSource;
    }
}

Transaction Management

Transaction boundaries are carefully defined to ensure data consistency:

@Service
@Transactional(readOnly = true)
public class OrderQueryService {
    
    @Autowired
    private OrderQueryRepository orderQueryRepository;
    
    @Transactional(readOnly = false)
    public void processOrderCommand(OrderCommand command) {
        // All operations within this method are in a single transaction
        Order order = new Order(command);
        orderRepository.save(order);
        
        // Publish event to message queue
        eventPublisher.publish(new OrderCreatedEvent(order.getId()));
    }
}

Query Optimization

Indexing and query optimization strategies are applied:

-- Indexes for performance optimization
CREATE INDEX idx_orders_customer_id ON orders(customer_id);
CREATE INDEX idx_orders_status_created_at ON orders(status, created_at);
CREATE INDEX idx_order_items_order_id ON order_items(order_id);

-- Optimized query with proper joins
SELECT o.id, o.customer_id, o.status, oi.product_id, oi.quantity
FROM orders o
JOIN order_items oi ON o.id = oi.order_id
WHERE o.customer_id = ?
ORDER BY o.created_at DESC;

Data Access Layer Testing Strategy

The repository layer is tested with embedded database and mock-based approaches:

@ExtendWith(MockitoExtension.class)
class OrderRepositoryTest {
    
    @Mock
    private EntityManager entityManager;
    
    @InjectMocks
    private OrderRepositoryImpl orderRepository;
    
    @Test
    void shouldSaveOrder() {
        // Given
        Order order = new Order();
        
        // When
        Order savedOrder = orderRepository.save(order);
        
        // Then
        verify(entityManager).persist(order);
    }
}

9. Asynchronous Processing & Messaging

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

In a CQRS framework, the asynchronous processing model is fundamental to achieving scalability and decoupling between read and write models. The decision to make an operation asynchronous is based on the following principles:

1. Write Model Operations: All write operations (e.g., createUser, updateOrder) are processed synchronously to ensure immediate consistency and validation.
2. Event Publishing: After a successful write operation, domain events are published asynchronously to notify downstream systems and update read models.
3. Read Model Updates: Read model updates are handled asynchronously through event consumers to avoid blocking the write operation and to support high-concurrency reads.
4. External Integrations: Integrations with external systems (e.g., payment gateways, email services) are asynchronous to prevent write operation timeouts and improve resilience.

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

We use JSON Schema for simplicity and flexibility in a CQRS context, with versioning handled through a namespace-based approach:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "UserCreatedEvent",
  "version": "1.0.0",
  "type": "object",
  "properties": {
    "eventId": { "type": "string" },
    "eventType": { "type": "string", "const": "UserCreatedEvent" },
    "timestamp": { "type": "string", "format": "date-time" },
    "userId": { "type": "string" },
    "email": { "type": "string" },
    "createdAt": { "type": "string", "format": "date-time" }
  },
  "required": ["eventId", "eventType", "timestamp", "userId", "email"]
}

Versioning is handled by prefixing the event name with a version: v1.UserCreatedEvent, v2.UserCreatedEvent.

Dead letter queue strategy and poison message handling

Poison messages (messages that repeatedly fail to process) are handled using a Dead Letter Queue (DLQ) pattern:

1. Retry Limit: Messages are retried up to 3 times with exponential backoff.
2. DLQ Routing: After max retries, the message is routed to a DLQ topic with metadata (original queue, error details).
3. Manual Inspection: Failed messages in DLQ are logged with full context and can be manually reprocessed by an operator.

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

We implement at-least-once delivery because:

1. Idempotency in Consumers: Read model consumers are designed to be idempotent, ensuring that duplicate events do not cause data inconsistency.
2. Event Sourcing Compatibility: CQRS naturally supports at-least-once delivery as events are immutable and can be replayed safely.
3. Simplicity: Implementing exactly-once delivery adds significant complexity (e.g., distributed transactions, message deduplication) that is not justified for most use cases in a CQRS framework.

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

Retry policy is implemented with the following configuration:

• Max Attempts: 3 retries
• Backoff Strategy: Exponential backoff with jitter (delay = baseDelay * 2^attempt + random(0, 100ms))
• Non-Retryable Errors:
  • ValidationException: Invalid input data (no point retrying)
  • AuthorizationException: Permission denied (retry won't help)
  • BusinessLogicException: Business rule violations

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

We do NOT use the outbox pattern in this design because:

1. Transactional Consistency: Events are published after the write transaction is committed, ensuring that if the write fails, the event is not published.
2. Simplified Architecture: The outbox pattern adds complexity (e.g., event table, background workers) that is unnecessary for a well-designed CQRS system with proper transactional boundaries.
3. Eventual Consistency: The system is designed to tolerate eventual consistency in read models, which aligns with the CQRS principle.

Sequence diagram for all async flows (Mermaid)

sequenceDiagram
    participant W as Write Model
    participant E as Event Publisher
    participant Q as Message Queue (e.g., Kafka)
    participant R as Read Model Consumer

    W->>E: Publish Domain Event
    E->>Q: Send to Queue
    Q->>R: Deliver Event
    R->>R: Process and Update Read Model

10. Error Handling Design

Error Classification Hierarchy

The error handling design follows a three-tier hierarchy:

1. Domain Errors - Business logic violations (e.g., insufficient funds, invalid state transitions)
2. Infrastructure Errors - System-level issues (e.g., database connection failure, timeout)
3. Validation Errors - Input validation failures (e.g., missing required fields, invalid format)

public abstract class BaseException extends RuntimeException {
    private final String errorCode;
    private final int httpStatus;

    public BaseException(String message, String errorCode, int httpStatus) {
        super(message);
        this.errorCode = errorCode;
        this.httpStatus = httpStatus;
    }

    // Getters
}

public class DomainException extends BaseException {
    public DomainException(String message, String errorCode) {
        super(message, errorCode, 400);
    }
}

public class InfrastructureException extends BaseException {
    public InfrastructureException(String message, String errorCode) {
        super(message, errorCode, 500);
    }
}

public class ValidationException extends BaseException {
    public ValidationException(String message, String errorCode) {
        super(message, errorCode, 400);
    }
}

Error Propagation Strategy

Errors are propagated up the call stack with appropriate wrapping to preserve context:

• Service Layer: Catch infrastructure exceptions and wrap them in domain exceptions
• Controller Layer: Convert domain/infrastructure exceptions to standardized API responses
• Global Exception Handler: Handle all uncaught exceptions and return proper HTTP status codes

Retry Strategy with Exponential Backoff and Jitter

For transient infrastructure errors (e.g., database timeouts, network issues), a retry mechanism is implemented:

public class RetryableOperation {
    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 <T> T executeWithRetry(Supplier<T> operation) throws Exception {
        int attempts = 0;
        Exception lastException = null;

        while (attempts < MAX_RETRIES) {
            try {
                return operation.get();
            } catch (InfrastructureException e) {
                lastException = e;
                attempts++;
                
                if (attempts >= MAX_RETRIES) {
                    throw e;
                }

                long delay = (long) (INITIAL_DELAY_MS * Math.pow(BACKOFF_MULTIPLIER, attempts - 1));
                delay += (long) (delay * JITTER_RANGE * (Math.random() - 0.5) * 2);
                
                Thread.sleep(delay);
            }
        }

        throw lastException;
    }
}

Circuit Breaker Design

For external dependencies, a circuit breaker pattern prevents cascading failures:

public class CircuitBreaker {
    private final int failureThreshold;
    private final long timeout;
    private final int successThreshold;

    private volatile CircuitState state = CircuitState.CLOSED;
    private int failureCount = 0;
    private long lastFailureTime = 0;

    public enum CircuitState {
        CLOSED, OPEN, HALF_OPEN
    }

    public <T> T execute(Supplier<T> operation) throws Exception {
        switch (state) {
            case CLOSED:
                try {
                    T result = operation.get();
                    onSuccess();
                    return result;
                } catch (Exception e) {
                    onFailure(e);
                    throw e;
                }
            case OPEN:
                if (System.currentTimeMillis() - lastFailureTime > timeout) {
                    state = CircuitState.HALF_OPEN;
                    return operation.get();
                } else {
                    throw new InfrastructureException("Circuit breaker is OPEN", "CIRCUIT_OPEN");
                }
            case HALF_OPEN:
                try {
                    T result = operation.get();
                    state = CircuitState.CLOSED;
                    return result;
                } catch (Exception e) {
                    state = CircuitState.OPEN;
                    throw e;
                }
        }
        throw new IllegalStateException("Unexpected circuit state");
    }

    private void onSuccess() {
        failureCount = 0;
        if (state == CircuitState.HALF_OPEN) {
            state = CircuitState.CLOSED;
        }
    }

    private void onFailure(Exception e) {
        failureCount++;
        lastFailureTime = System.currentTimeMillis();

        if (failureCount >= failureThreshold) {
            state = CircuitState.OPEN;
        }
    }
}

Fallback Behavior Design

For critical dependencies, fallbacks provide graceful degradation:

public class FallbackService {
    private final ExternalService externalService;
    
    public String getUserProfileFallback(String userId, Exception cause) {
        // Log the fallback for monitoring
        logger.warn("Using fallback for user profile lookup: {}", userId, cause);
        
        // Return default or cached value
        return "default-profile";
    }
    
    public CompletableFuture<String> getUserProfile(String userId) {
        return CompletableFuture.supplyAsync(() -> 
            externalService.getUserProfile(userId)
                .exceptionally(cause -> getUserProfileFallback(userId, cause)));
    }
}

Error Logging Format and Alerting Thresholds

Structured logging includes trace context and error details:

public class ErrorLogger {
    private static final Logger logger = LoggerFactory.getLogger(ErrorLogger.class);

    public void logError(String operation, Exception error, Map<String, Object> context) {
        logger.error(
            "Operation failed: {} | Error Code: {} | Message: {} | Context: {}",
            operation,
            error.getClass().getSimpleName(),
            error.getMessage(),
            context,
            error
        );
    }
}

Alerting thresholds:

• Critical: >50% of requests failing in 5-minute window (e.g., DB connection failures)
• Warning: >10% of requests failing in 5-minute window (e.g., validation errors)
• Info: >1% of requests failing in 5-minute window (e.g., business logic errors)

11. Security Design

Input Validation and Sanitization Strategy

All API inputs are validated using a combination of:

• Bean Validation (JSR-380) for structured validation rules (e.g., @NotNull, @Size, @Pattern)
• Custom validators for business-specific constraints (e.g., valid UUID format, allowed enum values)
• HTML sanitization using OWASP Java HTML Sanitizer for any user-provided content that gets rendered in UIs

SQL Injection Prevention

• Prepared statements are used exclusively for all database queries
• Parameterized queries with named parameters (e.g., SELECT * FROM users WHERE id = :userId) to prevent injection
• No dynamic query building from user input strings
• ORM usage (Hibernate) with proper mapping and no raw SQL string concatenation

Authentication Middleware Design

• JWT-based authentication with refresh token support for stateless session management
• JWTs are validated using JWK Set endpoint (for public key verification) or symmetric signing with shared secret (in dev)
• Middleware intercepts all requests and validates the JWT token, extracting user context for authorization
• Token expiration is enforced with 15-minute access tokens and 7-day refresh tokens

Authorization Enforcement

• Role-Based Access Control (RBAC) model with:
  • User roles: ADMIN, USER, ANONYMOUS
  • Resource-level permissions: READ, WRITE, DELETE
• Authorization checks performed at service layer using a centralized permission checker:

  public class PermissionChecker {
      public boolean hasPermission(User user, String resource, String action) {
          return user.getRoles().stream()
              .anyMatch(role -> role.hasPermission(resource, action));
      }
  }

• ABAC (Attribute-Based Access Control) is considered for future complex access rules but not implemented in this design

Sensitive Data Encryption

• At-rest encryption using AES-256-GCM with keys managed by AWS KMS (or equivalent in other cloud providers)
• In-transit encryption enforced via TLS 1.3 for all external communications
• Database fields containing sensitive data (e.g., personal identifiers, payment info) are encrypted using field-level encryption

PII Handling and Masking Strategy

• PII fields (e.g., email, phone, SSN) are encrypted at rest and masked in logs
• Masking policy:
  • Email: j***@gmail.com
  • Phone: +1 (555) ***-****
• Log redaction via structured logging with PII fields excluded from logs or masked

Secrets Management

• Secrets stored in AWS Secrets Manager (or equivalent) with rotation policies:
  • JWT signing keys rotated every 90 days
  • Database credentials rotated monthly
• Application code accesses secrets via secure SDKs with automatic caching and refresh

Audit Logging Design

• Audit events captured for all write operations (CREATE, UPDATE, DELETE) on sensitive entities
• Event structure includes:

  {
    "timestamp": "2023-10-27T14:30:00Z",
    "userId": "uuid-123",
    "action": "CREATE_USER",
    "resource": "user",
    "details": {
      "userId": "uuid-456",
      "ipAddress": "192.168.1.100"
    }
  }

• Retention policy: 90 days for audit logs, archived to S3 with lifecycle policies
• Log format follows structured JSON schema with trace_id and span_id for distributed tracing correlation

Security Headers and CORS

• Security headers added to all HTTP responses:
  • Content-Security-Policy
  • X-Content-Type-Options: nosniff
  • X-Frame-Options: DENY
  • X-XSS-Protection: 1; mode=block
• CORS policy configured to allow only trusted origins (e.g., https://app.mycompany.com)

Vulnerability Scanning and Dependency Management

• Static analysis tools (e.g., SonarQube, Checkmarx) integrated into CI/CD pipeline to detect vulnerabilities in code and dependencies
• Dependency scanning using OWASP Dependency Check or Snyk to identify known vulnerabilities in libraries
• Automated security testing as part of CI pipeline (e.g., OWASP ZAP integration)

12. Testing Strategy

Unit Testing Scope and Coverage Targets (%)

• Service Layer: 85% code coverage, focusing on business logic and edge cases
• Repository Layer: 90% coverage with mock database interactions
• Event Handlers: 80% coverage for event processing logic and error handling
• Domain Objects: 95% coverage including validation and state transitions
• API Layer: 70% coverage for request/response handling and middleware logic

Integration Testing Strategy

• Database Integration: Use Testcontainers for PostgreSQL/MySQL with schema migration fixtures to test actual database interactions
• External Service Mocking: WireMock for external API integrations (e.g., payment gateway, notification service)
• Event System: Embedded Kafka or RabbitMQ for testing event-driven flows and consumer behavior
• CQRS Read/Write Separation: Separate test suites for read models vs. command models to validate data consistency

Contract Testing

• Pact Framework: For API contract testing between services in a microservices architecture, ensuring that consumers and providers maintain compatible interfaces
• Versioned Endpoints: Contract tests for each API version to prevent breaking changes in backward compatibility
• Consumer-Driven Contracts: Define contracts from consumer perspective to ensure provider compliance

Performance Testing Tool and Key Scenarios

• Tool: k6 (JavaScript-based load testing) for realistic performance scenarios
• Key Scenarios:
  • High-concurrency command processing (1000 concurrent write operations)
  • Read model synchronization under load (500 concurrent read queries)
  • Event processing throughput (10,000 events/sec with retry handling)
  • End-to-end command-query flow latency (p95 < 200ms)

Test Data Management Strategy

• Fixtures: YAML/JSON fixtures for common test data sets (user profiles, orders, etc.)
• Factories: Factory classes using libraries like JavaFaker or custom builders to generate realistic test data
• Seeding: Database seed scripts with deterministic test data for consistent test runs
• Isolation: Use database transaction rollback per test method to maintain clean state between tests

// Example: Unit test for command handler with mock dependencies
@Test
public void testHandleCreateUserCommand_Success() {
    // Given
    CreateUserCommand command = new CreateUserCommand("john.doe@example.com", "John Doe");
    User user = new User("123", "john.doe@example.com", "John Doe");
    
    when(userRepository.save(any(User.class))).thenReturn(user);
    when(eventPublisher.publish(any(UserCreatedEvent.class))).thenReturn(CompletableFuture.completedFuture(null));
    
    // When
    User result = userCommandHandler.handle(command);
    
    // Then
    assertEquals("123", result.getId());
    verify(userRepository).save(any(User.class));
    verify(eventPublisher).publish(any(UserCreatedEvent.class));
}

13. Observability Design

Structured Logging Schema

The logging schema follows a structured format to enable efficient querying and analysis. All logs include the following required fields:

• timestamp: ISO 8601 formatted timestamp
• level: Log level (INFO, WARN, ERROR)
• trace_id: Distributed tracing identifier for correlation
• span_id: Span identifier within the trace
• service_name: Service name (e.g., "cqrs-framework")
• user_id: Authenticated user identifier (masked for PII)
• entity_id: Business entity ID (e.g., command ID, event ID)
• operation: Operation name (e.g., "handleCreateUserCommand")
• error_code: Error code when applicable
• message: Human-readable log message
• stack_trace: Full stack trace for ERROR level logs (optional in production)
• duration_ms: Operation duration in milliseconds

PII masking rules:

• User IDs are masked by truncating to first 4 characters and appending "..."
• Email addresses are masked as "user***@domain.com"
• Any field containing sensitive data is replaced with a placeholder

Distributed Tracing Instrumentation

We use OpenTelemetry for distributed tracing with the following custom spans:

1. Command Handling Span:

   • Name: cqrs.command.handle
   • Attributes:
     • command.type: Command class name (e.g., "CreateUserCommand")
     • command.id: Unique command identifier
     • command.duration_ms: Time spent processing the command

2. Event Publishing Span:

   • Name: cqrs.event.publish
   • Attributes:
     • event.type: Event class name (e.g., "UserCreatedEvent")
     • event.id: Unique event identifier
     • publisher.service: Service name that published the event

3. Query Handling Span:

   • Name: cqrs.query.handle
   • Attributes:
     • query.type: Query class name (e.g., "FindUserQuery")
     • query.duration_ms: Time spent processing the query

Metrics Design at Service Level

We implement RED (Rate, Errors, Duration) metrics for key operations:

Command Processing Metrics

Metric Name	Type	Description	Labels
cqrs_commands_processed_total	Counter	Total number of commands processed	command_type, status
cqrs_commands_duration_seconds	Histogram	Command processing duration	command_type
cqrs_commands_errors_total	Counter	Number of failed command processing attempts	command_type, error_code

Query Processing Metrics

Metric Name	Type	Description	Labels
cqrs_queries_processed_total	Counter	Total number of queries processed	query_type, status
cqrs_queries_duration_seconds	Histogram	Query processing duration	query_type
cqrs_queries_errors_total	Counter	Number of failed query processing attempts	query_type, error_code

Event Processing Metrics

Metric Name	Type	Description	Labels
cqrs_events_published_total	Counter	Total number of events published	event_type, status
cqrs_events_publish_duration_seconds	Histogram	Event publishing duration	event_type
cqrs_events_errors_total	Counter	Number of failed event publishing attempts	event_type, error_code

Health Check Endpoint Design

We implement three health check endpoints:

1. Readiness Check (/ready):

   • Checks database connectivity
   • Verifies message broker connection (if applicable)
   • Ensures cache service is available
   • Returns 200 OK when all dependencies are healthy

2. Liveness Check (/live):

   • Basic service status (memory usage, thread count)
   • Returns 200 OK when service is running

3. Startup Check (/startup):

   • Verifies that all required services have started
   • Confirms that the CQRS framework is initialized and ready to process commands

Alerting Thresholds per Key Metric

Metric	Warning Threshold	Critical Threshold	Alert Channel
cqrs_commands_errors_total	> 5 errors per minute	> 20 errors per minute	On-call
cqrs_commands_duration_seconds	p95 > 500ms	p99 > 2000ms	On-call
cqrs_queries_errors_total	> 5 errors per minute	> 20 errors per minute	On-call
cqrs_queries_duration_seconds	p95 > 300ms	p99 > 1500ms	On-call
cqrs_events_errors_total	> 3 errors per minute	> 15 errors per minute	On-call
cqrs_events_publish_duration_seconds	p95 > 200ms	p99 > 1000ms	On-call

Alerts are configured to escalate to on-call engineers after 5 minutes of sustained issues. Critical alerts trigger immediate page and SMS notifications to the engineering team.

graph TD
    A[Command Processing] --> B[OpenTelemetry Span]
    A --> C[Metrics Counter/Histogram]
    D[Query Processing] --> E[OpenTelemetry Span]
    D --> F[Metrics Counter/Histogram]
    G[Event Publishing] --> H[OpenTelemetry Span]
    G --> I[Metrics Counter/Histogram]
    J[Health Checks] --> K[/ready, /live, /startup]

14. Open Questions, Risks & Trade-offs

Decisions requiring product / engineering alignment

• Event versioning strategy: Should we use schema evolution with Avro/Protobuf or semantic versioning of events? This decision impacts backward compatibility and consumer complexity.
• CQRS vs. monolithic approach: Is the added complexity of CQRS justified for this system's scale and growth trajectory? The team needs to align on whether eventual consistency is acceptable.
• Command validation strategy: Should validation happen at the command level or at the event level? This affects error handling and user feedback mechanisms.
• Event sourcing vs. read model projection: Should we implement event sourcing for full auditability or rely on read model projections for performance?

Known trade-offs and justifications

Trade-off	Justification
Eventual consistency vs. strong consistency	CQRS introduces eventual consistency between command and query sides, which is acceptable for most use cases but requires careful design of read models to avoid stale data
Complexity vs. performance	CQRS adds significant architectural complexity but enables better performance through separate read/write models and horizontal scaling
Storage overhead vs. query flexibility	Maintaining separate read models increases storage requirements but enables highly optimized queries for different use cases
Event-driven vs. synchronous processing	Async processing provides better scalability but introduces complexity in error handling and debugging
Separation of concerns vs. operational overhead	Clear separation between command and query models improves maintainability but increases operational complexity with multiple data stores

Risks and mitigation strategies

Risk	Impact	Mitigation Strategy
Event loss in async processing	Data integrity failure	Implement outbox pattern with database transactional consistency and DLQ handling
Read model staleness	User experience degradation	Implement cache invalidation strategies and TTL-based refresh mechanisms
Complexity in distributed transactions	System instability	Use saga pattern for cross-service operations and implement circuit breaker patterns
Event schema evolution issues	Breaking changes to consumers	Implement comprehensive versioning strategy with backward compatibility guarantees
Increased operational overhead	Higher maintenance burden	Invest in automation tools for monitoring, alerting, and deployment pipelines
Testing complexity	Higher test maintenance cost	Implement comprehensive contract testing and use test containers for integration scenarios

Technical debt acknowledged

• Event schema evolution complexity: Future versions of events may require complex migration logic, increasing technical debt in the event handling layer.
• Read model synchronization lag: The eventual consistency between command and query models creates potential for data inconsistency during synchronization windows.
• Cross-service CQRS boundaries: When multiple services use CQRS, managing the distributed state and ensuring consistency across service boundaries introduces significant complexity.
• Event replay complexity: The ability to replay events for debugging or reprocessing adds operational overhead and requires careful design of event storage and processing.
• Monitoring and debugging complexity: Debugging issues in a distributed CQRS system requires sophisticated tracing and audit logging capabilities that add to the overall system complexity.

The framework design acknowledges these trade-offs and risks as inherent to implementing CQRS at scale, with mitigation strategies built into the architecture to reduce their impact on system reliability and maintainability.