Messaging System Design: Scalable Real-Time Communication Architecture

This document presents a comprehensive low-level design for a scalable messaging system that handles real-time communication with guaranteed delivery, persistence, and horizontal scalability across distributed systems.

1. Problem Statement & Requirements

Clarifying questions

1. What scale expectations are we targeting? What is the expected message volume (e.g., millions of messages per second) and concurrent users?
2. What consistency requirements exist? Are there strict ordering guarantees needed (e.g., FIFO, exactly-once delivery) or can eventual consistency be acceptable?
3. What are the key entities and their relationships? Are messages tied to specific users, topics, or channels? How do different message types relate (e.g., direct messages vs. group chats)?
4. What is the read/write ratio? Is this primarily a write-heavy system (e.g., chat app) or does it support frequent reads (e.g., notifications)?
5. What are the latency sensitivity requirements? Are real-time messages (e.g., chat) expected to arrive within 100ms, or is batch processing acceptable?
6. What authentication and authorization model is required? Should we support OAuth2, JWT tokens, or API keys? What level of granular access control is needed?
7. What external integrations or systems will this messaging system interact with? Will it integrate with third-party services (e.g., push notifications, SMS gateways) or support webhooks?

Functional requirements

• Message creation and delivery: Users can send messages to other users or groups.
• Message retrieval: Users can fetch their message history, either by conversation or by time range.
• Real-time messaging: Support for live chat with low latency (e.g., WebSocket or long-polling).
• Message persistence: Messages are stored for retrieval and replay.
• Message types: Support for text, image, file, and system-generated messages (e.g., read receipts).
• Message threading: Support for replies and nested conversations.
• User presence: Indicate when users are online or offline.
• Message read receipts: Track when messages have been seen by recipients.

Non-functional requirements

• Latency targets:
  • p50: < 100ms for real-time messages
  • p95: < 500ms for all message operations
  • p99: < 2s for message retrieval and delivery
• Throughput: Support at least 100,000 messages per second for real-time messaging.
• Availability: 99.9% uptime (5-minute maximum downtime per year).
• Scalability: Support horizontal scaling to handle millions of concurrent users.
• Message durability: Messages must survive system failures and restarts.

Constraints and assumptions

• Assumption: Messages are not expected to exceed 1MB in size.
• Assumption: Users will have a limited number of concurrent conversations (e.g., < 100).
• Constraint: Message history is not expected to be retained indefinitely; a retention policy (e.g., 30 days) will apply.
• Constraint: The system does not support message encryption at rest or in transit by default (this can be added later).
• Assumption: Authentication and authorization will be handled at the API gateway or middleware layer.

Out of scope

• Push notification integration: Not part of the core messaging engine.
• Message encryption: End-to-end encryption is not part of this design (can be added via plugins or extensions).
• File storage: File uploads are not handled by this system (e.g., file URLs stored in message content).
• Message search: Full-text search capabilities are not included.
• Cross-service federation: Messaging between different systems or domains is out of scope.

2. API Design

API Style Selection

For a messaging system, I'll select gRPC with Protocol Buffers as the primary API style. This choice provides:

• Strong typing and automatic code generation
• Efficient binary serialization for high-throughput messaging
• First-class support for streaming operations (bidirectional streams)
• Built-in support for versioning and schema evolution
• Better performance characteristics than REST for message-heavy workloads

Endpoint Definitions

Method	Path	Description	Auth
CreateMessage	/v1/messages	Create a new message in a conversation	JWT
GetMessage	/v1/messages/{id}	Retrieve a specific message by ID	JWT
ListMessages	/v1/conversations/{conversationId}/messages	List messages in a conversation with pagination	JWT
SendMessage	/v1/conversations/{conversationId}/send	Send a message to a conversation	JWT
UpdateMessage	/v1/messages/{id}	Update an existing message content	JWT
DeleteMessage	/v1/messages/{id}	Soft-delete a message by ID	JWT
MarkAsRead	/v1/conversations/{conversationId}/mark-read	Mark all messages as read in a conversation	JWT
GetConversation	/v1/conversations/{id}	Retrieve conversation details	JWT
ListConversations	/v1/conversations	List user's conversations with metadata	JWT

Request and Response Schemas

CreateMessageRequest

{
  "conversationId": "uuid",
  "content": "string",
  "attachments": [
    {
      "name": "string",
      "url": "string",
      "size": "int64"
    }
  ]
}

CreateMessageResponse

{
  "message": {
    "id": "uuid",
    "conversationId": "uuid",
    "senderId": "uuid",
    "content": "string",
    "attachments": [
      {
        "name": "string",
        "url": "string",
 "size": "int64"
      }
    ],
    "createdAt": "timestamp",
    "updatedAt": "timestamp",
    "status": "enum"
  }
}

ListMessagesRequest

{
  "conversationId": "uuid",
  "limit": "int32",
  "cursor": "string"
}

ListMessagesResponse

{
  "messages": [
    {
      "id": "uuid",
      "conversationId": "uuid",
      "senderId": "uuid",
      "content": "string",
      "attachments": [
        {
          "name": "string",
          "url": "string",
          "size": "int64"
        }
      ],
      "createdAt": "timestamp",
      "updatedAt": "timestamp",
      "status": "enum"
    }
  ],
  "nextCursor": "string"
}

HTTP Status Codes and Error Response Format

Standard HTTP status codes with custom error responses:

Status Code	Description
200	Success
201	Created
400	Bad Request (validation errors)
401	Unauthorized
403	Forbidden
404	Not Found
429	Rate Limited
500	Internal Server Error

Standard error response format:

{
  "error": {
    "code": "string",
    "message": "string",
    "details": [
      {
        "field": "string",
        "message": "string"
      }
    ]
  }
}

API Versioning Strategy

Versioned using path prefix: /v1/ with support for future versions like /v2/. This approach provides:

• Clear separation between versions
• Backward compatibility for existing clients
• Easy migration path for consumers

Pagination Design

Using cursor-based pagination with a cursor token that includes:

• Message ID
• Timestamp for consistency
• Direction (forward/backward)

This approach is preferred over offset-based pagination because:

• More stable with concurrent writes
• Better performance for large datasets
• Consistent results even when messages are deleted

Authentication and Authorization

All endpoints require JWT tokens with scopes:

• messages:read - for reading messages
• messages:write - for creating/updating messages
• conversations:read - for reading conversations

JWT claims will include:

{
  "sub": "user_id",
  "scope": ["messages:read", "messages:write"],
  "iat": "timestamp",
  "exp": "timestamp"
}

Rate Limiting

Per endpoint rate limiting:

Endpoint	Limit	Period
CreateMessage	100/minute	1 minute
ListMessages	500/minute	1 minute
SendMessage	50/minute	1 minute

Rate limiting enforced using Redis with token bucket algorithm to prevent bursty traffic.

Idempotency Design

For mutation endpoints, idempotency key strategy:

1. Clients provide optional Idempotency-Key header
2. Server stores operation result with key as identifier
3. If same key is used, return cached result instead of reprocessing
4. Key validity: 5 minutes

Implementation pattern:

public class IdempotencyService {
    private final RedisTemplate<String, String> redisTemplate;
    
    public <T> T executeWithIdempotency(String key, Supplier<T> operation) {
        String cacheKey = "idempotency:" + key;
        String result = redisTemplate.opsForValue().get(cacheKey);
        
        if (result != null) {
            return deserialize(result, operation.get().getClass());
        }
        
        T value = operation.get();
        redisTemplate.opsForValue().set(cacheKey, serialize(value), 5, TimeUnit.MINUTES);
        return value;
    }
}

3. Data Model Design

Entity-Relationship diagram (Mermaid)

erDiagram
    USER {
        string id PK
        string email
        string name
        timestamp created_at
        timestamp updated_at
    }

    MESSAGE {
        string id PK
        string content
        string sender_id FK
        string recipient_id FK
        string thread_id FK
        boolean is_read
        timestamp created_at
        timestamp updated_at
    }

    THREAD {
        string id PK
        string title
        string type
        timestamp created_at
        timestamp updated_at
    }

    THREAD_MEMBER {
        string thread_id FK
        string user_id FK
        timestamp joined_at
        timestamp left_at
    }

    USER {
        string id PK
        string email
        string name
        timestamp created_at
        timestamp updated_at
    }

    MESSAGE --> THREAD : belongs_to
    MESSAGE --> USER : sent_by
    MESSAGE --> USER : received_by
    THREAD_MEMBER --> THREAD : belongs_to
    THREAD_MEMBER --> USER : member_of

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

Users table

CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email VARCHAR(255) UNIQUE NOT NULL,
    name VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_users_email (email),
    INDEX idx_users_created_at (created_at)
);

Threads table

CREATE TABLE threads (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    title VARCHAR(255),
    type VARCHAR(50) NOT NULL, -- 'direct', 'group'
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_threads_created_at (created_at),
    INDEX idx_threads_type (type)
);

Thread members table

CREATE TABLE thread_members (
    thread_id UUID NOT NULL,
    user_id UUID NOT NULL,
    joined_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    left_at TIMESTAMP,
    
    PRIMARY KEY (thread_id, user_id),
    FOREIGN KEY (thread_id) REFERENCES threads(id) ON DELETE CASCADE,
    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
    
    INDEX idx_thread_members_thread_id (thread_id),
    INDEX idx_thread_members_user_id (user_id),
    INDEX idx_thread_members_joined_at (joined_at)
);

Messages table

CREATE TABLE messages (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    content TEXT NOT NULL,
    sender_id UUID NOT NULL,
    recipient_id UUID NOT NULL,
    thread_id UUID NOT NULL,
    is_read BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    FOREIGN KEY (sender_id) REFERENCES users(id) ON DELETE CASCADE,
    FOREIGN KEY (recipient_id) REFERENCES users(id) ON DELETE CASCADE,
    FOREIGN KEY (thread_id) REFERENCES threads(id) ON DELETE CASCADE,
    
    INDEX idx_messages_thread_id (thread_id),
    INDEX idx_messages_sender_id (sender_id),
    INDEX idx_messages_recipient_id (recipient_id),
    INDEX idx_messages_created_at (created_at),
    INDEX idx_messages_is_read (is_read),
    INDEX idx_messages_thread_created_at (thread_id, created_at)
);

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

Using UUID as primary key for all tables:

• Justification:
  • Enables distributed system design without coordination
  • Provides better sharding capabilities across multiple database instances
  • Prevents enumeration attacks and makes it harder to predict IDs
  • Supports eventual consistency in multi-region deployments
  • UUIDs are globally unique, avoiding potential conflicts in federated systems

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

1. Composite Indexes:

   • idx_messages_thread_created_at (thread_id, created_at) - For efficient pagination and querying by thread
   • idx_thread_members_thread_id (thread_id) - For fast lookups of members in a thread

2. Partial Indexes:

   • idx_messages_unread_by_recipient (recipient_id, is_read) WHERE is_read = FALSE - For efficient unread message counts per user

3. Covering Indexes:

   • idx_messages_thread_id_content (thread_id, content) - For queries that need to fetch message content within a thread without additional lookups

Data partitioning / sharding strategy

Sharding by Thread ID:

• Partition messages by thread_id to colocate related messages
• Each thread's messages are stored together, enabling efficient retrieval and updates
• Thread IDs can be sharded across database instances using consistent hashing or range-based partitioning

User-based sharding for thread membership:

• Thread members table can be sharded by user_id to support efficient user thread queries

Soft delete vs. hard delete decision with justification

Soft Delete:

• All tables use soft delete pattern with deleted_at timestamp field
• Justification:
  • Preserves data integrity for audit and compliance requirements
  • Allows for recovery of accidentally deleted messages/threads
  • Maintains referential integrity in complex relationships (e.g., thread members)
  • Supports data analytics and historical analysis

Audit trail design (who, what, when)

All tables include audit fields:

• created_at - When record was created
• updated_at - When record was last modified
• created_by - User who created the record (if applicable)
• updated_by - User who last modified the record (if applicable)

Data archival strategy and retention policy

Archival Strategy:

• Messages older than 30 days are moved to archival storage (e.g., S3 with Glacier)
• Thread metadata older than 2 years is archived
• User data retention: 5 years for inactive users (based on last login)

Retention Policy:

• Messages: 30 days
• Threads: 2 years (after all members leave)
• User profiles: 5 years (after last activity)

4. Class Design & Service Layer

classDiagram
    class MessageService {
        +send(Message message)
        +receive(String userId, String messageId)
        +delete(String userId, String messageId)
        +listUserMessages(String userId, int limit, String cursor)
        +markAsRead(String userId, String messageId)
    }
    
    class Message {
        -id: UUID
        -senderId: String
        -recipientId: String
        -content: String
        -status: MessageStatus
        -createdAt: Instant
        -readAt: Instant
        -expiresAt: Instant
        +validate()
    }
    
    class MessageRepository {
        +save(Message message)
        +findById(String id)
        +findByRecipient(String recipientId, int limit, String cursor)
        +updateStatus(String id, MessageStatus status)
        +deleteById(String id)
    }
    
    class MessageEventPublisher {
        +publish(MessageEvent event)
    }
    
    class MessageEvent {
        -id: UUID
        -messageId: String
        -eventType: EventType
        -timestamp: Instant
    }
    
    class MessageValidator {
        +validate(Message message)
    }
    
    class MessageProcessor {
        +process(Message message)
    }
    
    class MessageStatus {
        PENDING
        SENT
        READ
        DELETED
    }
    
    class EventType {
        MESSAGE_SENT
        MESSAGE_READ
        MESSAGE_DELETED
    }
    
    MessageService --> Message : uses
    MessageService --> MessageRepository : persists
    MessageService --> MessageEventPublisher : publishes
    MessageService --> MessageValidator : validates
    MessageService --> MessageProcessor : processes
    
    Message --> MessageStatus : has
    Message --> EventType : triggers

stateDiagram-v2
    [*] --> PENDING
    PENDING --> SENT : send()
    SENT --> READ : markAsRead()
    SENT --> DELETED : delete()
    READ --> DELETED : delete()
    DELETED --> [*] : cleanup()

Service Class Structure and Responsibilities

The messaging system is designed with a clean separation of concerns:

1. MessageService - Main orchestrator that handles business logic and coordinates between components
2. MessageRepository - Data persistence layer for message entities
3. MessageEventPublisher - Event publishing infrastructure
4. MessageValidator - Input validation and business rule enforcement
5. MessageProcessor - Message processing pipeline (e.g., content filtering, encryption)

Interface / Abstract Class Design

public interface MessageRepository {
    Message save(Message message);
    Optional<Message> findById(String id);
    List<Message> findByRecipient(String recipientId, int limit, String cursor);
    void updateStatus(String id, MessageStatus status);
    void deleteById(String id);
}

public interface MessageEventPublisher {
    void publish(MessageEvent event);
}

public interface MessageValidator {
    ValidationResult validate(Message message);
}

public enum MessageStatus {
    PENDING, SENT, READ, DELETED
}

Business Logic Encapsulation Patterns

Strategy Pattern - For handling different message content types:

public interface ContentProcessor {
    String process(String content, MessageContext context);
}

@Component
public class TextContentProcessor implements ContentProcessor {
    @Override
    public String process(String content, MessageContext context) {
        return content.trim();
    }
}

@Component
public class EncryptedContentProcessor implements ContentProcessor {
    @Override
    public String process(String content, MessageContext context) {
        return encrypt(content, context.getEncryptionKey());
    }
}

public class MessageProcessor {
    private final Map<String, ContentProcessor> processors;
    
    public String processContent(String content, MessageContext context) {
        ContentProcessor processor = processors.get(context.getContentType());
        return processor != null ? processor.process(content, context) : content;
    }
}

Critical-Path Method Implementation

@Transactional
public Message send(Message message) {
    // Validate message
    ValidationResult validation = validator.validate(message);
    if (!validation.isValid()) {
        throw new ValidationException("Message validation failed: " + validation.getErrors());
    }
    
    // Set message properties
    message.setId(UUID.randomUUID().toString());
    message.setStatus(MessageStatus.PENDING);
    message.setCreatedAt(Instant.now());
    
    // Apply content processing
    MessageContext context = new MessageContext(
        message.getRecipientId(), 
        "text/plain", 
        null // encryption key
    );
    message.setContent(contentProcessor.processContent(message.getContent(), context));
    
    // Save to database
    Message saved = repository.save(message);
    
    // Publish event for async processing
    MessageEvent event = new MessageEvent(
        UUID.randomUUID().toString(),
        saved.getId(),
        EventType.MESSAGE_SENT,
        Instant.now()
    );
    
    eventPublisher.publish(event);
    
    // Update status to SENT
    repository.updateStatus(saved.getId(), MessageStatus.SENT);
    
    return saved;
}

Transaction Boundary Design

A single transaction encompasses:

1. Message validation and content processing
2. Database persistence of the message record
3. Status update to SENT state

This ensures atomicity for the core messaging operation.

Service Interface Contracts

public interface MessageService {
    Message send(Message message);
    Optional<Message> receive(String userId, String messageId);
    void delete(String userId, String messageId);
    List<Message> listUserMessages(String userId, int limit, String cursor);
    void markAsRead(String userId, String messageId);
}

Saga / Orchestration Pattern

For complex operations like message forwarding or group messaging, a saga pattern would be implemented:

public class MessageForwardSaga {
    private final List<SagaStep> steps;
    
    public void execute(ForwardRequest request) {
        // Step 1: Validate forward request
        // Step 2: Create forwarded message copy
        // Step 3: Send to recipient
        // Step 4: Update original message status if needed
        // Step 5: Handle rollback on failure
    }
}

Domain Object / Value Object Design

public class Message {
    private final String id;
    private final String senderId;
    private final String recipientId;
    private final String content;
    private MessageStatus status;
    private final Instant createdAt;
    private Instant readAt;
    private final Instant expiresAt;
    
    // Constructor, getters, setters...
}

public class MessageContext {
    private final String recipientId;
    private final String contentType;
    private final String encryptionKey;
    
    // Constructor, getters...
}

public class ValidationResult {
    private final boolean valid;
    private final List<String> errors;
    
    // Constructor, getters...
}

5. Concurrency & Thread Safety

Concurrency Model

The messaging system will leverage a multi-threaded approach using Java's java.util.concurrent package and thread pools for handling concurrent message processing. The system will use a fixed-size thread pool with configurable size based on the expected throughput and CPU core count. Each message processing task will be executed in a separate thread from the pool to ensure non-blocking operation and prevent thread starvation.

Race Condition Prevention Strategy

Race conditions in the messaging system will be prevented through:

1. Immutable message objects - Once created, messages cannot be modified
2. Thread-local storage for temporary state during processing
3. Synchronized access to shared mutable state using ReentrantLock or StampedLock
4. Atomic operations for counters and sequence numbers using java.util.concurrent.atomic classes

Optimistic vs. Pessimistic Locking Selection

Optimistic locking will be used for message processing operations due to:

• Low contention in typical messaging workloads
• Better performance under concurrent access patterns
• Message processing is primarily read-heavy with minimal updates
• Implementation simplicity using version numbers or timestamps

For message delivery state updates, we'll use pessimistic locking when updating delivery status to prevent race conditions during concurrent delivery attempts.

Deadlock Prevention Strategy

Deadlock prevention will be implemented through:

1. Lock ordering - All threads acquire locks in a consistent order (e.g., by message ID)
2. Timeout-based lock acquisition - Using tryLock() with timeout to prevent indefinite blocking
3. Hierarchical lock management - Locks acquired in a defined hierarchy to avoid circular dependencies
4. Thread monitoring - Periodic checks for long-running locks

Idempotency Implementation for Concurrent Duplicate Requests

Idempotency will be handled using:

1. Message ID uniqueness - Each message has a globally unique identifier that prevents duplicate processing
2. Idempotency keys - For operations like message delivery, a key can be used to identify duplicate requests
3. Database-level uniqueness constraints - Primary keys and unique indexes prevent duplicate message storage

public class MessageProcessor {
    private final Map<String, CompletableFuture<Void>> processingFutures = new ConcurrentHashMap<>();
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    
    public CompletableFuture<Void> processMessageAsync(Message message) {
        String messageId = message.getId();
        
        // Check if already processing
        CompletableFuture<Void> existingFuture = processingFutures.get(messageId);
        if (existingFuture != null && !existingFuture.isDone()) {
            return existingFuture;
        }
        
        // Create new processing future
        CompletableFuture<Void> newFuture = CompletableFuture.runAsync(() -> {
            try {
                // Acquire write lock for processing
                lock.writeLock().lock();
                
                // Check again if message was processed while waiting for lock
                if (processingFutures.containsKey(messageId)) {
                    return;
                }
                
                // Process message
                processMessage(message);
                
                // Mark as completed
                processingFutures.put(messageId, CompletableFuture.completedFuture(null));
            } finally {
                lock.writeLock().unlock();
            }
        });
        
        processingFutures.put(messageId, newFuture);
        return newFuture;
    }
    
    private void processMessage(Message message) {
        // Message processing logic here
        // This method is thread-safe due to lock protection
    }
}

Thread Safety in Message Queue Operations

The message queue implementation will ensure thread safety through:

1. Concurrent collections - ConcurrentHashMap for message metadata, BlockingQueue for message storage
2. Atomic operations - For counters and sequence numbers
3. Synchronized methods for critical sections that require exclusive access

public class ThreadSafeMessageQueue {
    private final BlockingQueue<Message> queue = new LinkedBlockingQueue<>();
    private final AtomicInteger messageCount = new AtomicInteger(0);
    
    public void enqueue(Message message) {
        queue.offer(message);
        messageCount.incrementAndGet();
    }
    
    public Message dequeue() {
        return queue.poll();
    }
    
    public int size() {
        return messageCount.get();
    }
}

Message Delivery State Management

Delivery state updates will use optimistic concurrency control:

public class MessageDeliveryService {
    private final Map<String, AtomicLong> deliveryAttempts = new ConcurrentHashMap<>();
    
    public boolean updateDeliveryStatus(String messageId, DeliveryStatus status) {
        // Use optimistic locking with versioning
        AtomicLong attemptCounter = deliveryAttempts.computeIfAbsent(messageId, k -> new AtomicLong(0));
        long currentAttempt = attemptCounter.get();
        
        // Simulate version check - in real system would use database version field
        if (attemptCounter.compareAndSet(currentAttempt, currentAttempt + 1)) {
            // Safe to update - no concurrent modification occurred
            return updateMessageDeliveryState(messageId, status);
        }
        
        // Concurrent modification - retry or handle appropriately
        return false;
    }
    
    private boolean updateMessageDeliveryState(String messageId, DeliveryStatus status) {
        // Actual database update logic here
        return true;
    }
}

Performance Considerations

1. Thread pool sizing - Configurable based on CPU cores and message volume
2. Batch processing - Concurrent processing of batches to improve throughput
3. Lock-free alternatives - Use java.util.concurrent primitives where possible to reduce contention
4. Memory barriers - Proper visibility guarantees using volatile fields and atomic operations

The design balances performance with thread safety, ensuring that high-concurrency scenarios are handled gracefully while maintaining data consistency and preventing race conditions.

6. Sequence Diagrams for Key Flows

sequenceDiagram
    participant U as User
    participant API as Messaging API
    participant AuthService as Auth Service
    participant DB as Database
    participant Cache as Redis Cache
    participant MessageBroker as Kafka/EventBus

    U->>API: Send Message Request (POST /messages)
    API->>AuthService: Validate JWT Token
    AuthService-->>API: Authenticated User ID
    API->>DB: Create Message Record
    DB-->>API: Message ID
    API->>Cache: Cache Message Metadata
    Cache-->>API: Cached
    API->>MessageBroker: Publish Message Event
    MessageBroker-->>API: Event Published
    API-->>U: 201 Created with Message ID

sequenceDiagram
    participant U as User
    participant API as Messaging API
    participant AuthService as Auth Service
    participant DB as Database
    participant MessageBroker as Kafka/EventBus
    participant Consumer as Message Consumer

    U->>API: Get Messages Request (GET /messages?userId=123)
    API->>AuthService: Validate JWT Token
    AuthService-->>API: Authenticated User ID
    API->>DB: Query Messages for User
    DB-->>API: Message List
    API-->>U: 200 OK with Messages

    Note over Consumer,MessageBroker: Async processing of message events
    MessageBroker->>Consumer: Message Event (user_id=123, content="Hello")
    Consumer->>DB: Update Message Status
    DB-->>Consumer: Updated
    Consumer->>Cache: Invalidate User Cache
    Cache-->>Consumer: Cache invalidated

sequenceDiagram
    participant U as User
    participant API as Messaging API
    participant AuthService as Auth Service
    participant DB as Database
    participant MessageBroker as Kafka/EventBus
    participant Consumer as Message Consumer
    participant EmailService as Email Service

    U->>API: Send Message Request (POST /messages)
    API->>AuthService: Validate JWT Token
    AuthService-->>API: Authenticated User ID
    API->>DB: Create Message Record (status=pending)
    DB-->>API: Message ID
    API->>MessageBroker: Publish Message Event (message_id=123)
    MessageBroker-->>API: Event Published
    API-->>U: 201 Created with Message ID

    MessageBroker->>Consumer: Message Event (message_id=123, content="Hello World")
    Consumer->>DB: Fetch Message Details
    DB-->>Consumer: Message Data
    Consumer->>EmailService: Send Email Notification
    EmailService-->>Consumer: Email Sent
    Consumer->>DB: Update Message Status (status=sent)
    DB-->>Consumer: Updated
    Consumer->>Cache: Invalidate Message Cache
    Cache-->>Consumer: Cache invalidated

7. Caching Design

Cache Strategy Selection

For the messaging system, we'll use a cache-aside pattern with LRU eviction policy. This approach provides the best balance between performance and consistency for a messaging system where recent messages are most frequently accessed.

Cache Key Design

Cache keys will follow this namespace structure:

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

Example key format for a message cache:

msg:inbox:{userId}:{messageId}:v1
msg:thread:{threadId}:{messageId}:v1

TTL Strategy

Data Type	TTL	Reason
Recent messages (inbox)	30 minutes	High access frequency, recent messages most important
Thread history	2 hours	Moderate access, longer retention for context
User metadata	1 hour	Less volatile than messages
Message content	30 minutes	Content changes infrequently but needs freshness

Cache Eviction Policy

We'll implement LRU (Least Recently Used) eviction with probabilistic early expiry to prevent cache stampede:

public class CacheManager {
    private final RedisTemplate<String, Object> redisTemplate;
    private final int cacheTtlSeconds = 1800; // 30 minutes
    
    public Object getCachedValue(String key) {
        String cacheKey = "msg:" + key;
        Object cached = redisTemplate.opsForValue().get(cacheKey);
        
        if (cached != null) {
            // Probabilistic early expiry - refresh 10% before TTL
            long ttl = redisTemplate.getExpire(cacheKey);
            if (ttl < cacheTtlSeconds * 0.9) {
                // Refresh in background
                CompletableFuture.runAsync(() -> refreshCache(cacheKey));
            }
        }
        
        return cached;
    }
    
    private void refreshCache(String key) {
        // Implementation for cache warmup
    }
}

Cache Stampede Prevention

We'll use distributed locks with probabilistic early expiry:

public class CacheStampedePrevention {
    private static final String LOCK_PREFIX = "cache_lock:";
    private static final int LOCK_TTL_SECONDS = 5;
    
    public Object getCachedWithLock(String key, Supplier<Object> cacheLoader) {
        String lockKey = LOCK_PREFIX + key;
        String lockValue = UUID.randomUUID().toString();
        
        try {
            // Try to acquire lock
            if (redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 
                Duration.ofSeconds(LOCK_TTL_SECONDS))) {
                
                // Load from cache or database
                Object value = getCachedValue(key);
                if (value == null) {
                    value = cacheLoader.get();
                    setCachedValue(key, value);
                }
                
                return value;
            } else {
                // Wait and retry
                Thread.sleep(50);
                return getCachedValue(key);
            }
        } finally {
            // Release lock if owned by this instance
            releaseLock(lockKey, lockValue);
        }
    }
    
    private void releaseLock(String lockKey, String lockValue) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), 
            Collections.singletonList(lockKey), lockValue);
    }
}

Distributed Cache Architecture

We'll use Redis Cluster with:

• 3 master nodes for high availability
• 3 replica nodes per master for read scaling
• Redis Sentinel for automatic failover
• Cache sharding based on user ID modulo cluster size

Cache Invalidation Strategy

We'll implement event-driven invalidation:

@Component
public class MessageCacheInvalidator {
    
    @EventListener
    public void handleMessageCreated(MessageCreatedEvent event) {
        // Invalidate inbox cache for recipient
        String inboxKey = "msg:inbox:" + event.getRecipientId();
        redisTemplate.delete(inboxKey);
        
        // Invalidate thread cache if applicable
        String threadKey = "msg:thread:" + event.getThreadId();
        redisTemplate.delete(threadKey);
        
        // Invalidate user metadata cache
        String userKey = "msg:user:" + event.getRecipientId();
        redisTemplate.delete(userKey);
    }
    
    @EventListener
    public void handleMessageUpdated(MessageUpdatedEvent event) {
        // Invalidate specific message cache
        String msgKey = "msg:message:" + event.getMessageId();
        redisTemplate.delete(msgKey);
        
        // Invalidate thread cache
        String threadKey = "msg:thread:" + event.getThreadId();
        redisTemplate.delete(threadKey);
    }
}

Cache Hit/Miss Metrics

@Component
public class CacheMetrics {
    private final MeterRegistry meterRegistry;
    
    public void recordCacheHit(String cacheType) {
        Counter.builder("cache.hits")
            .tag("type", cacheType)
            .register(meterRegistry)
            .increment();
    }
    
    public void recordCacheMiss(String cacheType) {
        Counter.builder("cache.misses")
            .tag("type", cacheType)
            .register(meterRegistry)
            .increment();
    }
}

Cache Warmup Strategy

For high-traffic users, we'll implement cache warmup during user session establishment:

@Service
public class CacheWarmupService {
    
    public void warmUpUserCache(String userId) {
        // Preload recent messages for user
        List<Message> recentMessages = messageRepository.findRecentByUser(userId, 50);
        
        for (Message msg : recentMessages) {
            String key = "msg:message:" + msg.getId();
            redisTemplate.opsForValue().set(key, msg, Duration.ofMinutes(30));
        }
        
        // Preload thread metadata
        String threadKey = "msg:thread:" + msg.getThreadId();
        redisTemplate.opsForValue().set(threadKey, threadMetadata, Duration.ofHours(2));
    }
}

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 database operations. Each entity will have its own repository interface and implementation.

public interface MessageRepository {
    Message save(Message message);
    Optional<Message> findById(UUID id);
    List<Message> findByConversationId(UUID conversationId, Pageable pageable);
    void deleteById(UUID id);
    long countByConversationId(UUID conversationId);
}

@Repository
public class MessageRepositoryImpl implements MessageRepository {
    private final JdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;
    
    public MessageRepositoryImpl(JdbcTemplate jdbcTemplate, ObjectMapper objectMapper) {
        this.jdbcTemplate = jdbcTemplate;
        this.objectMapper = objectMapper;
    }
    
    @Override
    public Message save(Message message) {
        String sql = """
            INSERT INTO messages (id, conversation_id, sender_id, content, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?)
            """;
        
        jdbcTemplate.update(sql, 
            message.getId(),
            message.getConversationId(),
            message.getSenderId(),
            message.getContent(),
            message.getCreatedAt(),
            message.getUpdatedAt()
        );
        
        return message;
    }
    
    @Override
    public Optional<Message> findById(UUID id) {
        String sql = "SELECT * FROM messages WHERE id = ?";
        try {
            return Optional.of(jdbcTemplate.queryForObject(sql, new MessageRowMapper(), id));
        } catch (EmptyResultDataAccessException e) {
            return Optional.empty();
        }
    }
    
    @Override
    public List<Message> findByConversationId(UUID conversationId, Pageable pageable) {
        String sql = """
            SELECT * FROM messages 
            WHERE conversation_id = ? 
            ORDER BY created_at ASC 
            LIMIT ? OFFSET ?
            """;
        
        return jdbcTemplate.query(sql, new MessageRowMapper(), 
            conversationId, 
            pageable.getPageSize(), 
            pageable.getOffset());
    }
    
    @Override
    public void deleteById(UUID id) {
        String sql = "DELETE FROM messages WHERE id = ?";
        jdbcTemplate.update(sql, id);
    }
    
    @Override
    public long countByConversationId(UUID conversationId) {
        String sql = "SELECT COUNT(*) FROM messages WHERE conversation_id = ?";
        return jdbcTemplate.queryForObject(sql, Long.class, conversationId);
    }
}

ORM vs Raw Query Decisions

For this messaging system, raw SQL queries are preferred over ORM for the following reasons:

1. Performance: Raw queries provide better performance with minimal overhead
2. Control: Fine-grained control over query execution and optimization
3. Complexity: The message model is relatively simple, so ORM complexity isn't justified
4. Database portability: Raw queries can be optimized per database engine

Read vs Write Separation

The messaging system uses a read-write separation strategy to handle high-volume reads efficiently:

public interface MessageReadRepository {
    List<Message> findByConversationId(UUID conversationId, Pageable pageable);
    long countByConversationId(UUID conversationId);
    Optional<Message> findById(UUID id);
}

public interface MessageWriteRepository {
    Message save(Message message);
    void deleteById(UUID id);
}

Connection Pool Configuration

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

N+1 Query Prevention Strategy

To prevent N+1 query problems, eager loading is implemented for conversation metadata:

public class MessageWithMetadata {
    private Message message;
    private ConversationMetadata conversation;
    
    // Getters and setters
}

public class MessageService {
    public List<MessageWithMetadata> getMessagesWithConversationMetadata(
            UUID conversationId, Pageable pageable) {
        // Single query to fetch messages with metadata
        String sql = """
            SELECT m.*, c.title as conversation_title, c.participant_count
            FROM messages m
            JOIN conversations c ON m.conversation_id = c.id
            WHERE m.conversation_id = ?
            ORDER BY m.created_at ASC
            LIMIT ? OFFSET ?
            """;
            
        return jdbcTemplate.query(sql, new MessageWithMetadataRowMapper(), 
            conversationId, pageable.getPageSize(), pageable.getOffset());
    }
}

Batch Operation Design

Batch operations are implemented for bulk message creation:

@Repository
public class MessageBatchRepository {
    
    public void saveBatch(List<Message> messages) {
        String sql = """
            INSERT INTO messages (id, conversation_id, sender_id, content, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?)
            """;
            
        jdbcTemplate.batchUpdate(sql, messages, messages.size(), 
            (ps, message) -> {
                ps.setObject(1, message.getId());
                ps.setObject(2, message.getConversationId());
                ps.setString(3, message.getSenderId().toString());
                ps.setString(4, message.getContent());
                ps.setTimestamp(5, Timestamp.valueOf(message.getCreatedAt()));
                ps.setTimestamp(6, Timestamp.valueOf(message.getUpdatedAt()));
            });
    }
}

Transaction Management

All repository operations are wrapped in proper transaction boundaries:

@Service
@Transactional
public class MessageService {
    
    public Message createMessage(Message message) {
        // Transaction boundary starts here
        return messageRepository.save(message);
        // Transaction commits here
    }
    
    @Transactional(readOnly = true)
    public Page<Message> getMessagesByConversation(UUID conversationId, Pageable pageable) {
        return new PageImpl<>(
            messageRepository.findByConversationId(conversationId, pageable),
            pageable,
            messageRepository.countByConversationId(conversationId)
        );
    }
}

Query Optimization

Indexing is crucial for performance:

-- Primary index on ID
CREATE INDEX idx_messages_id ON messages(id);

-- Composite index for conversation queries
CREATE INDEX idx_messages_conversation_created ON messages(conversation_id, created_at);

-- Unique constraint for message uniqueness
ALTER TABLE messages ADD CONSTRAINT uk_message_conversation_sender 
    UNIQUE (conversation_id, sender_id, content);

Data Access Layer Testing Strategy

@ExtendWith(MockitoExtension.class)
class MessageRepositoryImplTest {
    
    @Mock
    private JdbcTemplate jdbcTemplate;
    
    @InjectMocks
    private MessageRepositoryImpl repository;
    
    @Test
    void shouldSaveMessage() {
        // Given
        Message message = createSampleMessage();
        
        // When
        repository.save(message);
        
        // Then
        verify(jdbcTemplate).update(anyString(), any(UUID.class), any(UUID.class), 
            any(String.class), any(String.class), any(Timestamp.class), any(Timestamp.class));
    }
    
    @Test
    void shouldFindMessageById() {
        // Given
        UUID messageId = UUID.randomUUID();
        
        // When
        Optional<Message> result = repository.findById(messageId);
        
        // Then
        verify(jdbcTemplate).queryForObject(anyString(), any(RowMapper.class), any(UUID.class));
    }
}

9. Asynchronous Processing & Messaging

Event-driven boundaries in this feature

The messaging system follows a hybrid synchronous/asynchronous approach. Core operations like message sending and delivery are handled synchronously to ensure immediate acknowledgment. Asynchronous processing is used for:

1. Message persistence and indexing - After successful message creation, we asynchronously index the message for search capabilities
2. Notification delivery - Sending push notifications or email confirmations to users after message receipt
3. Analytics tracking - Aggregating usage metrics and behavioral data without blocking the primary flow
4. Message archival - Moving older messages to cold storage for cost optimization

This approach ensures that the user experience remains fast and responsive while allowing background processing to handle non-critical operations efficiently.

Message schema design

{
  "version": "1.0",
  "type": "message.created",
  "timestamp": "2023-11-15T14:30:00Z",
  "source": "messaging-service",
  "payload": {
    "messageId": "msg-1234567890",
    "conversationId": "conv-0987654321",
    "senderId": "user-123",
    "recipientId": "user-456",
    "content": "Hello there!",
    "timestamp": "2023-11-15T14:30:00Z",
    "status": "delivered"
  }
}

Versioning is handled through the version field in the message schema, with semantic versioning strategy to ensure backward compatibility during migration.

Dead letter queue strategy and poison message handling

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

1. Retry limit: Maximum 3 retries with exponential backoff
2. DLQ routing: Messages that fail after max retries are routed to a dedicated DLQ topic
3. Monitoring: DLQ messages trigger alerts for manual investigation
4. Reprocessing: Failed messages in DLQ can be manually reprocessed after root cause analysis

At-least-once vs. exactly-once delivery

At-least-once delivery is chosen for this messaging system because:

1. User experience: Receiving duplicate messages is less disruptive than missing important ones
2. Simpler implementation: Easier to reason about and maintain compared to exactly-once semantics
3. Business requirements: Message duplication is acceptable for chat and communication systems

The system uses idempotency keys to handle duplicate messages gracefully.

Retry policy for consumers

public class MessageRetryPolicy {
    private static final int MAX_RETRIES = 3;
    private static final long BASE_DELAY_MS = 1000;
    private static final double BACKOFF_MULTIPLIER = 2.0;
    private static final double JITTER_RANGE = 0.1; // 10% jitter
    
    public RetryResult executeWithRetry(Supplier<CompletableFuture<Boolean>> operation) {
        int attempts = 0;
        long delayMs = BASE_DELAY_MS;
        
        while (attempts < MAX_RETRIES) {
            try {
                CompletableFuture<Boolean> result = operation.get();
                if (result != null && result.get()) {
                    return RetryResult.SUCCESS;
                }
            } catch (Exception e) {
                if (shouldRetry(e)) {
                    attempts++;
                    if (attempts < MAX_RETRIES) {
                        long jitter = (long) (delayMs * JITTER_RANGE * (Math.random() - 0.5) * 2);
                        long actualDelay = delayMs + jitter;
                        Thread.sleep(actualDelay);
                        delayMs *= BACKOFF_MULTIPLIER;
                    } else {
                        return RetryResult.FAILURE;
                    }
                } else {
                    return RetryResult.NON_RETRYABLE_ERROR;
                }
            }
        }
        return RetryResult.FAILURE;
    }
    
    private boolean shouldRetry(Exception exception) {
        // Retry on network issues, timeout, and transient errors
        return exception instanceof IOException || 
               exception instanceof TimeoutException ||
               (exception.getCause() != null && 
                shouldRetry(exception.getCause()));
    }
}

Outbox pattern for reliable event publishing

The outbox pattern is implemented to ensure message delivery reliability:

public class MessageOutboxService {
    private final MessageRepository messageRepository;
    private final EventPublisher eventPublisher;
    
    public void sendMessage(Message message) {
        // 1. Save message to database
        Message savedMessage = messageRepository.save(message);
        
        // 2. Create outbox entry
        OutboxEvent outboxEvent = new OutboxEvent(
            savedMessage.getId(),
            "message.created",
            savedMessage.toJson(),
            System.currentTimeMillis()
        );
        
        // 3. Save to outbox table (transactionally)
        outboxRepository.save(outboxEvent);
        
        // 4. Publish to message queue (in same transaction)
        eventPublisher.publish(outboxEvent);
    }
}

Sequence diagram for async flows

sequenceDiagram
    participant Client
    participant MessagingService
    participant Database
    participant OutboxRepository
    participant MessageQueue
    participant NotificationService
    
    Client->>MessagingService: Send Message Request
    MessagingService->>Database: Save Message
    MessagingService->>OutboxRepository: Create Outbox Entry
    MessagingService->>MessageQueue: Publish Event
    MessageQueue-->>NotificationService: Deliver to Consumer
    
    Note over NotificationService: Async processing
    NotificationService->>Database: Update User Status
    NotificationService->>Database: Log Analytics
    
    Note over MessagingService: Transaction committed
    MessagingService-->>Client: Acknowledge Success

10. Error Handling Design

Error Classification Hierarchy

The error handling strategy follows a three-tier classification system to enable appropriate response handling:

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

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

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

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

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

Error Propagation Strategy

Errors are handled at the service layer and propagated upward through the call stack. For external API calls, errors are wrapped in infrastructure errors to maintain consistent error handling patterns.

Retry Strategy with Exponential Backoff and Jitter

For transient infrastructure errors, a retry mechanism is implemented using exponential backoff with jitter to prevent thundering herd:

public class RetryConfig {
    private static final int MAX_RETRIES = 3;
    private static final long INITIAL_DELAY_MS = 1000;
    private static final double JITTER_RANGE = 0.5;
    
    public static long calculateDelay(int attempt) {
        long delay = (long) (INITIAL_DELAY_MS * Math.pow(2, attempt));
        double jitter = JITTER_RANGE * (Math.random() - 0.5);
        return Math.max(100, delay + (long) (delay * jitter));
    }
    
    public static boolean shouldRetry(Exception ex, int attempt) {
        return attempt < MAX_RETRIES && 
               (ex instanceof InfrastructureError || ex instanceof TimeoutException);
    }
}

Circuit Breaker Design

The circuit breaker pattern prevents cascading failures when downstream services are unavailable:

public class CircuitBreaker {
    private final String name;
    private volatile CircuitState state = CircuitState.CLOSED;
    private int failureCount = 0;
    private long lastFailureTime = 0;
    private final int failureThreshold = 5;
    private final long timeoutMs = 30000; // 30 seconds
    
    public enum CircuitState {
        CLOSED, OPEN, HALF_OPEN
    }
    
    public boolean canExecute() {
        switch (state) {
            case CLOSED:
                return true;
            case HALF_OPEN:
                return failureCount == 0; // Allow one request to test
            case OPEN:
                if (System.currentTimeMillis() - lastFailureTime > timeoutMs) {
                    state = CircuitState.HALF_OPEN;
                    return true;
                }
                return false;
            default:
                return false;
        }
    }
    
    public void recordSuccess() {
        if (state == CircuitState.HALF_OPEN) {
            state = CircuitState.CLOSED;
        }
        failureCount = 0;
    }
    
    public void recordFailure() {
        failureCount++;
        lastFailureTime = System.currentTimeMillis();
        
        if (failureCount >= failureThreshold) {
            state = CircuitState.OPEN;
        }
    }
}

Fallback Behavior Design

Fallbacks are implemented for critical dependencies to maintain system availability:

public class FallbackStrategy {
    public static String getFallbackMessage(String originalMessage) {
        return "System temporarily unavailable. Please try again later.";
    }
    
    public static List<Message> getFallbackMessages(List<Message> original) {
        return Collections.emptyList();
    }
}

Error Logging Format and Alerting Thresholds

Structured logging includes trace context, user ID, and error details:

public class ErrorLogger {
    private static final Logger logger = LoggerFactory.getLogger(ErrorLogger.class);
    
    public static void logError(String traceId, String userId, String operation, 
                               Exception ex, Map<String, Object> additionalContext) {
        Map<String, Object> logContext = new HashMap<>();
        logContext.put("trace_id", traceId);
        logContext.put("user_id", userId);
        logContext.put("operation", operation);
        logContext.put("error_code", extractErrorCode(ex));
        logContext.put("error_message", ex.getMessage());
        logContext.put("stack_trace", ex.getStackTrace());
        logContext.putAll(additionalContext);
        
        logger.error("Messaging system error", new RuntimeException(ex.getMessage(), ex));
    }
    
    private static String extractErrorCode(Exception ex) {
        if (ex instanceof MessagingError) {
            return ((MessagingError) ex).getErrorCode();
        }
        return "UNKNOWN_ERROR";
    }
}

Alerting Thresholds

Critical alerts are configured for:

• Rate of infrastructure errors: >5% of total operations within 5 minutes
• Consecutive circuit breaker opens: >3 consecutive failures
• High latency errors: >95th percentile response times exceeding 5 seconds

These thresholds are monitored via Prometheus metrics and trigger alerts through PagerDuty for on-call escalation.

11. Security Design

Input Validation and Sanitization Strategy

All incoming requests are validated using a combination of:

• Library: Apache Commons Validator for basic validation (email, URL, phone number)
• Custom rules: Regex patterns for message content length (max 10k chars), allowed characters (alphanumeric + spaces + basic punctuation)
• Sanitization: HTML escaping using OWASP Java HTML Sanitizer for message content to prevent XSS attacks

public class MessageValidator {
    private static final int MAX_MESSAGE_LENGTH = 10000;
    
    public ValidationResult validateMessage(String content) {
        if (content == null || content.trim().isEmpty()) {
            return ValidationResult.error("Message content cannot be empty");
        }
        
        if (content.length() > MAX_MESSAGE_LENGTH) {
            return ValidationResult.error("Message exceeds maximum length of " + MAX_MESSAGE_LENGTH);
        }
        
        // Validate content against allowed characters (basic whitelist)
        if (!content.matches("^[a-zA-Z0-9\\s\\p{P}]+$")) {
            return ValidationResult.error("Message contains invalid characters");
        }
        
        return ValidationResult.success();
    }
}

SQL Injection Prevention

• ORM usage: All database operations use JPA/Hibernate with parameterized queries
• No dynamic SQL construction - all queries are built using JPQL or Criteria API
• Input escaping: All user inputs are properly parameterized in prepared statements

Authentication Middleware Design

• JWT Bearer Token: Authentication handled via JWT tokens with 24-hour expiration
• Token Refresh: Refresh tokens with 7-day validity for secure token rotation
• Middleware: Custom Spring Security filter that validates JWT and extracts user claims

@Component
public class JwtAuthenticationFilter extends OncePerRequestFilter {
    
    @Autowired
    private JwtTokenProvider tokenProvider;
    
    @Override
    protected void doFilterInternal(HttpServletRequest request, 
                                   HttpServletResponse response, 
                                   FilterChain filterChain) throws ServletException, IOException {
        
        String token = extractToken(request);
        if (token != null && tokenProvider.validateToken(token)) {
            Authentication auth = tokenProvider.getAuthentication(token);
            SecurityContextHolder.getContext().setAuthentication(auth);
        }
        
        filterChain.doFilter(request, response);
    }
}

Authorization Enforcement

• RBAC Implementation: Role-based access control with roles: USER, ADMIN, MODERATOR
• Permission Checking: Service-level authorization using Spring Security annotations
• Resource Scoping: Users can only access messages they own or are part of the conversation

@PreAuthorize("@messageAuthorizationService.canAccessMessage(principal, messageId)")
public Message getMessage(String messageId) {
    return messageRepository.findById(messageId).orElseThrow(() -> new MessageNotFoundException(messageId));
}

@PreAuthorize("@messageAuthorizationService.canModifyMessage(principal, message)")
public void updateMessage(String messageId, MessageUpdateRequest request) {
    Message message = messageRepository.findById(messageId).orElseThrow();
    // Update logic here
}

Sensitive Data Encryption at Rest and in Transit

• At Rest: AES-256 encryption using AWS KMS for message content storage
• In Transit: TLS 1.3 with HSTS header enforcement (HTTPS only)
• Key Management: Centralized key management via AWS KMS with automatic rotation

PII Handling and Masking Strategy

• PII Detection: Automated detection of phone numbers, email addresses, SSNs in messages
• Masking: PII fields are masked in logs and responses (e.g., "123-456-7890" → "XXX-XXX-7890")
• Audit: All PII handling is logged with audit trail

Secrets Management

• Vault Integration: AWS Secrets Manager for storing API keys, database credentials, and encryption keys
• Rotation: Automatic secret rotation every 90 days with seamless failover
• Access Control: Principle of least privilege - only necessary services have access to specific secrets

Audit Logging Design

• Events Logged: Message creation, modification, deletion, read access, user joins/leaves
• Format: Structured JSON with fields: timestamp, userId, action, resourceId, ip, userAgent
• Retention: 90-day retention with archiving to S3 for compliance

{
  "timestamp": "2023-10-15T14:30:00Z",
  "userId": "user_123",
  "action": "MESSAGE_CREATE",
  "resourceId": "msg_456",
  "ip": "192.168.1.100",
  "userAgent": "Mozilla/5.0...",
  "details": {
    "messageContent": "XXX-XXX-7890"
  }
}

12. Testing Strategy

Unit Testing Scope and Coverage Targets (%)

• Service Layer: 85% code coverage with focus on business logic, error handling, and boundary conditions
• Repository Layer: 90% coverage with all CRUD operations and query edge cases
• Domain Objects: 100% coverage for value objects and immutable entities
• Event Handlers: 80% coverage with integration points mocked
• Security Middleware: 95% coverage for authentication and authorization logic
• Critical Path Operations: 100% coverage with all state transitions and error paths

Integration Testing Strategy

• Database Integration: Use Testcontainers with PostgreSQL 13+ for end-to-end database tests
• External Service Mocks: WireMock for third-party API integrations (e.g., payment providers, notification services)
• Message Broker Integration: Embedded Kafka or RabbitMQ for async flow testing
• Authentication Integration: In-memory JWT provider with test key rotation
• Caching Layer: Redis Testcontainers for cache behavior verification

Contract Testing

• API Contract Testing: Pact for consumer-driven contract testing between messaging service and downstream consumers
• Event Schema Contracts: Use JSON Schema or Avro schema registry with contract validation tests
• Version Compatibility: Test backward compatibility for API and event schema versions

Performance Testing Tool and Key Scenarios

• Tool: k6 (JavaScript-based load testing) with Grafana for visualization
• Key Scenarios:
  • High concurrent message publishing (1000+ messages/sec)
  • Message queue throughput under burst load (5000 msg/sec)
  • Latency for message delivery (p99 < 200ms)
  • Concurrent consumer processing (100+ concurrent consumers)
  • Message retention and archival under sustained load

Test Data Management Strategy

• Fixtures: YAML-based fixtures for common test data (users, channels, permissions)
• Factories: Java-based factory classes with configurable parameters for test data generation
• Seeding: Database seed scripts for initial test environment setup with deterministic IDs
• Isolation: Per-test database schema or transaction rollback for clean test state
• Data Generation: Faker libraries for realistic but non-sensitive test data (names, emails, content)

13. Observability Design

Structured Logging Schema

All logs follow a structured JSON format with consistent fields to enable automated parsing and analysis:

{
  "timestamp": "2023-10-27T14:30:45.123Z",
  "level": "INFO",
  "service": "messaging-service",
  "trace_id": "12345678-1234-1234-1234-123456789012",
  "span_id": "abcdef12-3456-7890-abcd-ef1234567890",
  "user_id": "user_12345",
  "message_id": "msg_abcde12345",
  "operation": "sendMessage",
  "message_type": "notification",
  "status": "success",
  "duration_ms": 150,
  "error_code": null,
  "error_message": null,
  "payload_size_bytes": 1024
}

PII masking rules:

• User IDs are logged as user_XXXXX (masked)
• Email addresses are masked to ***@***.com
• Phone numbers are masked to ***-***-****

Distributed Tracing Instrumentation

Using OpenTelemetry with custom spans for key operations:

// Example: Tracing in sendMessage method
Span span = tracer.spanBuilder("sendMessage")
    .setParent(parentContext)
    .startSpan();

try (Scope scope = span.makeCurrent()) {
    // Operation logic here
    span.setAttribute("message.id", message.getId());
    span.setAttribute("message.type", message.getType());
    span.setAttribute("user.id", message.getRecipientId());
    
    // Publish to queue
    messagingPublisher.publish(message);
    
    span.setStatus(StatusCode.OK);
} catch (Exception e) {
    span.recordException(e);
    span.setStatus(StatusCode.ERROR);
    throw e;
} finally {
    span.end();
}

Key spans to capture:

• sendMessage - End-to-end message sending flow
• processMessage - Message consumption and processing
• validateMessage - Input validation logic
• publishToQueue - Queue publishing operation

Metrics Design

RED Method Metrics (Rate, Errors, Duration):

Metric Name	Type	Description	Labels
messaging_messages_sent_total	Counter	Total messages sent	message_type, status
messaging_messages_processed_total	Counter	Total messages processed	message_type, status
messaging_messages_failed_total	Counter	Failed message operations	message_type, error_code
messaging_message_processing_duration_seconds	Histogram	Message processing time	message_type, status
messaging_queue_size	Gauge	Current queue size	queue_name
messaging_message_retry_count	Counter	Retry attempts per message	message_type, retry_number

Health Check Endpoint Design

@RestController
@RequestMapping("/health")
public class HealthController {
    
    @GetMapping("/live")
    public ResponseEntity<Map<String, Object>> live() {
        Map<String, Object> response = new HashMap<>();
        response.put("status", "healthy");
        response.put("timestamp", Instant.now());
        return ResponseEntity.ok(response);
    }
    
    @GetMapping("/ready")
    public ResponseEntity<Map<String, Object>> ready() {
        Map<String, Object> response = new HashMap<>();
        response.put("status", "ready");
        response.put("timestamp", Instant.now());
        response.put("database", databaseHealthCheck());
        response.put("message_broker", brokerHealthCheck());
        response.put("cache", cacheHealthCheck());
        
        if (isReady()) {
            return ResponseEntity.ok(response);
        } else {
            return ResponseEntity.status(503).body(response);
        }
    }
    
    @GetMapping("/startup")
    public ResponseEntity<Map<String, Object>> startup() {
        Map<String, Object> response = new HashMap<>();
        response.put("status", "startup_complete");
        response.put("timestamp", Instant.now());
        return ResponseEntity.ok(response);
    }
}

Alerting Thresholds

Metric	Warning Level	Critical Level	Alerting Channel
messaging_messages_failed_total (rate)	> 10/minute	> 50/minute	On-call
messaging_message_processing_duration_seconds (p99)	> 500ms	> 2s	On-call
messaging_queue_size (count)	> 10,000 messages	> 50,000 messages	On-call
messaging_messages_sent_total (rate)	> 10,000/minute	> 50,000/minute	On-call
messaging_messages_processed_total (rate)	< 100/minute	< 10/minute	On-call

Alerting thresholds use sliding window analysis and trigger alerts when sustained above thresholds for 5 minutes. Critical alerts escalate to on-call engineers after 30 seconds of sustained threshold breaches.

14. Open Questions, Risks & Trade-offs

Decisions requiring product / engineering alignment

• Message persistence model: Should messages be persisted in database or in dedicated message store (e.g., Kafka, RabbitMQ)? This decision impacts latency, scalability, and operational complexity.
• Message delivery guarantee: Should we support exactly-once delivery or at-least-once delivery? This affects the complexity of the system and the consistency model.
• Cross-service messaging: Should we implement a centralized message broker or allow direct service-to-service communication? This impacts system coupling and operational overhead.
• Message format standardization: Should we use a single message format (e.g., JSON, Protobuf) or allow multiple formats? This affects interoperability and maintenance burden.
• Message versioning strategy: How do we handle breaking changes in message schemas? Should we use schema registries or backward-compatible formats?

Known trade-offs and justifications

Trade-off	Option A	Option B	Justification
Persistence vs. Performance	Store messages in database	Use in-memory queue with persistence layer	Database approach provides durability but introduces latency; in-memory offers better performance but requires additional infrastructure for persistence
Synchronous vs. Asynchronous	All messaging is synchronous	All messaging is asynchronous	Synchronous ensures immediate consistency but can cause cascading failures; asynchronous enables better scalability and resilience
Monolithic vs. Distributed	Single messaging service	Multiple microservices for different message types	Monolithic approach simplifies development and debugging but creates a single point of failure; distributed approach provides better scalability but increases complexity
Event-driven vs. Request-Response	Pure event-driven architecture	Hybrid request-response + event-driven	Event-driven provides better decoupling but can make debugging harder; hybrid approach offers flexibility but increases complexity
Message Format	JSON format	Protobuf format	JSON is human-readable and easy to debug but less efficient; Protobuf offers better performance and size efficiency but requires more tooling

Risks and mitigation strategies

• Message loss: Risk of losing messages during system failures or network issues.

  • Mitigation: Implement durable message storage with acknowledgments and dead-letter queue handling.

• Message duplication: Risk of processing the same message multiple times.

  • Mitigation: Use idempotent message handlers and implement deduplication logic.

• Performance bottlenecks: Risk of message processing becoming a bottleneck under high load.

  • Mitigation: Implement horizontal scaling, queue partitioning, and async processing.

• Security vulnerabilities: Risk of unauthorized access to message queues or content interception.

  • Mitigation: Implement proper authentication, encryption in transit and at rest, and access control policies.

• Data consistency: Risk of message ordering issues or inconsistency between systems.

  • Mitigation: Use sequence numbers, timestamps, and transactional message handling where appropriate.

Technical debt acknowledged

• Message format evolution: Future changes to message schemas may require complex migration strategies and backward compatibility handling.
• Monitoring complexity: As the messaging system grows, maintaining comprehensive observability across all message flows becomes increasingly complex.
• Cross-service dependencies: If multiple services depend on a shared messaging layer, changes to that layer can impact all dependent services.
• Eventual consistency: The asynchronous nature of messaging introduces eventual consistency concerns that must be carefully managed in business logic.
• Schema management overhead: Maintaining message schema versions and ensuring compatibility across different service versions adds operational complexity.

The design choices made in this messaging system are intended to provide a balance between performance, scalability, and maintainability while acknowledging that some trade-offs are unavoidable in any distributed system design.