Observer-Based Event System Design

This document presents a robust observer-based event system implementation that supports decoupled communication between components through publish-subscribe patterns, featuring thread-safe operations and flexible event handling mechanisms.

1. Problem Statement & Requirements

Clarifying questions

1. What is the scale of events expected to be handled? Are we looking at hundreds, thousands, or millions of events per second?
2. What consistency requirements exist for event delivery? Should events be guaranteed to be delivered exactly once, at-least-once, or is eventual consistency acceptable?
3. What are the key entities in this system and their relationships? Is it a simple event publishing-subscription model or does it involve complex event streams with transformations and routing?
4. What is the read/write ratio for events? Are most operations focused on publishing or subscribing to events?
5. What latency sensitivity is expected for event delivery? Is sub-millisecond delivery required or can it be in the range of milliseconds to seconds?
6. What authentication and authorization model is expected? Should events be accessible only by authenticated users or can they be publicly accessible?
7. Are there any external integrations or third-party systems that will consume events? Do we need to support webhooks, Kafka, or other messaging systems?

Functional requirements

• Event publishing: Allow clients to publish events with a unique identifier, timestamp, and payload.
• Event subscription: Enable clients to subscribe to events based on topic or type filters.
• Event filtering: Support filtering of events by topic, type, timestamp range, and custom attributes.
• Event persistence: Store events for a configurable period or until explicitly deleted.
• Event retrieval: Allow clients to retrieve past events by ID, topic, or timestamp range.
• Event deduplication: Prevent duplicate event processing based on unique identifiers.

Non-functional requirements

• Latency targets:
  • p50: < 10ms for event publishing and subscription
  • p95: < 50ms for event publishing and subscription
  • p99: < 200ms for event publishing and subscription
• Throughput:
  • 10,000 events/sec for publishing
  • 5,000 events/sec for subscription
• Availability: 99.9% uptime with graceful degradation during outages

Constraints and assumptions

• Assumption: Events are not expected to exceed 1MB in size.
• Constraint: System must support horizontal scaling across multiple instances.
• Assumption: Event consumers are expected to be available and responsive within 5 seconds of event publication.
• Constraint: Event storage is limited to 10TB total across all instances.
• Assumption: The system will be used within a single data center or region.

Out of scope

• Event transformation: The system does not include support for transforming or enriching events before delivery.
• Event routing: Complex routing logic (e.g., conditional routing based on event content) is not part of this design.
• Event versioning: Versioning of events or backward compatibility for event schemas is not included.
• Cross-region replication: Multi-region support and replication are out of scope.
• Event compression: Compression of events before storage or transmission is not part of this design.

2. API Design

API Style Selection

For an observer-based event system, 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 systems where schema evolution matters
2. Efficient binary serialization: Better performance characteristics than JSON for high-throughput event processing
3. Built-in support for streaming: Critical for real-time event delivery and observer patterns
4. First-class support for async operations: Enables efficient pub/sub with bidirectional streams
5. Service-level observability: gRPC provides built-in metrics and tracing capabilities

Endpoint Definitions

Method	Path	Description	Auth
PublishEvent	/event/v1/publish	Publish a new event to the system	JWT
SubscribeToEvents	/event/v1/subscribe	Establish a bidirectional stream for event subscription	JWT
GetEventTypes	/event/v1/types	Retrieve list of supported event types	JWT
GetEventSubscriptions	/event/v1/subscriptions	Get user's current event subscriptions	JWT
CreateSubscription	/event/v1/subscriptions	Create new event subscription	JWT
DeleteSubscription	/event/v1/subscriptions/{id}	Delete an existing subscription	JWT

Request and Response Schemas

PublishEvent Request:

message PublishEventRequest {
  string event_type = 1;
  string event_id = 2;
  bytes payload = 3;
  google.protobuf.Timestamp timestamp = 4;
  map<string, string> headers = 5;
}

message PublishEventResponse {
  string event_id = 1;
  bool success = 2;
  string message = 3;
}

SubscribeToEvents Request:

message SubscribeToEventsRequest {
  repeated string event_types = 1;
  string subscription_id = 2;
}

message EventStream {
  string event_id = 1;
  string event_type = 2;
  bytes payload = 3;
  google.protobuf.Timestamp timestamp = 4;
  map<string, string> headers = 5;
}

GetEventTypes Response:

message GetEventTypesResponse {
  repeated EventType event_types = 1;
}

message EventType {
  string name = 1;
  string description = 2;
  google.protobuf.Timestamp created_at = 3;
}

CreateSubscription Request:

message CreateSubscriptionRequest {
  string event_type = 1;
  string callback_url = 2;
  string subscription_name = 3;
  map<string, string> filters = 4;
}

message CreateSubscriptionResponse {
  string subscription_id = 1;
  bool success = 2;
  string message = 3;
}

HTTP Status Codes and Error Response Format

Status Code	Description
200	Success
400	Bad Request - Invalid input
401	Unauthorized - Missing or invalid auth
403	Forbidden - Insufficient permissions
429	Too Many Requests - Rate limited
500	Internal Server Error

Standardized error response:

{
  "error": {
    "code": "INVALID_ARGUMENT",
    "message": "Event type is required",
    "details": [
      {
        "field": "event_type",
        "message": "Event type is required"
      }
    ]
  }
}

API Versioning Strategy

Versioning will be handled through the path prefix:

• /event/v1/ - Current stable version
• /event/v2/ - Next major version with breaking changes

Pagination Design

For list operations, we'll use cursor-based pagination:

message ListRequest {
  int32 page_size = 1;
  string page_token = 2;
}

message ListResponse {
  repeated Event items = 1;
  string next_page_token = 2;
}

Justification:

• Cursor-based pagination is more stable and handles data changes better than offset-based
• Provides consistent results even when items are inserted/deleted between queries
• Better performance for large datasets

Authentication and Authorization

All endpoints require JWT token with the following scopes:

• event:publish - For publishing events
• event:subscribe - For subscribing to events
• event:manage - For managing subscriptions

Authentication middleware will:

1. Validate JWT token signature
2. Check token expiration
3. Verify required scopes for endpoint access
4. Extract user ID from claims for audit logging

Rate Limiting Strategy

Rate limiting will be implemented at the service level with:

• Per-user rate limits: 1000 requests/minute for publishing events
• Per-subscription rate limits: 5000 requests/minute for subscription endpoints
• Token bucket algorithm with burst capacity of 100 requests

Rate limit enforcement will:

1. Track per-user and per-subscription counters
2. Return 429 status with Retry-After header when limits exceeded
3. Log rate limit violations for monitoring

Idempotency Design

For mutation endpoints, we'll implement idempotency key strategy:

message IdempotencyRequest {
  string idempotency_key = 1;
  oneof request {
    PublishEventRequest publish_request = 2;
    CreateSubscriptionRequest create_subscription_request = 3;
  }
}

Implementation approach:

1. Store idempotency keys in Redis with TTL of 24 hours
2. For the same key, return the same response as the first request
3. Use a distributed lock to handle concurrent requests with same key
4. Log idempotency key usage for debugging and audit purposes

API Documentation

API documentation will be generated using:

1. OpenAPI/Swagger for REST endpoints (for client SDKs)
2. gRPC documentation using gRPC Gateway for HTTP/JSON compatibility
3. Interactive API console with example requests and responses
4. Versioned documentation with changelog for breaking changes

Sample API Usage

Publishing an event:

curl -X POST "https://api.example.com/event/v1/publish" \
  -H "Authorization: Bearer <JWT_TOKEN>" \
  -H "Content-Type: application/json" \
  -d '{
    "event_type": "user.created",
    "event_id": "evt_1234567890",
    "payload": "eyJ1c2VyX2lkIjogIjEyMzQ1Njc4OTAiLCAibmFtZSI6ICJBY2VzIiwgImVtYWlsIjogImFjZXNAZWN4YW1wbGUuY29tIn0=",
    "timestamp": "2023-12-01T10:00:00Z"
  }'

Subscribing to events:

curl -X POST "https://api.example.com/event/v1/subscribe" \
  -H "Authorization: Bearer <JWT_TOKEN>" \
  -H "Content-Type: application/json" \
  -d '{
    "event_types": ["user.created", "user.updated"],
    "subscription_id": "sub_1234567890"
  }'

3. Data Model Design

Entity-Relationship diagram (Mermaid)

erDiagram
    EVENT {
        string id PK
        string name
        string description
        string eventType
        timestamp createdAt
        timestamp updatedAt
        boolean isActive
    }

    EVENT_SUBSCRIPTION {
        string id PK
        string eventId FK
        string subscriberId
        string callbackUrl
        string secret
        timestamp createdAt
        timestamp updatedAt
        string status
    }

    EVENT_PAYLOAD {
        string id PK
        string eventId FK
        json payload
        timestamp createdAt
    }

    EVENT_DELIVERY {
        string id PK
        string subscriptionId FK
        string eventId FK
        string deliveryStatus
        timestamp createdAt
        timestamp updatedAt
        string errorMessage
    }

    EVENT {
        string id
        string name
        string description
        string eventType
        timestamp createdAt
        timestamp updatedAt
        boolean isActive
    }

    EVENT_SUBSCRIPTION {
        string id
        string eventId
        string subscriberId
        string callbackUrl
        string secret
        timestamp createdAt
        timestamp updatedAt
        string status
    }

    EVENT_PAYLOAD {
        string id
        string eventId
        json payload
        timestamp createdAt
    }

    EVENT_DELIVERY {
        string id
        string subscriptionId
        string eventId
        string deliveryStatus
        timestamp createdAt
        timestamp updatedAt
        string errorMessage
    }

    EVENT ||--o{ EVENT_SUBSCRIPTION : has
    EVENT ||--o{ EVENT_PAYLOAD : has
    EVENT_SUBSCRIPTION ||--o{ EVENT_DELIVERY : triggers

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

-- Events table to store event metadata
CREATE TABLE events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    description TEXT,
    event_type VARCHAR(100) NOT NULL,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

-- Indexes for events table
CREATE INDEX idx_events_type ON events(event_type);
CREATE INDEX idx_events_active ON events(is_active);

-- Event subscriptions table to store subscription details
CREATE TABLE event_subscriptions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    event_id UUID NOT NULL REFERENCES events(id) ON DELETE CASCADE,
    subscriber_id VARCHAR(255) NOT NULL,
    callback_url TEXT NOT NULL,
    secret VARCHAR(255),
    status VARCHAR(50) DEFAULT 'ACTIVE',
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

-- Indexes for event_subscriptions table
CREATE INDEX idx_event_subscriptions_event_id ON event_subscriptions(event_id);
CREATE INDEX idx_event_subscriptions_status ON event_subscriptions(status);
CREATE INDEX idx_event_subscriptions_subscriber_id ON event_subscriptions(subscriber_id);

-- Event payloads table to store event payload data
CREATE TABLE event_payloads (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    event_id UUID NOT NULL REFERENCES events(id) ON DELETE CASCADE,
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

-- Indexes for event_payloads table
CREATE INDEX idx_event_payloads_event_id ON event_payloads(event_id);

-- Event deliveries table to track delivery status
CREATE TABLE event_deliveries (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    subscription_id UUID NOT NULL REFERENCES event_subscriptions(id) ON DELETE CASCADE,
    event_id UUID NOT NULL REFERENCES events(id) ON DELETE CASCADE,
    delivery_status VARCHAR(50) DEFAULT 'PENDING',
    error_message TEXT,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

-- Indexes for event_deliveries table
CREATE INDEX idx_event_deliveries_subscription_id ON event_deliveries(subscription_id);
CREATE INDEX idx_event_deliveries_event_id ON event_deliveries(event_id);
CREATE INDEX idx_event_deliveries_status ON event_deliveries(delivery_status);

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

Using UUIDs as primary keys for all tables due to:

1. Distributed system compatibility: UUIDs ensure uniqueness across distributed systems without coordination
2. Scalability: No need to coordinate with a central sequence generator in distributed environments
3. No hotspots: Unlike auto-increment, UUIDs don't create write hotspots in sharded systems
4. Security: UUIDs don't reveal sequential information that could be exploited

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

1. Events table:

   • idx_events_type: For filtering by event type (common query pattern)
   • idx_events_active: For querying only active events (performance critical)

2. Event subscriptions table:

   • idx_event_subscriptions_event_id: For querying all subscriptions for a specific event
   • idx_event_subscriptions_status: For querying only active subscriptions (performance critical)
   • idx_event_subscriptions_subscriber_id: For querying all subscriptions for a specific subscriber

3. Event payloads table:

   • idx_event_payloads_event_id: For retrieving payload for a specific event

4. Event deliveries table:

   • idx_event_deliveries_subscription_id: For querying delivery history for a specific subscription
   • idx_event_deliveries_event_id: For querying all deliveries for a specific event
   • idx_event_deliveries_status: For querying pending or failed deliveries (important for retry logic)

Data partitioning / sharding strategy

Partition by event_type and created_at:

1. Event type partitioning: Events can be sharded by event type (e.g., user.created, order.completed) to optimize queries for specific event types
2. Time-based partitioning: Events can be partitioned by creation date (monthly or daily) to support time-range queries and data retention policies

For PostgreSQL, this can be implemented using partitioned tables:

CREATE TABLE events_partitioned (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    description TEXT,
    event_type VARCHAR(100) NOT NULL,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
) PARTITION BY LIST (event_type);

CREATE TABLE events_user_partitioned PARTITION OF events_partitioned FOR VALUES IN ('user.created', 'user.updated');
CREATE TABLE events_order_partitioned PARTITION OF events_partitioned FOR VALUES IN ('order.created', 'order.completed');

Soft delete vs. hard delete decision with justification

Soft delete is used for all tables to:

1. Audit trail preservation: Maintain historical information about subscriptions and deliveries for compliance
2. Recovery capability: Allow restoration of deleted subscriptions or events if needed
3. Data integrity: Prevent orphaned records in related tables (e.g., event_deliveries referencing deleted events)
4. Analytics: Support analytics on subscription and delivery patterns over time

Audit trail design (who, what, when)

Audit logs are maintained through:

1. Timestamps: All tables include created_at and updated_at fields
2. Event tracking: Events are versioned to track changes over time
3. Status tracking: Delivery status changes are logged in event_deliveries table

Data archival strategy and retention policy

1. Event payload archiving: Payloads older than 30 days are archived to cold storage (S3 or similar)
2. Delivery history retention: Delivery records older than 90 days are archived
3. Subscription history: Active subscriptions are retained indefinitely, inactive ones after 6 months
4. Event metadata retention: Event definitions are retained indefinitely for historical reference

Archival process:

-- Archive old event payloads to external storage (e.g., S3)
INSERT INTO archived_event_payloads (
    event_id,
    payload,
    archived_at
) SELECT 
    event_id,
    payload,
    CURRENT_TIMESTAMP
FROM event_payloads 
WHERE created_at < CURRENT_DATE - INTERVAL '30 days';

-- Delete archived payloads from main table
DELETE FROM event_payloads 
WHERE created_at < CURRENT_DATE - INTERVAL '30 days';

4. Class Design & Service Layer

UML Class Diagram

classDiagram
    class EventSystem {
        +publishEvent(event: Event)
        +subscribe(observer: Observer, eventType: String)
        +unsubscribe(observer: Observer, eventType: String)
    }
    
    class Event {
        -id: UUID
        -type: String
        -payload: Map~String, Object~
        -timestamp: Instant
        -source: String
        +getId(): UUID
        +getType(): String
        +getPayload(): Map~String, Object~
        +getTimestamp(): Instant
        +getSource(): String
    }
    
    class Observer {
        +handleEvent(event: Event)
    }
    
    class EventPublisher {
        -observers: Map~String, List~Observer~~
        +publish(event: Event)
        +subscribe(observer: Observer, eventType: String)
        +unsubscribe(observer: Observer, eventType: String)
    }
    
    class EventConsumer {
        -eventPublisher: EventPublisher
        +handleEvent(event: Event)
    }
    
    class EventProcessor {
        -eventPublisher: EventPublisher
        +processEvent(event: Event)
    }
    
    class AsyncEventProcessor {
        -eventPublisher: EventPublisher
        -executorService: ExecutorService
        +processEvent(event: Event)
    }
    
    class EventStore {
        -events: List~Event~
        +save(event: Event)
        +findAll(): List~Event~
    }
    
    class EventFilter {
        +shouldProcess(event: Event): boolean
    }
    
    EventSystem --> EventPublisher : uses
    EventSystem --> EventStore : persists
    EventPublisher --> Observer : notifies
    EventConsumer --> EventPublisher : subscribes
    EventProcessor --> EventPublisher : subscribes
    AsyncEventProcessor --> EventPublisher : subscribes
    EventStore --> Event : persists
    EventFilter --> Event : filters
    
    class ObserverAdapter {
        -observer: Observer
        +handleEvent(event: Event)
    }
    
    ObserverAdapter --> Observer : wraps

State Machine Diagram for Event Processing

stateDiagram-v2
    [*] --> Ready
    Ready --> Processing : EventReceived
    Processing --> Success : Processed
    Processing --> Failed : Error
    Success --> [*]
    Failed --> Ready : RetryableError
    Failed --> [*] : NonRetryableError

Service Class Structure and Responsibilities

The event system is composed of several key service classes:

1. EventSystem: Main orchestrator that manages the lifecycle of events and observers
2. EventPublisher: Core component that handles observer registration and event dispatching
3. EventConsumer: Interface for concrete event handlers to implement
4. EventProcessor: Synchronous processor that handles events in the same thread context
5. AsyncEventProcessor: Asynchronous processor that offloads event handling to a thread pool
6. EventStore: Persistent storage for events (optional but recommended for audit/tracking)
7. EventFilter: Optional component to filter events before processing

Interface / Abstract Class Design

// Observer interface - defines contract for event consumers
public interface Observer {
    void handleEvent(Event event);
}

// EventConsumer - concrete implementation of observer
public abstract class EventConsumer implements Observer {
    protected final String eventType;
    
    public EventConsumer(String eventType) {
        this.eventType = eventType;
    }
    
    @Override
    public void handleEvent(Event event) {
        if (event.getType().equals(eventType)) {
            processEvent(event);
        }
    }
    
    protected abstract void processEvent(Event event);
}

// EventFilter interface - optional filtering capability
public interface EventFilter {
    boolean shouldProcess(Event event);
}

Business Logic Encapsulation Patterns

We implement the Strategy pattern for event processing to allow flexible handling of different event types:

// Strategy interface for event processing
public interface EventProcessingStrategy {
    void process(Event event);
}

// Concrete strategy implementations
public class UserEventProcessingStrategy implements EventProcessingStrategy {
    @Override
    public void process(Event event) {
        // Handle user-related events (e.g., user created, updated)
        switch (event.getType()) {
            case "USER_CREATED":
                handleUserCreated(event);
                break;
            case "USER_UPDATED":
                handleUserUpdated(event);
                break;
            default:
                throw new IllegalArgumentException("Unsupported user event type: " + event.getType());
        }
    }
    
    private void handleUserCreated(Event event) {
        // Implementation for user creation handling
        User user = deserializeUser(event.getPayload());
        userRepository.save(user);
    }
    
    private void handleUserUpdated(Event event) {
        // Implementation for user update handling
        User user = deserializeUser(event.getPayload());
        userRepository.update(user);
    }
    
    private User deserializeUser(Map<String, Object> payload) {
        // Implementation for deserializing user from event payload
        return new User();
    }
}

// Strategy context - selects appropriate strategy based on event type
public class EventProcessingContext {
    private final Map<String, EventProcessingStrategy> strategies;
    
    public EventProcessingContext() {
        this.strategies = new HashMap<>();
        this.strategies.put("USER_CREATED", new UserEventProcessingStrategy());
        this.strategies.put("USER_UPDATED", new UserEventProcessingStrategy());
        // Add more strategies as needed
    }
    
    public void process(Event event) {
        EventProcessingStrategy strategy = strategies.get(event.getType());
        if (strategy != null) {
            strategy.process(event);
        } else {
            throw new IllegalArgumentException("No processing strategy found for event type: " + event.getType());
        }
    }
}

Critical-Path Method Implementation

public class EventSystem {
    private final EventPublisher eventPublisher;
    private final EventStore eventStore;
    private final EventProcessingContext processingContext;
    private final ExecutorService asyncExecutor;
    
    public EventSystem() {
        this.eventPublisher = new EventPublisher();
        this.eventStore = new EventStore();
        this.processingContext = new EventProcessingContext();
        this.asyncExecutor = Executors.newFixedThreadPool(10);
    }
    
    /**
     * Critical path for event publishing - handles transaction boundary,
     * domain object mutations, and event publishing in a single atomic operation
     */
    public void publishEvent(Event event) {
        // Transaction boundary - all operations within this method should be atomic
        try {
            // 1. Validate event before processing
            validateEvent(event);
            
            // 2. Store event in persistent store (optional but recommended)
            eventStore.save(event);
            
            // 3. Process synchronous handlers in same thread context
            eventPublisher.publish(event);
            
            // 4. Queue asynchronous processing if needed
            asyncExecutor.submit(() -> {
                try {
                    // Apply event processing strategy
                    processingContext.process(event);
                } catch (Exception e) {
                    // Log and handle async processing errors
                    log.error("Async event processing failed for event: {}", event.getId(), e);
                }
            });
            
        } catch (Exception e) {
            // Handle transaction rollback or error propagation
            log.error("Failed to publish event: {}", event.getId(), e);
            throw new EventProcessingException("Event publishing failed", e);
        }
    }
    
    private void validateEvent(Event event) {
        if (event == null) {
            throw new IllegalArgumentException("Event cannot be null");
        }
        if (event.getType() == null || event.getType().isEmpty()) {
            throw new IllegalArgumentException("Event type cannot be null or empty");
        }
        if (event.getId() == null) {
            throw new IllegalArgumentException("Event ID cannot be null");
        }
    }
}

Transaction Boundary Design

The transaction boundary for event publishing includes:

1. Event validation
2. Persistent storage of event (if enabled)
3. Synchronous observer notifications
4. Asynchronous processing queuing

All operations within the publishEvent method form a single logical transaction. For systems requiring strict ACID guarantees, we would wrap this in database transactions or use distributed transaction protocols.

Service Interface Contracts

public interface EventSystemService {
    void publishEvent(Event event);
    void subscribe(Observer observer, String eventType);
    void unsubscribe(Observer observer, String eventType);
    List<Event> getEventHistory(String eventType, int limit);
}

Saga / Orchestration Pattern

For distributed event systems, we would implement a saga pattern to coordinate events across microservices:

public class EventSaga {
    private final List<Event> steps;
    private final String sagaId;
    
    public EventSaga(String sagaId) {
        this.sagaId = sagaId;
        this.steps = new ArrayList<>();
    }
    
    public void addStep(Event event) {
        steps.add(event);
    }
    
    public void execute() {
        // Execute all steps in order, with compensation logic for failures
        for (Event step : steps) {
            try {
                eventSystem.publishEvent(step);
            } catch (Exception e) {
                // Implement compensation logic or rollback
                handleCompensation();
                throw new SagaExecutionException("Saga failed at step: " + step.getType(), e);
            }
        }
    }
    
    private void handleCompensation() {
        // Implement compensation logic for failed steps
        // This could involve publishing rollback events or calling compensating services
    }
}

Domain Object / Value Object Design

// Event value object - immutable and thread-safe
public class Event {
    private final UUID id;
    private final String type;
    private final Map<String, Object> payload;
    private final Instant timestamp;
    private final String source;
    
    public Event(UUID id, String type, Map<String, Object> payload, String source) {
        this.id = Objects.requireNonNull(id);
        this.type = Objects.requireNonNull(type);
        this.payload = payload != null ? new HashMap<>(payload) : Collections.emptyMap();
        this.timestamp = Instant.now();
        this.source = source;
    }
    
    // Getters with defensive copying
    public UUID getId() { return id; }
    public String getType() { return type; }
    public Map<String, Object> getPayload() { 
        return new HashMap<>(payload); 
    }
    public Instant getTimestamp() { return timestamp; }
    public String getSource() { return source; }
    
    // Builder pattern for construction
    public static class Builder {
        private UUID id;
        private String type;
        private Map<String, Object> payload = new HashMap<>();
        private String source;
        
        public Builder withId(UUID id) {
            this.id = id;
            return this;
        }
        
        public Builder withType(String type) {
            this.type = type;
            return this;
        }
        
        public Builder withPayload(Map<String, Object> payload) {
            this.payload = payload != null ? new HashMap<>(payload) : Collections.emptyMap();
            return this;
        }
        
        public Builder withSource(String source) {
            this.source = source;
            return this;
        }
        
        public Event build() {
            return new Event(id, type, payload, source);
        }
    }
}

Key Design Considerations

1. Thread Safety: The event system is designed to be thread-safe with proper synchronization in observer registration
2. Scalability: Supports both synchronous and asynchronous processing paths
3. Extensibility: Strategy pattern allows easy addition of new event types and processing logic
4. Auditability: Optional persistent storage enables event tracking and debugging
5. Error Handling: Comprehensive error handling with fallback strategies for async processing

5. Concurrency & Thread Safety

Concurrency Model

The event system will use a thread pool-based concurrency model with the following characteristics:

• Fixed-size thread pool for event processing (typically 10-50 threads depending on system load)
• Each event handler runs in its own thread from the pool
• Thread-local state management for per-request context (trace IDs, user contexts)
• Async/Await pattern for I/O-bound operations with proper thread reuse

Race Condition Prevention Strategy

Race conditions are mitigated through:

1. Immutable event objects - Events are built once and never modified after publication
2. Thread-safe observer registration - ConcurrentHashMap for observer management with read-write locks
3. Atomic state updates - Use of atomic reference types for internal state tracking
4. Synchronized access to shared resources - Critical sections protected with mutexes where necessary

Optimistic vs. Pessimistic Locking Selection

For the observer system, optimistic locking is selected because:

• Observer registration/deregistration is a low-frequency operation
• Events are published asynchronously and don't require shared resource locks during processing
• The system prioritizes throughput over strict consistency in observer management
• Event handlers can process concurrently without interfering with each other

Deadlock Prevention Strategy

Deadlock prevention is achieved through:

1. Fixed ordering of resource acquisition - All operations follow a consistent lock ordering (e.g., always acquire observer registry lock before event processing lock)
2. Timeout-based acquisition - Use of timed locks to prevent indefinite blocking
3. Hierarchical lock management - Clear lock hierarchy (registry → event → handler)
4. Avoid nested locking - Minimize nested lock acquisition patterns

Idempotency Implementation for Concurrent Duplicate Requests

Idempotency is handled at the observer registration level:

public class EventObserverRegistry {
    private final Map<String, Set<Observer>> observers = new ConcurrentHashMap<>();
    private final ReadWriteLock registryLock = new ReentrantReadWriteLock();
    
    public boolean registerObserver(String eventType, Observer observer) {
        // Use CAS for idempotent registration
        return observers.computeIfAbsent(eventType, k -> Collections.newSetFromMap(new ConcurrentHashMap<>()))
                       .add(observer);
    }
    
    public boolean deregisterObserver(String eventType, Observer observer) {
        Set<Observer> eventObservers = observers.get(eventType);
        if (eventObservers != null) {
            return eventObservers.remove(observer);
        }
        return false;
    }
}

The system ensures that duplicate observer registrations are idempotent by using a Set data structure with proper equality checks and atomic operations to prevent race conditions during concurrent registration attempts.

6. Sequence Diagrams for Key Flows

sequenceDiagram
    participant Client
    participant EventPublisher
    participant EventStore
    participant EventSubscriber
    participant SubscriberService

    Client->>EventPublisher: Publish Event (e.g., OrderCreated)
    EventPublisher->>EventStore: Store Event
    EventPublisher->>EventSubscriber: Notify Subscribers
    EventSubscriber->>SubscriberService: Handle Event
    SubscriberService-->>EventSubscriber: Acknowledge
    EventSubscriber-->>EventPublisher: Confirmation
    EventPublisher-->>Client: Success Response

sequenceDiagram
    participant Client
    participant EventPublisher
    participant EventStore
    participant EventSubscriber
    participant SubscriberService

    Client->>EventPublisher: Publish Event (e.g., OrderCreated)
    EventPublisher->>EventStore: Store Event
    EventPublisher->>EventSubscriber: Notify Subscribers
    EventSubscriber->>SubscriberService: Handle Event (fails)
    SubscriberService-->>EventSubscriber: Error Response
    EventSubscriber->>EventPublisher: Retry Request
    EventPublisher->>EventStore: Mark as Failed
    EventPublisher-->>Client: Error Response (after retries exhausted)

sequenceDiagram
    participant Client
    participant EventPublisher
    participant EventStore
    participant EventSubscriber
    participant SubscriberService

    Client->>EventPublisher: Publish Event (e.g., OrderCreated)
    EventPublisher->>EventStore: Store Event
    EventPublisher->>EventSubscriber: Notify Subscribers (async)
    EventSubscriber->>SubscriberService: Handle Event (async)
    SubscriberService->>EventStore: Log Processing Result
    SubscriberService-->>EventSubscriber: Acknowledge
    EventSubscriber->>EventPublisher: Confirm Processing
    EventPublisher-->>Client: Success Response (after async completion)

7. Caching Design

Cache Strategy Selection

For the observer-based event system, we'll use a cache-aside pattern with LRU eviction policy. This approach provides the best balance between performance and consistency for our use case where:

• Events are published frequently but not all need to be cached
• Read-heavy workloads with predictable access patterns
• Need for eventual consistency over strong consistency

Cache Key Design

Cache keys will follow a structured namespace pattern:

event-system:{entityType}:{entityId}:{eventType}:{version}

Example key for a user creation event:

event-system:user:12345:created:v1

TTL Strategy

Data Type	TTL	Reason
Event subscriptions	24h	Subscriptions are relatively stable
Event metadata	1h	Metadata changes infrequently
Recent events (last N)	30m	High read frequency for recent events
Event delivery logs	7d	Audit and debugging needs

Cache Eviction Policy

LRU (Least Recently Used) with time-based expiration:

• Primary eviction based on access time
• Secondary cleanup for expired entries via background task
• Cache size limited to 1GB with automatic eviction

Cache Stampede Prevention

Implement probabilistic early expiry:

• Set cache TTL to 2x the configured timeout
• Use distributed locks for cache population
• Implement cache warming strategy for hot keys

Distributed Cache Architecture

Using Redis Cluster with:

• 3 master nodes, 3 replica nodes
• Automatic failover and sharding
• Redis Streams for event persistence (not just caching)
• Memory policy: allkeys-lru with maxmemory 1GB per node

Cache Invalidation Strategy

Event-driven invalidation:

• When an event is published, invalidate related cache entries
• Use Redis Pub/Sub to broadcast invalidation messages
• Implement batch invalidation for bulk operations

@Component
public class EventCacheInvalidator {
    private final RedisTemplate<String, Object> redisTemplate;
    
    public void invalidateEventCache(String eventType, String entityId) {
        // Invalidate specific event cache entries
        String pattern = "event-system:*:" + entityId + ":" + eventType + ":*";
        Set<String> keys = redisTemplate.keys(pattern);
        
        if (!keys.isEmpty()) {
            redisTemplate.delete(keys);
        }
        
        // Broadcast invalidation to other services
        redisTemplate.convertAndSend("cache-invalidation", 
            new CacheInvalidationEvent(eventType, entityId));
    }
}

Cache Read/Write Path

Read path:

1. Check cache for event data
2. If miss, fetch from database and populate cache
3. Apply TTL and LRU eviction policies

Write path:

1. Update database record
2. Invalidate relevant cache entries
3. Publish to event stream for async processing

Cache Warmup Strategy

Implement background warming for frequently accessed event types:

@Scheduled(fixedRate = 300000) // Every 5 minutes
public void warmEventCache() {
    List<String> hotEventTypes = getHotEventTypes();
    for (String eventType : hotEventTypes) {
        // Pre-populate cache with recent events
        List<Event> recentEvents = eventRepository.findRecentByType(eventType, 100);
        for (Event event : recentEvents) {
            cacheEvent(event);
        }
    }
}

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 entity has its own repository interface with standardized CRUD operations.

public interface EventRepository {
    Event save(Event event);
    Optional<Event> findById(UUID id);
    List<Event> findAll();
    List<Event> findByType(String type);
    void delete(UUID id);
    long count();
}

public interface EventSubscriptionRepository {
    EventSubscription save(EventSubscription subscription);
    Optional<EventSubscription> findByUserIdAndEventTypeId(UUID userId, UUID eventTypeId);
    List<EventSubscription> findByUserId(UUID userId);
    List<EventSubscription> findByEventTypeId(UUID eventTypeId);
    void deleteByUserIdAndEventTypeId(UUID userId, UUID eventTypeId);
}

ORM vs. Raw Query Decisions

For the event system, we use JPA/Hibernate for most operations due to its maturity and ability to handle complex object graphs. However, raw SQL is used for high-performance read-heavy operations like event statistics aggregation.

// JPA/Hibernate for complex domain operations
@PersistenceContext
private EntityManager entityManager;

// Raw SQL for performance-critical analytics queries
@Query(value = "SELECT event_type_id, COUNT(*) as count FROM events WHERE created_at >= :startTime GROUP BY event_type_id", nativeQuery = true)
List<EventCountProjection> getEventCountsByType(@Param("startTime") LocalDateTime startTime);

Read vs. Write Separation

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

// Write model - used for event creation and subscription management
public class EventWriteRepository {
    public Event save(Event event) {
        return entityManager.merge(event);
    }
    
    public void delete(UUID id) {
        Event event = entityManager.find(Event.class, id);
        if (event != null) {
            entityManager.remove(event);
        }
    }
}

// Read model - optimized for querying and analytics
public class EventReadRepository {
    @Query("SELECT new com.example.event.dto.EventSummary(e.id, e.type, e.timestamp) FROM Event e WHERE e.userId = :userId")
    List<EventSummary> findUserEvents(@Param("userId") UUID userId);
}

Connection Pool Configuration

Connection pool is configured with HikariCP for optimal performance:

spring:
  datasource:
    hikari:
      maximum-pool-size: 20
      minimum-idle: 5
      connection-timeout: 30000
      idle-timeout: 600000
      max-lifetime: 1800000
      validation-timeout: 5000

N+1 Query Prevention Strategy

We use eager loading and batch fetching to prevent N+1 queries:

@Entity
public class Event {
    @Id
    private UUID id;
    
    @ManyToOne(fetch = FetchType.EAGER)
    @JoinColumn(name = "user_id")
    private User user;
    
    @OneToMany(mappedBy = "event", fetch = FetchType.LAZY, cascade = CascadeType.ALL)
    @BatchSize(size = 20)
    private List<EventSubscription> subscriptions = new ArrayList<>();
    
    // ... other fields
}

Batch Operation Design

Batch operations are implemented for bulk event creation and subscription management:

public class BatchEventRepository {
    public void saveBatch(List<Event> events) {
        String sql = "INSERT INTO events (id, type, payload, user_id, timestamp) VALUES (?, ?, ?, ?, ?)";
        
        try (Connection conn = dataSource.getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            for (Event event : events) {
                stmt.setObject(1, event.getId());
                stmt.setString(2, event.getType());
                stmt.setString(3, event.getPayload());
                stmt.setObject(4, event.getUserId());
                stmt.setTimestamp(5, Timestamp.valueOf(event.getTimestamp()));
                stmt.addBatch();
            }
            
            stmt.executeBatch();
        } catch (SQLException e) {
            throw new DataAccessException("Failed to batch save events", e);
        }
    }
}

Transaction Management

All repository operations are wrapped in proper transaction boundaries:

@Repository
@Transactional
public class EventRepositoryImpl implements EventRepository {
    
    @PersistenceContext
    private EntityManager entityManager;
    
    @Override
    public Event save(Event event) {
        if (event.getId() == null) {
            entityManager.persist(event);
        } else {
            entityManager.merge(event);
        }
        return event;
    }
    
    @Override
    public Optional<Event> findById(UUID id) {
        return Optional.ofNullable(entityManager.find(Event.class, id));
    }
    
    @Override
    public void delete(UUID id) {
        Event event = entityManager.find(Event.class, id);
        if (event != null) {
            entityManager.remove(event);
        }
    }
}

Read-Through Cache Integration

Repository methods are enhanced with read-through caching:

@Repository
public class CachedEventRepository implements EventRepository {
    
    private final EventRepository delegate;
    private final CacheManager cacheManager;
    
    public CachedEventRepository(EventRepository delegate, CacheManager cacheManager) {
        this.delegate = delegate;
        this.cacheManager = cacheManager;
    }
    
    @Override
    @Cacheable(value = "events", key = "#id")
    public Optional<Event> findById(UUID id) {
        return delegate.findById(id);
    }
    
    @Override
    @CacheEvict(value = "events", key = "#event.id")
    public Event save(Event event) {
        return delegate.save(event);
    }
    
    @Override
    @CacheEvict(value = "events", key = "#id")
    public void delete(UUID id) {
        delegate.delete(id);
    }
}

9. Asynchronous Processing & Messaging

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

The observer-based event system follows a hybrid approach where:

• Synchronous operations handle immediate business logic that must complete before the client receives a response (e.g., event registration, validation)
• Asynchronous operations handle side effects that can be processed later without blocking the main flow (e.g., notifications, analytics, audit logging, sending emails)

This separation ensures that critical business validations and state changes occur synchronously while non-essential operations are decoupled for better scalability and resilience.

Message schema design with a sample payload

{
  "version": "1.0",
  "eventId": "uuid-4f89b3a7-5c2d-4e1a-a9b8-c7d6e5f4g3h2",
  "eventType": "observer.registered",
  "timestamp": "2023-10-15T14:30:00Z",
  "source": "event-system",
  "payload": {
    "observerId": "uuid-12345678-1234-1234-1234-123456789012",
    "eventTypeId": "uuid-abcdef12-3456-7890-abcd-ef1234567890",
    "data": {
      "userId": "user-12345",
      "action": "login",
      "timestamp": "2023-10-15T14:30:00Z"
    }
  },
  "metadata": {
    "correlationId": "corr-12345678-1234-1234-1234-123456789012",
    "priority": "normal"
  }
}

Dead letter queue strategy and poison message handling

Poison messages are handled through a dead-letter queue (DLQ) pattern:

• Messages that fail more than 3 times are moved to a DLQ topic
• DLQ messages are logged with full context for manual inspection and retry
• A separate monitoring dashboard alerts on DLQ message counts and content

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

At-least-once delivery is chosen for the observer event system because:

• The system primarily handles notifications and analytics, where duplicate processing is acceptable
• Event consumers are designed to be idempotent where possible
• The complexity of maintaining exactly-once semantics (e.g., deduplication, transactional publishing) outweighs the benefits for this use case

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 Duration BASE_DELAY = Duration.ofSeconds(1);
    private static final Duration MAX_DELAY = Duration.ofMinutes(5);
    
    public boolean shouldRetry(Exception exception) {
        // Don't retry on validation or business logic errors
        return !(exception instanceof ValidationException || 
                 exception instanceof BusinessLogicException);
    }
    
    public Duration getDelay(int attempt) {
        long delaySeconds = (long) Math.pow(2, attempt) * BASE_DELAY.getSeconds();
        return Duration.ofSeconds(Math.min(delaySeconds, MAX_DELAY.getSeconds()));
    }
}

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

The outbox pattern is omitted for this design because:

• The system uses a single database for event storage and publishing
• Events are published within the same transaction as business logic, ensuring consistency
• The system is designed for eventual consistency rather than strong consistency
• Adding outbox would introduce additional complexity without significant benefit in this context

Sequence diagram for all async flows (Mermaid)

sequenceDiagram
    participant Client
    participant EventPublisher
    participant EventStore
    participant EventConsumer
    
    Client->>EventPublisher: RegisterObserver(eventType)
    EventPublisher->>EventStore: Store observer in DB
    EventPublisher->>EventStore: Store event in DB
    EventPublisher->>EventConsumer: Publish to message queue
    
    loop Retry up to 3 times
        EventConsumer->>EventConsumer: Process event
        alt Success
            EventConsumer->>EventConsumer: Log success
        else Failure
            EventConsumer->>EventConsumer: Retry with exponential backoff
            EventConsumer->>EventConsumer: Move to DLQ after max retries
        end
    end

10. Error Handling Design

Error Classification Hierarchy

Errors are categorized into three main levels:

1. Domain Errors - Business logic violations (e.g., insufficient funds, invalid state transitions)
2. Infrastructure Errors - System-level issues (e.g., database connection failures, cache unavailability)
3. Validation Errors - Input data issues (e.g., missing fields, malformed JSON)

Each error type has a dedicated exception class hierarchy:

// Base exception for all custom errors
public abstract class EventSystemException extends RuntimeException {
    protected final String errorCode;
    protected final int httpStatus;

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

    public String getErrorCode() { return errorCode; }
    public int getHttpStatus() { return httpStatus; }
}

// Domain-specific errors
public class EventProcessingException extends EventSystemException {
    public EventProcessingException(String message) {
        super("EVENT_PROCESSING_ERROR", 400, message);
    }
}

// Infrastructure errors
public class EventSystemInfrastructureException extends EventSystemException {
    public EventSystemInfrastructureException(String message) {
        super("EVENT_SYSTEM_INFRASTRUCTURE_ERROR", 500, message);
    }
}

// Validation errors
public class EventValidationException extends EventSystemException {
    public EventValidationException(String message) {
        super("EVENT_VALIDATION_ERROR", 400, message);
    }
}

Error Propagation Strategy

Errors follow a "bubble up" pattern within service boundaries, but are wrapped with contextual information:

try {
    eventProcessor.processEvent(event);
} catch (DatabaseException e) {
    throw new EventSystemInfrastructureException(
        "Failed to persist event: " + e.getMessage());
} catch (ValidationException e) {
    throw new EventValidationException(
        "Event validation failed: " + e.getMessage());
}

Retry Strategy with Exponential Backoff and Jitter

Retry logic is implemented using exponential backoff with jitter to prevent thundering herd:

public class RetryableOperation {
    private static final int MAX_RETRIES = 3;
    private static final long BASE_DELAY_MS = 1000;
    private static final double JITTER_FACTOR = 0.2;

    public <T> T executeWithRetry(Supplier<T> operation) {
        int attempts = 0;
        while (attempts < MAX_RETRIES) {
            try {
                return operation.get();
            } catch (EventSystemInfrastructureException e) {
                attempts++;
                if (attempts >= MAX_RETRIES) throw e;
                
                long delay = calculateBackoffDelay(attempts);
                Thread.sleep(delay);
            }
        }
        throw new RuntimeException("Max retries exceeded");
    }

    private long calculateBackoffDelay(int attempt) {
        long delay = (long) (BASE_DELAY_MS * Math.pow(2, attempt));
        long jitter = (long) (delay * JITTER_FACTOR * (Math.random() - 0.5) * 2);
        return delay + jitter;
    }
}

Circuit Breaker Design

A circuit breaker pattern is implemented to prevent cascading failures:

public class CircuitBreaker {
    private final int failureThreshold;
    private final long timeout;
    private final AtomicInteger failureCount = new AtomicInteger(0);
    private final AtomicLong lastFailureTime = new AtomicLong(0);
    private volatile CircuitState state = CircuitState.CLOSED;

    enum CircuitState {
        CLOSED, OPEN, HALF_OPEN
    }

    public boolean canExecute() {
        if (state == CircuitState.CLOSED) return true;
        if (state == CircuitState.HALF_OPEN && isTimeoutElapsed()) {
            state = CircuitState.OPEN;
            return true;
        }
        return false;
    }

    public void recordSuccess() {
        failureCount.set(0);
        state = CircuitState.CLOSED;
    }

    public void recordFailure() {
        int failures = failureCount.incrementAndGet();
        lastFailureTime.set(System.currentTimeMillis());
        
        if (failures >= failureThreshold) {
            state = CircuitState.OPEN;
        }
    }

    private boolean isTimeoutElapsed() {
        return System.currentTimeMillis() - lastFailureTime.get() > timeout;
    }
}

Fallback Behavior Design

Fallbacks are implemented for critical dependencies:

public class EventService {
    private final CircuitBreaker circuitBreaker = new CircuitBreaker();
    
    public EventResult processEventWithFallback(Event event) {
        try {
            if (circuitBreaker.canExecute()) {
                return processEvent(event);
            } else {
                // Fallback to default behavior or cached result
                return handleEventWithFallback(event);
            }
        } catch (Exception e) {
            circuitBreaker.recordFailure();
            return handleEventWithFallback(event);
        }
    }

    private EventResult handleEventWithFallback(Event event) {
        // Return default or cached result
        return new EventResult(event.getId(), "fallback_processed");
    }
}

Error Logging Format and Alerting Thresholds

Structured logging includes trace context, error codes, and severity levels:

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

    public void logError(String traceId, String spanId, String userId, 
                        String errorCode, String errorMessage, Exception ex) {
        logger.error(
            "EVENT_ERROR - traceId: {}, spanId: {}, userId: {}, errorCode: {}, message: {}",
            traceId, spanId, userId, errorCode, errorMessage,
            ex);
    }
}

Alerting thresholds:

Metric	Warning Threshold	Critical Threshold
Event processing failures	5 per minute	20 per minute
Circuit breaker trips	3 per hour	10 per hour
Retry attempts exceeded	2 per minute	5 per minute

Error Response Format

Standardized JSON error response:

{
  "error": {
    "code": "EVENT_PROCESSING_ERROR",
    "message": "Failed to process event due to system error",
    "timestamp": "2023-10-15T14:30:00Z",
    "traceId": "abc123def456"
  }
}

11. Security Design

Input Validation and Sanitization Strategy

All inputs are validated using a combination of:

• Regex validation for format constraints (e.g., email, phone number)
• Length limits to prevent buffer overflow attacks
• Type checking for primitive fields (string, integer, boolean)
• Whitelisting for enum-like values and allowed characters

public class InputValidator {
    private static final Pattern EMAIL_PATTERN = 
        Pattern.compile("^[A-Za-z0-9+_.-]+@([A-Za-z0-9.-]+\\.[A-Za-z]{2,})$");
    
    public static boolean isValidEmail(String email) {
        return email != null && EMAIL_PATTERN.matcher(email).matches();
    }
    
    public static boolean isValidStringLength(String input, int maxLen) {
        return input != null && input.length() <= maxLen;
    }
}

SQL Injection Prevention

All database queries use parameterized statements with prepared statements to prevent injection attacks:

// Instead of string concatenation
String query = "SELECT * FROM events WHERE id = ?";
PreparedStatement stmt = connection.prepareStatement(query);
stmt.setString(1, eventId);
ResultSet rs = stmt.executeQuery();

Authentication Middleware Design

Authentication middleware intercepts all requests and validates:

• JWT tokens for stateless authentication
• API key validation for service-to-service communication
• Session tokens for web applications

@Component
public class AuthenticationMiddleware {
    private final JwtService jwtService;
    
    public void validateRequest(HttpServletRequest request) {
        String token = extractToken(request);
        if (!jwtService.validate(token)) {
            throw new AuthenticationException("Invalid token");
        }
    }
}

Authorization Enforcement

RBAC (Role-Based Access Control) is implemented at service layer:

@PreAuthorize("hasRole('EVENT_ADMIN') or hasPermission(#event, 'WRITE')")
public void updateEvent(Event event) {
    // Update logic here
}

Sensitive Data Encryption

• At rest: AES-256 encryption using AWS KMS for event payloads
• In transit: TLS 1.3 with HSTS header enforcement

public class DataEncryptionService {
    private final AesGcmEncryptor encryptor;
    
    public String encrypt(String data) {
        return encryptor.encrypt(data, encryptionKey);
    }
}

PII Handling and Masking Strategy

PII fields (user ID, email, phone) are:

• Stored in encrypted format
• Masked in logs and monitoring (e.g., "user@example.com" → "u***@example.com")
• Removed from non-essential event metadata

Secrets Management

Secrets are managed through AWS Secrets Manager:

• API keys, database credentials, encryption keys
• Automatic rotation with 90-day lifecycle
• IAM policies restrict access to only necessary services

Audit Logging Design

All security-relevant events are logged with:

{
  "timestamp": "2023-12-01T10:30:00Z",
  "user_id": "user-123",
  "action": "EVENT_CREATE",
  "ip_address": "192.168.1.100",
  "user_agent": "Mozilla/5.0...",
  "event_data": {
    "event_id": "event-456",
    "status": "CREATED"
  }
}

Log retention is set to 365 days with automatic archiving to S3 Glacier.

12. Testing Strategy

Unit Testing Scope and Coverage Targets

Unit tests will cover 85%+ of business logic and service layer code with 90%+ line coverage. Each service class will have dedicated test classes with:

• Happy path scenarios for all public methods
• Edge case validation (null inputs, boundary conditions)
• Exception handling verification
• Mocked dependencies for isolated testing

Integration Testing Strategy

Integration tests will validate the complete flow from API to database and external service interactions:

Database Integration:

• Test data persistence with real database connections using testcontainers
• Transaction boundary validation and rollback scenarios
• Index usage verification through query plan analysis

External Service Integration:

• Mocked external service calls using WireMock or similar for:
  • Event publishing to message broker
  • Third-party notifications (email/SMS)
  • Authentication service integration

Cross-component Testing:

• End-to-end flow testing from event creation to observer notification
• Concurrency scenario validation with multiple observers

Contract Testing

Contract testing will be implemented using Pact for:

• API contract verification between event system and consumers
• Message format validation for event payloads
• Consumer-driven contract testing for external integrations

Performance Testing Strategy

Performance testing will focus on:

• Load Testing: Using k6 to simulate concurrent event publishing (1000+ events/sec)
• Stress Testing: Identifying throughput limits and memory usage patterns
• Latency Targets:
  • p50: <50ms for event publishing
  • p95: <200ms for event publishing
  • p99: <500ms for event publishing

Test Data Management Strategy

Test Data Generation:

• Factory pattern for generating valid test entities (Event, Observer, Subscription)
• Fixture-based setup with pre-defined test data sets
• Data seeding strategy using database migrations for test environments

Test Environment Isolation:

• Separate test databases per test suite
• In-memory database for fast unit tests (H2)
• Docker-based integration test environments

Test Data Lifecycle:

• Automatic cleanup after each test run
• Test data versioning for regression testing
• Snapshot testing for event payload structures

// Example unit test structure for observer system
@Test
void shouldNotifyObserversWhenEventPublished() {
    // Given
    Event event = createTestEvent();
    Observer mockObserver = mock(Observer.class);
    
    // When
    eventSystem.publish(event, Set.of(mockObserver));
    
    // Then
    verify(mockObserver).onEvent(event);
}

13. Observability Design

Structured Logging Schema

All logs must follow a consistent schema to enable effective search, filtering, and analysis:

{
  "timestamp": "2023-10-27T14:30:45.123Z",
  "level": "INFO|WARN|ERROR",
  "service": "event-observer-service",
  "trace_id": "1234567890abcdef1234567890abcdef",
  "span_id": "abcdef1234567890abcdef1234567890",
  "user_id": "user-123",
  "entity_id": "event-987654321",
  "event_type": "USER_LOGIN",
  "message": "User successfully logged in",
  "error_code": "AUTH_001",
  "error_details": {
    "stack_trace": "...",
    "cause": "Invalid credentials"
  },
  "duration_ms": 45,
  "tags": ["auth", "security"]
}

PII masking rules:

• User IDs and emails are redacted in logs
• Passwords and tokens are never logged
• Any sensitive fields in payloads are masked

Distributed Tracing Instrumentation

Using OpenTelemetry for distributed tracing:

Key Operations to Trace:

1. EventPublisher.publish() - publish event to system
2. EventObserver.handle() - process observed event
3. EventSubscription.create() - create new subscription
4. EventFilter.evaluate() - filter event against criteria

Custom Spans to Add:

• event.publish - capture from publish start to completion
• event.delivery - span for each observer's handling
• event.filtering - time spent evaluating filters

Attributes to Capture:

Span currentSpan = Span.current();
currentSpan.setAttribute("event_type", eventType);
currentSpan.setAttribute("observer_count", observerCount);
currentSpan.setAttribute("filter_match", filterMatched);

Metrics Design

RED Method Metrics (Rate, Errors, Duration)

Metric Name	Type	Labels	Description
event_published_total	Counter	event_type, status	Total events published
event_processing_errors	Counter	event_type, error_code	Errors during event processing
event_processing_duration_ms	Histogram	event_type, observer_id	Time to process event
subscription_count	Gauge	event_type	Active subscriptions
observer_failure_rate	Gauge	observer_id, event_type	Failure rate per observer

Additional Metrics:

• event_queue_size - gauge for pending events
• observer_concurrent_requests - counter for concurrent handler calls

Health Check Endpoint Design

GET /health
GET /ready  
GET /live
GET /startup

/health - Returns 200 when service is fully operational:

{
  "status": "healthy",
  "timestamp": "2023-10-27T14:30:45Z",
  "components": {
    "database": "healthy",
    "message_queue": "healthy",
    "cache": "healthy"
  }
}

/ready - Returns 200 when service is ready to receive traffic:

{
  "status": "ready",
  "timestamp": "2023-10-27T14:30:45Z",
  "dependencies": {
    "db_connection": true,
    "cache_connected": true
  }
}

/live - Returns 200 when service is alive:

{
  "status": "alive",
  "timestamp": "2023-10-27T14:30:45Z"
}

/startup - Returns 200 when service has completed startup:

{
  "status": "startup_complete",
  "timestamp": "2023-10-27T14:30:45Z",
  "startup_time_ms": 1250
}

Alerting Thresholds

Metric	Warning Threshold	Critical Threshold	Escalation Path
event_processing_errors	> 5 errors/minute	> 20 errors/minute	On-call engineer
event_processing_duration_ms	> 500ms p95	> 2000ms p95	SRE team
observer_failure_rate	> 1% failure rate	> 5% failure rate	Engineering lead
event_queue_size	> 1000 pending events	> 5000 pending events	SRE team
event_published_total	Drop below 50% of baseline	Drop below 10% of baseline	Platform team

Alerts should be sent via PagerDuty for critical issues and Slack for warnings.

14. Open Questions, Risks & Trade-offs

Decisions requiring product / engineering alignment

• Event ordering guarantees: Should the system guarantee strict ordering of events across all subscribers, or is eventual consistency acceptable? This decision impacts the complexity of the event store and subscriber coordination mechanisms.

• Event versioning strategy: How should we handle breaking changes to event schemas? Should we use a single global version or allow per-subscriber versioning?

• Event persistence model: Should events be persisted in the same database as business data, or in a dedicated event store? This affects consistency, performance, and system complexity.

Known trade-offs and justifications

Trade-off	Option 1	Option 2	Justification
Event Store vs. DB	Single database for events and business data	Dedicated event store (e.g., Kafka)	Single DB is simpler to implement and maintain but may not scale well for high-throughput event processing. Kafka provides better performance, durability, and horizontal scaling but adds operational complexity.
Synchronous vs. Asynchronous	All events processed synchronously in same transaction	Events published asynchronously	Synchronous processing ensures strong consistency but introduces latency and potential failure points. Asynchronous processing provides better performance and resilience but requires careful handling of eventual consistency and error recovery.
Event Retention	Unlimited event retention	Limited retention (e.g., 30 days)	Unlimited retention provides maximum flexibility but can lead to storage and performance issues. Limited retention reduces storage costs but may prevent debugging of older events or compliance requirements.
Event Schema Validation	No validation at publish time	Strict schema validation with versioning	No validation makes it easy to publish events but increases risk of malformed data. Schema validation ensures data quality but requires more complex infrastructure and versioning strategy.

Risks and mitigation strategies

• Event loss: Risk of losing events due to system failures or network issues.

  • Mitigation: Implement an outbox pattern with database-backed event storage, use durable message brokers (e.g., Kafka), and implement proper error handling with dead-letter queues.

• Event ordering: Risk of events being processed in wrong order, especially for related operations.

  • Mitigation: Use sequence numbers or timestamps to enforce ordering, implement idempotent consumers, and design events with sufficient context for proper processing order.

• Performance degradation: Risk of event publishing or consumption causing performance bottlenecks.

  • Mitigation: Implement proper caching strategies, use connection pooling, apply batch processing where appropriate, and monitor system metrics to identify bottlenecks early.

• Data inconsistency: Risk of subscribers processing events inconsistently with the system state.

  • Mitigation: Implement idempotent event handling, use transactional messaging where possible, and maintain proper audit trails for debugging.

Technical debt acknowledged

• Event schema evolution complexity: Future changes to event schemas may require complex migration strategies and backward compatibility handling, leading to technical debt in the event processing layer.

• Subscriber management overhead: Managing multiple subscribers with different capabilities and requirements introduces complexity in the system, particularly around event routing and filtering.

• Monitoring and debugging challenges: With distributed event processing, debugging issues across systems becomes more complex without proper observability tooling and centralized logging.

• Event data duplication: Events may be duplicated across subscribers or systems, creating challenges in maintaining data consistency and increasing storage requirements.

• Circuit breaker complexity: Implementing proper circuit breaker patterns for event publishers and subscribers adds significant complexity to the system without clear benefits in low-volume scenarios.