Content Moderation Pipeline Design

A content moderation pipeline system that processes user-generated content through multiple layers of automated and human review to maintain platform safety and quality standards while ensuring low latency and high throughput for large-scale applications.

1. Problem Statement & Requirements

Clarifying questions

1. What types of content are we moderating? (e.g., text, images, videos, user-generated posts, comments)
2. What is the expected scale and throughput? (e.g., millions of content items per day, peak burst capacity)
3. What are the consistency and latency requirements? (e.g., real-time moderation, batch processing acceptable)
4. What is the expected read/write ratio for content and moderation decisions?
5. What are the key entities in the system and their relationships? (e.g., user, content, moderation decision, policy)
6. What authentication and authorization model is in place? (e.g., OAuth2, JWT, role-based access)
7. Are there external moderation services or APIs in scope? (e.g., AWS Rekognition, Google Cloud Vision, third-party AI models)

Functional requirements

• Accept content submissions (text, image, video) from users
• Automatically moderate content using AI/ML models and rule-based checks
• Manually review flagged content by human moderators
• Apply moderation actions (reject, flag for review, approve)
• Track moderation history and decisions
• Support content categorization and policy enforcement
• Allow policy configuration and updates
• Provide moderation dashboard for administrators

Non-functional requirements

• Latency Targets:
  • p50: < 200ms for content submission and initial moderation
  • p95: < 500ms for content submission and initial moderation
  • p99: < 1s for content submission and initial moderation
• Throughput: Handle 50,000 content items per second during peak hours
• Availability: 99.9% uptime for moderation service (99.99% for admin dashboard)
• Scalability: Support horizontal scaling across multiple regions
• Auditability: Full audit trail of moderation actions within 5 seconds

Constraints and assumptions

• Content is submitted via REST API or message queue
• Moderation decisions are made within 5 seconds of content submission
• AI models are assumed to be available via external APIs or microservices
• Moderation actions must be idempotent and support retries
• Human moderation is performed asynchronously with notifications
• Content can be stored in a distributed file system or cloud storage (e.g., S3)
• Moderation policies can be updated without downtime

Out of scope

• Content generation or creation (user posts, comments, etc.)
• User management and authentication (handled by separate services)
• Real-time content streaming or live moderation
• Deep learning model training or retraining
• Content recommendation or personalization features
• Multi-language content translation or localization
• Moderation of third-party content (e.g., links to external sites)
• Content archiving or long-term storage policies

2. API Design

API Style Selection

For content moderation pipeline, I'll select gRPC as the primary API style:

• Justification: gRPC provides strong typing, efficient binary serialization, and built-in support for streaming. It's ideal for real-time content moderation workflows where low latency and high throughput are critical.
• Advantages:
  • Strong contract enforcement via Protocol Buffers
  • Built-in support for bidirectional streaming (real-time moderation)
  • Efficient binary encoding reduces payload size
  • First-class support for distributed tracing and metrics
• Trade-offs:
  • More complex client implementation compared to REST
  • Less human-readable than REST for debugging

Endpoint Definitions

Method	Path	Description	Auth	Rate Limit
SubmitContent	/moderation.v1.ContentModerationService/SubmitContent	Submit content for moderation review	JWT	100/hour
GetContentStatus	/moderation.v1.ContentModerationService/GetContentStatus	Retrieve moderation status for content	JWT	1000/hour
ApproveContent	/moderation.v1.ContentModerationService/ApproveContent	Approve content manually	JWT + Admin Role	50/hour
RejectContent	/moderation.v1.ContentModerationService/RejectContent	Reject content manually	JWT + Admin Role	50/hour
GetModerationQueue	/moderation.v1.ContentModerationService/GetModerationQueue	Fetch content awaiting moderation	JWT + Admin Role	20/hour
GetModerationReport	/moderation.v1.ContentModerationService/GetModerationReport	Generate moderation statistics report	JWT + Admin Role	5/hour

Request and Response Schemas

SubmitContent Request:

message SubmitContentRequest {
  string content_id = 1;
  string content_type = 2; // "text", "image", "video"
  string content_data = 3;
  string user_id = 4;
  map<string, string> metadata = 5;
  int32 priority = 6; // 1-10 scale
}

message SubmitContentResponse {
  string content_id = 1;
  string status = 2; // "pending", "approved", "rejected"
  string moderation_id = 3;
  int64 created_at = 4;
}

GetContentStatus Response:

message GetContentStatusResponse {
  string content_id = 1;
  string status = 2; // "pending", "approved", "rejected", "failed"
  string moderation_id = 3;
  string reason = 4; // rejection reason if applicable
  int64 created_at = 5;
  int64 updated_at = 6;
  repeated ModerationResult results = 7;
}

message ModerationResult {
  string engine_name = 1;
  string status = 2; // "pass", "fail", "error"
  double score = 3;
  string reason = 4;
}

HTTP Status Codes and Error Response Format

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

Standard error response format:

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Content type must be one of [text, image, video]",
    "details": [
      {
        "field": "content_type",
        "message": "Invalid content type"
      }
    ]
  }
}

API Versioning Strategy

Versioned using gRPC package naming:

• moderation.v1 - Current stable version
• Future: moderation.v2 for breaking changes

Pagination Design

Cursor-based pagination selected for:

• Efficient handling of large moderation queues
• Consistent results despite concurrent modifications
• Better performance for large datasets

Authentication and Authorization

Authentication: JWT Bearer tokens with audience validation Authorization:

• All endpoints require valid JWT token
• Admin endpoints require moderation.admin scope
• Content submission requires moderation.submit scope

Rate Limiting Strategy

Endpoint	Limit	Period
SubmitContent	100	Hour
GetContentStatus	1000	Hour
ApproveContent/RejectContent	50	Hour
GetModerationQueue	20	Hour
GetModerationReport	5	Hour

Rate limiting enforced at API gateway level using Redis-based token bucket algorithm.

Idempotency Design

For mutation endpoints, idempotency key strategy:

// Idempotency key format: "moderation:{user_id}:{content_id}:{timestamp}"
public class IdempotencyKey {
    private String key;
    
    public static String generateKey(String userId, String contentId, long timestamp) {
        return "moderation:" + userId + ":" + contentId + ":" + timestamp;
    }
}

Each request includes X-Idempotency-Key header that is stored in Redis with TTL of 1 hour. If same key exists, return cached response instead of reprocessing.

// Example usage in gRPC service method
public SubmitContentResponse submitContent(SubmitContentRequest request) {
    String idempotencyKey = extractIdempotencyKey();
    
    // Check if this request was already processed
    String cachedResponse = redisTemplate.get(idempotencyKey);
    if (cachedResponse != null) {
        return parseCachedResponse(cachedResponse);
    }
    
    // Process content moderation...
    SubmitContentResponse response = processContent(request);
    
    // Cache the result for idempotency
    redisTemplate.setex(idempotencyKey, response, Duration.ofHours(1));
    
    return response;
}

3. Data Model Design

Entity-Relationship Diagram

erDiagram
    USER ||--o{ CONTENT : "creates"
    USER ||--o{ MODERATION_DECISION : "makes"
    CONTENT ||--o{ MODERATION_DECISION : "has"
    MODERATION_DECISION ||--o{ MODERATION_ACTION : "triggers"
    CONTENT ||--o{ REPORT : "gets_reported"
    REPORT ||--o{ MODERATION_DECISION : "leads_to"
    MODERATION_ACTION ||--o{ CONTENT : "affects"
    
    USER {
        string id PK
        string email
        string username
        string role
        timestamp created_at
        timestamp updated_at
    }
    
    CONTENT {
        string id PK
        string type
        string content
        string author_id FK
        timestamp created_at
        timestamp updated_at
        boolean is_deleted
    }
    
    MODERATION_DECISION {
        string id PK
        string content_id FK
        string decision_type
        string status
        string reason
        string reviewer_id FK
        timestamp created_at
        timestamp updated_at
    }
    
    MODERATION_ACTION {
        string id PK
        string decision_id FK
        string action_type
        timestamp executed_at
        string executed_by_id FK
    }
    
    REPORT {
        string id PK
        string content_id FK
        string reporter_id FK
        string reason
        timestamp created_at
        timestamp updated_at
    }

Table Schema Definitions

Users Table

CREATE TABLE users (
    id VARCHAR(36) PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    username VARCHAR(100) UNIQUE NOT NULL,
    role VARCHAR(50) NOT NULL DEFAULT 'user',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    
    INDEX idx_email (email),
    INDEX idx_username (username)
);

Content Table

CREATE TABLE content (
    id VARCHAR(36) PRIMARY KEY,
    type VARCHAR(50) NOT NULL,
    content TEXT NOT NULL,
    author_id VARCHAR(36) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    is_deleted BOOLEAN DEFAULT FALSE,
    
    INDEX idx_author_id (author_id),
    INDEX idx_type (type),
    INDEX idx_created_at (created_at),
    INDEX idx_is_deleted (is_deleted),
    
    FOREIGN KEY (author_id) REFERENCES users(id)
);

Reports Table

CREATE TABLE reports (
    id VARCHAR(36) PRIMARY KEY,
    content_id VARCHAR(36) NOT NULL,
    reporter_id VARCHAR(36) NOT NULL,
    reason TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    
    INDEX idx_content_id (content_id),
    INDEX idx_reporter_id (reporter_id),
    INDEX idx_created_at (created_at),
    
    FOREIGN KEY (content_id) REFERENCES content(id) ON DELETE CASCADE,
    FOREIGN KEY (reporter_id) REFERENCES users(id)
);

Moderation Decisions Table

CREATE TABLE moderation_decisions (
    id VARCHAR(36) PRIMARY KEY,
    content_id VARCHAR(36) NOT NULL,
    decision_type VARCHAR(50) NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'pending',
    reason TEXT,
    reviewer_id VARCHAR(36),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    
    INDEX idx_content_id (content_id),
    INDEX idx_status (status),
    INDEX idx_decision_type (decision_type),
    INDEX idx_reviewer_id (reviewer_id),
    
    FOREIGN KEY (content_id) REFERENCES content(id) ON DELETE CASCADE,
    FOREIGN KEY (reviewer_id) REFERENCES users(id)
);

Moderation Actions Table

CREATE TABLE moderation_actions (
    id VARCHAR(36) PRIMARY KEY,
    decision_id VARCHAR(36) NOT NULL,
    action_type VARCHAR(50) NOT NULL,
    executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    executed_by_id VARCHAR(36),
    
    INDEX idx_decision_id (decision_id),
    INDEX idx_action_type (action_type),
    INDEX idx_executed_at (executed_at),
    
    FOREIGN KEY (decision_id) REFERENCES moderation_decisions(id) ON DELETE CASCADE,
    FOREIGN KEY (executed_by_id) REFERENCES users(id)
);

Primary Key Strategy

UUIDv4 is used for all primary keys to:

• Enable distributed system compatibility
• Prevent enumeration attacks
• Support sharding by content type or user ID
• Avoid sequential ID patterns that might leak information

Index Design Justification

1. Content table:

   • author_id: For content retrieval by user
   • type: For filtering content by type (text, image, video)
   • created_at: For chronological queries and archiving
   • is_deleted: For soft-delete filtering

2. Reports table:

   • content_id: For report aggregation and content lookup
   • reporter_id: For user report history

3. Moderation Decisions table:

   • content_id: For content decision lookup
   • status: For status-based filtering (pending, reviewed)
   • decision_type: For content-type specific moderation rules
   • reviewer_id: For reviewer performance tracking

Data Partitioning Strategy

Partition content by type:

• text_content: For text-based content (posts, comments)
• image_content: For image-based content (images, thumbnails)
• video_content: For video-based content (videos, streams)

Each partition will have:

• Separate tables for content and related metadata
• Sharding key: type + author_id hash
• Read replicas for high-availability

Soft Delete vs. Hard Delete

Soft delete is implemented:

• Content is marked is_deleted = TRUE instead of physical deletion
• Maintains referential integrity for reports and moderation history
• Enables content recovery if needed
• Supports audit trails and compliance requirements

Audit Trail Design

All modification operations log to a separate audit_log table:

CREATE TABLE audit_log (
    id VARCHAR(36) PRIMARY KEY,
    operation_type VARCHAR(20) NOT NULL,
    table_name VARCHAR(50) NOT NULL,
    record_id VARCHAR(36) NOT NULL,
    user_id VARCHAR(36),
    changes JSON,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Data Archival Strategy

Content older than 2 years will be archived to a separate storage system (S3 or similar) with:

• Metadata preserved in database for search and retrieval
• Content moved to cold storage (e.g., Glacier)
• Archive retention policy: 5 years for compliance
• Archival job runs daily at 2 AM UTC

4. Class Design & Service Layer

UML Class Diagram

classDiagram
    class ContentModerationService {
        +moderateContent(ContentRequest) ContentResponse
        +processContent(Content) ModerationResult
    }
    
    class ContentProcessor {
        +process(Content) ModerationResult
    }
    
    class ContentFilter {
        +applyFilters(Content) FilterResult
    }
    
    class ContentValidator {
        +validateContent(Content) ValidationResult
    }
    
    class ModerationRule {
        +id: UUID
        +name: String
        +description: String
        +conditions: List~FilterCondition~
        +action: ModerationAction
    }
    
    class FilterCondition {
        +type: FilterType
        +value: String
        +operator: Operator
    }
    
    class Content {
        +id: UUID
        +type: ContentType
        +data: String
        +metadata: Map~String, Object~
        +createdAt: Instant
        +status: ContentStatus
    }
    
    class ModerationResult {
        +contentId: UUID
        +status: ModerationStatus
        +violations: List~Violation~
        +actionTaken: ModerationAction
        +processedAt: Instant
    }
    
    class Violation {
        +ruleId: UUID
        +ruleName: String
        +severity: Severity
        +description: String
    }
    
    class ContentRequest {
        +content: Content
        +requestId: String
        +priority: Priority
    }
    
    class ContentResponse {
        +requestId: String
        +result: ModerationResult
        +timestamp: Instant
    }
    
    class FilterEngine {
        +applyRules(Content, List~ModerationRule~) ModerationResult
    }
    
    class RuleEngine {
        +loadRules() List~ModerationRule~
        +evaluateRule(ModerationRule, Content) FilterResult
    }
    
    class EventPublisher {
        +publishModerationEvent(ModerationResult) void
    }
    
    ContentModerationService --> ContentProcessor : uses
    ContentProcessor --> FilterEngine : uses
    FilterEngine --> RuleEngine : uses
    FilterEngine --> ContentFilter : uses
    ContentFilter --> ContentValidator : uses
    RuleEngine --> ModerationRule : uses
    ContentModerationService --> EventPublisher : publishes
    
    class ContentStatus {
        +PENDING
        +MODERATED
        +REJECTED
        +APPROVED
    }
    
    class ModerationStatus {
        +PENDING
        +APPROVED
        +REJECTED
        +NEEDS_REVIEW
    }
    
    class ModerationAction {
        +APPROVE
        +REJECT
        +FLAG_FOR_REVIEW
        +MODIFY
    }
    
    class FilterType {
        +TEXT_MATCH
        +IMAGE_CONTENT
        +VIDEO_CONTENT
        +METADATA_CHECK
    }
    
    class Severity {
        +LOW
        +MEDIUM
        +HIGH
        +CRITICAL
    }
    
    class Priority {
        +LOW
        +NORMAL
        +HIGH
        +CRITICAL
    }

State Machine Diagram for Content

stateDiagram-v2
    [*] --> PENDING
    PENDING --> MODERATED : Processed
    PENDING --> REJECTED : Violation Found
    PENDING --> NEEDS_REVIEW : Flagged For Review
    MODERATED --> APPROVED : Manual Review Passed
    MODERATED --> REJECTED : Manual Review Failed
    NEEDS_REVIEW --> APPROVED : Manual Review Passed
    NEEDS_REVIEW --> REJECTED : Manual Review Failed
    APPROVED --> [*]
    REJECTED --> [*]

Service Class Structure and Responsibilities

The content moderation system consists of several key service components:

1. ContentModerationService - Main entry point for content moderation requests
2. ContentProcessor - Core logic for processing content through filters and rules
3. FilterEngine - Orchestrates rule application and violation detection
4. RuleEngine - Evaluates individual moderation rules against content
5. ContentValidator - Validates content format and structure before processing
6. EventPublisher - Publishes moderation events for downstream consumers

Interface and Abstract Class Design

public interface ContentProcessor {
    ModerationResult process(Content content);
}

public interface RuleEvaluator {
    FilterResult evaluate(ModerationRule rule, Content content);
}

public abstract class BaseContentProcessor implements ContentProcessor {
    protected final RuleEngine ruleEngine;
    protected final EventPublisher eventPublisher;
    
    public BaseContentProcessor(RuleEngine ruleEngine, EventPublisher eventPublisher) {
        this.ruleEngine = ruleEngine;
        this.eventPublisher = eventPublisher;
    }
    
    @Override
    public ModerationResult process(Content content) {
        // Validate input
        ValidationResult validation = validateContent(content);
        if (!validation.isValid()) {
            return createRejectedResult(content, "Invalid content format");
        }
        
        // Apply rules
        ModerationResult result = applyRules(content);
        
        // Publish event
        eventPublisher.publishModerationEvent(result);
        
        return result;
    }
    
    protected abstract ValidationResult validateContent(Content content);
    protected abstract ModerationResult applyRules(Content content);
}

public enum ContentStatus {
    PENDING, MODERATED, REJECTED, APPROVED
}

public enum ModerationStatus {
    PENDING, APPROVED, REJECTED, NEEDS_REVIEW
}

public enum ModerationAction {
    APPROVE, REJECT, FLAG_FOR_REVIEW, MODIFY
}

Business Logic Encapsulation Patterns

Strategy Pattern Implementation

public interface FilterStrategy {
    FilterResult applyFilter(Content content, FilterCondition condition);
}

@Component
public class TextMatchFilterStrategy implements FilterStrategy {
    private final PatternCache patternCache;
    
    public FilterResult applyFilter(Content content, FilterCondition condition) {
        String contentText = extractContentText(content);
        Pattern pattern = patternCache.getOrCreatePattern(condition.getValue());
        
        if (pattern.matcher(contentText).find()) {
            return new FilterResult(true, "Text match violation", condition);
        }
        
        return new FilterResult(false, null, condition);
    }
    
    private String extractContentText(Content content) {
        // Implementation to extract text from various content types
        return content.getData();
    }
}

@Component
public class ImageContentFilterStrategy implements FilterStrategy {
    private final ImageAnalyzer imageAnalyzer;
    
    public FilterResult applyFilter(Content content, FilterCondition condition) {
        try {
            byte[] imageData = extractImageData(content);
            ImageAnalysisResult analysis = imageAnalyzer.analyze(imageData);
            
            if (analysis.hasInappropriateContent()) {
                return new FilterResult(true, "Inappropriate image content", condition);
            }
            
            return new FilterResult(false, null, condition);
        } catch (Exception e) {
            throw new ContentProcessingException("Failed to analyze image content", e);
        }
    }
    
    private byte[] extractImageData(Content content) {
        // Implementation to extract image data
        return content.getData().getBytes(StandardCharsets.UTF_8);
    }
}

Critical-Path Method Implementation

public class ContentModerationService {
    private final ContentProcessor contentProcessor;
    private final TransactionManager transactionManager;
    private final EventPublisher eventPublisher;
    
    public ContentResponse moderateContent(ContentRequest request) {
        String requestId = request.getRequestId();
        Content content = request.getContent();
        
        // Transaction boundary - all operations within this scope
        return transactionManager.executeInTransaction(() -> {
            try {
                // Validate content first
                ValidationResult validation = validateContent(content);
                if (!validation.isValid()) {
                    return createErrorResponse(requestId, "Invalid content format");
                }
                
                // Process content with proper locking to prevent race conditions
                ModerationResult result = processContentWithLocking(content, requestId);
                
                // Publish moderation event
                eventPublisher.publishModerationEvent(result);
                
                return new ContentResponse(requestId, result, Instant.now());
                
            } catch (Exception e) {
                // Log error and return appropriate response
                log.error("Failed to moderate content for request: {}", requestId, e);
                return createErrorResponse(requestId, "Processing failed");
            }
        });
    }
    
    private ModerationResult processContentWithLocking(Content content, String requestId) {
        // Implement optimistic locking with retry logic
        int maxRetries = 3;
        int attempt = 0;
        
        while (attempt < maxRetries) {
            try {
                // Acquire lock for content ID to prevent concurrent modification
                if (acquireContentLock(content.getId(), requestId)) {
                    try {
                        // Check if content is already processed
                        ModerationResult existing = findExistingResult(content.getId());
                        if (existing != null) {
                            return existing;
                        }
                        
                        // Process content through filter engine
                        ModerationResult result = contentProcessor.process(content);
                        
                        // Update content status in database
                        updateContentStatus(content.getId(), result.getStatus());
                        
                        return result;
                    } finally {
                        releaseContentLock(content.getId(), requestId);
                    }
                } else {
                    // Wait before retrying
                    Thread.sleep(100 * (attempt + 1));
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new ContentProcessingException("Interrupted while processing content", e);
            } catch (Exception e) {
                log.warn("Content processing attempt {} failed for content {}", attempt + 1, content.getId(), e);
            }
            
            attempt++;
        }
        
        throw new ContentProcessingException("Failed to process content after " + maxRetries + " attempts");
    }
    
    private boolean acquireContentLock(UUID contentId, String requestId) {
        // Implementation of distributed lock acquisition (e.g., Redis-based)
        return true; // Simplified for example
    }
    
    private void releaseContentLock(UUID contentId, String requestId) {
        // Implementation of distributed lock release
    }
    
    private ModerationResult findExistingResult(UUID contentId) {
        // Check if result already exists in cache/database
        return null; // Simplified for example
    }
    
    private void updateContentStatus(UUID contentId, ModerationStatus status) {
        // Update database with new moderation status
    }
    
    private ContentResponse createErrorResponse(String requestId, String message) {
        return new ContentResponse(
            requestId,
            new ModerationResult(
                null, 
                ModerationStatus.REJECTED, 
                Collections.emptyList(), 
                ModerationAction.REJECT,
                Instant.now()
            ),
            Instant.now()
        );
    }
}

Transaction Boundary Design

The transaction boundary encompasses:

1. Content validation and format checking
2. Rule evaluation and violation detection
3. Status update in database
4. Event publishing to message queue

All operations within this boundary are atomic, ensuring data consistency and preventing partial updates.

Service Interface Contracts

public interface ContentModerationService {
    ContentResponse moderateContent(ContentRequest request);
    List<ModerationResult> batchModerate(List<ContentRequest> requests);
    ModerationResult getModerationResult(UUID contentId);
    void reprocessContent(UUID contentId);
}

Saga/Orchestration Pattern for Distributed Transactions

For distributed content moderation involving external services (image analysis, text classification), we implement a saga pattern:

public class ContentModerationSaga {
    private final List<SagaStep> steps;
    
    public void execute(ContentRequest request) {
        SagaContext context = new SagaContext();
        
        for (SagaStep step : steps) {
            try {
                step.execute(context);
                context.addCompletedStep(step.getName());
            } catch (Exception e) {
                // Compensate previous steps
                compensate(context, step.getName());
                throw new ContentModerationException("Saga failed at step: " + step.getName(), e);
            }
        }
    }
    
    private void compensate(SagaContext context, String failedStep) {
        // Reverse the effects of completed steps
        List<String> stepsToCompensate = context.getCompletedSteps().stream()
            .skip(context.getCompletedSteps().indexOf(failedStep))
            .collect(Collectors.toList());
            
        // Execute compensation logic for each step
    }
}

Domain Object / Value Object Design

public class Content {
    private final UUID id;
    private final ContentType type;
    private final String data;
    private final Map<String, Object> metadata;
    private final Instant createdAt;
    private ContentStatus status;
    
    // Constructor, getters, and immutable methods
    
    public static class Builder {
        private UUID id;
        private ContentType type;
        private String data;
        private Map<String, Object> metadata = new HashMap<>();
        private Instant createdAt = Instant.now();
        private ContentStatus status = ContentStatus.PENDING;
        
        public Builder withId(UUID id) {
            this.id = id;
            return this;
        }
        
        public Builder withType(ContentType type) {
            this.type = type;
            return this;
        }
        
        public Builder withData(String data) {
            this.data = data;
            return this;
        }
        
        public Builder withMetadata(Map<String, Object> metadata) {
            this.metadata = metadata;
            return this;
        }
        
        public Content build() {
            return new Content(id, type, data, metadata, createdAt, status);
        }
    }
}

public class ModerationResult {
    private final UUID contentId;
    private final ModerationStatus status;
    private final List<Violation> violations;
    private final ModerationAction actionTaken;
    private final Instant processedAt;
    
    // Immutable fields with proper validation
    
    public static class Builder {
        private UUID contentId;
        private ModerationStatus status;
        private List<Violation> violations = new ArrayList<>();
        private ModerationAction actionTaken;
        private Instant processedAt = Instant.now();
        
        public Builder withContentId(UUID contentId) {
            this.contentId = contentId;
            return this;
        }
        
        public Builder withStatus(ModerationStatus status) {
            this.status = status;
            return this;
        }
        
        public Builder withViolations(List<Violation> violations) {
            this.violations = violations;
            return this;
        }
        
        public Builder withActionTaken(ModerationAction action) {
            this.actionTaken = action;
            return this;
        }
        
        public ModerationResult build() {
            return new ModerationResult(contentId, status, violations, actionTaken, processedAt);
        }
    }
}

public class Violation {
    private final UUID ruleId;
    private final String ruleName;
    private final Severity severity;
    private final String description;
    
    // Immutable fields with validation
    
    public static class Builder {
        private UUID ruleId;
        private String ruleName;
        private Severity severity;
        private String description;
        
        public Builder withRuleId(UUID ruleId) {
            this.ruleId = ruleId;
            return this;
        }
        
        public Builder withRuleName(String ruleName) {
:
            this.ruleName = ruleName;
            return this;
        }
        
        public Builder withSeverity(Severity severity) {
            this.severity = severity;
            return this;
        }
        
        public Builder withDescription(String description) {
            this.description = description;
            return this;
        }
        
        public Violation build() {
            return new Violation(ruleId, ruleName, severity, description);
        }
    }
}

5. Concurrency & Thread Safety

Concurrency Model

The content moderation pipeline will utilize a thread pool-based concurrency model with Java's ExecutorService for managing worker threads. The system will leverage the CompletableFuture API for asynchronous processing and coordination, enabling non-blocking operations while maintaining thread safety. For high-throughput content moderation, we'll use a fixed-size thread pool with a configurable number of worker threads (e.g., 20-50 threads) to balance resource utilization and throughput.

Race Condition Prevention Strategy

Race conditions in content moderation will be prevented through:

1. Immutable data structures for content metadata and moderation decisions
2. Synchronized access to shared mutable state using ReentrantReadWriteLock
3. Atomic operations for counters and status tracking
4. Thread-local storage for per-request context to avoid cross-contamination

Optimistic vs. Pessimistic Locking Selection

Optimistic locking will be used for content moderation operations due to:

• Low contention in moderation workflows
• High read-to-write ratio (content is read more frequently than updated)
• Better performance under concurrent access patterns
• Reduced lock acquisition overhead

Content items will use version numbers in database operations with retry logic for optimistic lock failures. For critical moderation actions (e.g., content removal), we'll implement a retry mechanism with exponential backoff.

Deadlock Prevention Strategy

Deadlock prevention mechanisms include:

1. Lock ordering - All threads acquire locks in a consistent order (e.g., by content ID hash)
2. Timeout-based acquisition - 5-second timeout on lock acquisition with automatic release
3. Hierarchical locking - Content items are locked before moderation actions
4. Thread interruption handling - Proper cleanup on thread interruption

Idempotency Implementation for Concurrent Duplicate Requests

Idempotency will be implemented using a combination of:

1. Idempotency keys in request headers (UUID-based)
2. Database-level uniqueness constraints on moderation actions
3. In-memory cache of recent idempotency keys (LRU cache with 10k capacity)
4. Database transaction isolation level set to REPEATABLE READ

public class ModerationService {
    private final Map<String, CompletableFuture<ModerationResult>> idempotencyCache = 
        new ConcurrentHashMap<>();
    private final ReadWriteLock cacheLock = new ReentrantReadWriteLock();
    
    public CompletableFuture<ModerationResult> processContentWithIdempotency(
            String contentId, 
            ModerationRequest request,
            String idempotencyKey) {
        
        return idempotencyCache.computeIfAbsent(idempotencyKey, key -> {
            // Check if already processed
            ModerationResult existing = getExistingResult(key);
            if (existing != null) {
                return CompletableFuture.completedFuture(existing);
            }
            
            // Process content with proper locking
            return processContent(contentId, request)
                .thenApply(result -> {
                    // Cache result for idempotency
                    cacheResult(key, result);
                    return result;
                });
        });
    }
    
    private CompletableFuture<ModerationResult> processContent(
            String contentId, 
            ModerationRequest request) {
        return CompletableFuture.supplyAsync(() -> {
            try (Transaction tx = transactionManager.begin()) {
                // Acquire content lock
                acquireContentLock(contentId);
                
                // Process moderation logic
                ModerationResult result = moderateContent(request);
                
                // Store result
                storeModerationResult(contentId, result);
                
                tx.commit();
                return result;
            } catch (Exception e) {
                throw new RuntimeException("Content moderation failed", e);
            }
        });
    }
}

The implementation ensures that concurrent duplicate requests for the same content with identical idempotency keys return the same result without reprocessing, while maintaining thread safety and preventing race conditions in concurrent content moderation workflows.

6. Sequence Diagrams for Key Flows

sequenceDiagram
    participant U as User
    participant API as ContentAPI
    participant Queue as MessageQueue
    participant Worker as ModerationWorker
    participant DB as Database
    participant ML as MLService
    participant Review as ReviewService

    U->>API: SubmitContent(content, metadata)
    API->>DB: StoreInitialContent(contentId, status="pending")
    API->>Queue: PublishContentEvent(contentId)
    Queue-->>Worker: ContentEvent(contentId)
    Worker->>ML: AnalyzeContent(contentId)
    ML-->>Worker: AnalysisResult(score, flags)
    alt Content flagged
        Worker->>DB: UpdateContentStatus(contentId, status="flagged")
        Worker->>Review: NotifyModerator(contentId)
        Review-->>Worker: ReviewDecision(decision, reason)
        Worker->>DB: UpdateContentStatus(contentId, status="moderated")
    else Content acceptable
        Worker->>DB: UpdateContentStatus(contentId, status="approved")
    end
    Worker->>Queue: PublishContentApprovedEvent(contentId)

sequenceDiagram
    participant U as User
    participant API as ContentAPI
    participant Queue as MessageQueue
    participant Worker as ModerationWorker
    participant DB as Database
    participant ML as MLService
    participant Review as ReviewService

    U->>API: SubmitContent(content, metadata)
    API->>DB: StoreInitialContent(contentId, status="pending")
    API->>Queue: PublishContentEvent(contentId)
    Queue-->>Worker: ContentEvent(contentId)
    Worker->>ML: AnalyzeContent(contentId)
    ML-->>Worker: AnalysisResult(score, flags)
    Worker->>DB: UpdateContentStatus(contentId, status="flagged")
    Worker->>Review: NotifyModerator(contentId)
    Review-->>Worker: ReviewDecision(decision, reason)
    alt Review decision is rejection
        Worker->>DB: UpdateContentStatus(contentId, status="rejected")
        Worker->>Queue: PublishContentRejectedEvent(contentId)
    else Review decision is approval
        Worker->>DB: UpdateContentStatus(contentId, status="approved")
        Worker->>Queue: PublishContentApprovedEvent(contentId)
    end

sequenceDiagram
    participant Worker as ModerationWorker
    participant Queue as MessageQueue
    participant DB as Database
    participant ML as MLService
    participant RetryQueue as RetryQueue
    participant ErrorLogger as ErrorLogger

    Worker->>Queue: ConsumeContentEvent(contentId)
    Worker->>ML: AnalyzeContent(contentId)
    ML-->>Worker: AnalysisResult(score, flags)
    alt Analysis fails
        Worker->>ErrorLogger: LogAnalysisFailure(contentId, error)
        Worker->>RetryQueue: QueueForRetry(contentId, retryCount=0)
        Worker->>Queue: AcknowledgeEvent()
    else Analysis succeeds
        Worker->>DB: UpdateContentStatus(contentId, status="flagged")
        Worker->>Queue: PublishContentFlaggedEvent(contentId)
        Worker->>Queue: AcknowledgeEvent()
    end

7. Caching Design

Cache Strategy Selection

For content moderation pipeline, we'll use cache-aside pattern with write-through for critical moderation decisions. This approach provides:

1. Consistency guarantee: Cache and database stay in sync
2. Performance: Reduce latency for repeated moderation checks
3. Scalability: Handle high read volumes without database overload

Cache Key Design

Cache keys will follow a structured namespace pattern:

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

Example key structure:

// Moderation decision cache key
String cacheKey = String.format("moderation:decision:%s:v1", contentId);

// Content metadata cache key  
String cacheKey = String.format("moderation:metadata:%s:v1", contentId);

TTL Strategy

Data Type	TTL	Reason
Moderation Decision	1 hour	Decisions may change with policy updates
Content Metadata	24 hours	Metadata rarely changes
User Reputation	1 hour	Reputation scores can update frequently
Content Category	7 days	Categories are relatively stable
Flagged Content List	15 minutes	Recent flags need quick propagation

Cache Eviction Policy

We'll use LRU (Least Recently Used) eviction policy with:

• TTL-based expiration as primary mechanism
• Size-based limits (e.g., 10GB max memory usage)
• Priority-based eviction for critical content

Cache Stampede Prevention

To prevent cache stampedes:

1. Probabilistic early expiry: 80% TTL for cache entries
2. Distributed locks during cache miss handling
3. Cache warming for high-traffic content

public class CacheStampedePrevention {
    private static final double EARLY_EXPIRY_RATIO = 0.8;
    
    public boolean acquireLock(String key) {
        String lockKey = "lock:" + key;
        return redisTemplate.opsForValue()
            .setIfAbsent(lockKey, "locked", 
                Duration.ofSeconds(30), TimeUnit.SECONDS);
    }
    
    public void releaseLock(String key) {
        String lockKey = "lock:" + key;
        redisTemplate.delete(lockKey);
    }
}

Distributed Cache Architecture

Using Redis Cluster with:

• 3 master nodes for data distribution
• 3 replica nodes for high availability
• Redis Sentinel for automatic failover
• Sharding key pattern: {contentId} to ensure content and its metadata stay together

Cache Invalidation Strategy

We'll implement event-driven invalidation:

1. Content Update Events: Invalidate moderation decisions when content changes
2. Policy Change Events: Invalidate all cached decisions when moderation policies update
3. User Reputation Updates: Invalidate user-based cache entries
4. Manual Flag Events: Invalidate content from flagged lists

@Component
public class CacheInvalidationService {
    
    @EventListener
    public void handleContentUpdate(ContentUpdatedEvent event) {
        String key = "moderation:decision:" + event.getContentId();
        redisTemplate.delete(key);
        
        // Invalidate related metadata
        String metaKey = "moderation:metadata:" + event.getContentId();
        redisTemplate.delete(metaKey);
    }
    
    @EventListener
    public void handlePolicyUpdate(PolicyUpdatedEvent event) {
        // Invalidate all moderation decisions (broad invalidation)
        Set<String> keys = redisTemplate.keys("moderation:decision:*");
        redisTemplate.delete(keys);
    }
}

Cache Read/Write Patterns

public class ModerationCacheService {
    
    public ModerationResult getDecision(String contentId) {
        String cacheKey = "moderation:decision:" + contentId;
        
        // Try cache first
        ModerationResult cached = redisTemplate.opsForValue().get(cacheKey);
        if (cached != null) {
            return cached;
        }
        
        // Handle cache miss with distributed lock
        if (acquireLock(cacheKey)) {
            try {
                // Double-check pattern
                cached = redisTemplate.opsForValue().get(cacheKey);
                if (cached != null) {
                    return cached;
                }
                
                // Fetch from database
                ModerationResult result = moderationRepository.getDecision(contentId);
                
                // Write through to cache
                redisTemplate.opsForValue()
                    .set(cacheKey, result, Duration.ofHours(1));
                
                return result;
            } finally {
                releaseLock(cacheKey);
            }
        }
        
        // Fallback to database if lock acquisition fails
        return moderationRepository.getDecision(contentId);
    }
}

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

public interface ContentRepository {
    Content save(Content content);
    Optional<Content> findById(UUID id);
    List<Content> findByStatus(ContentStatus status, Pageable pageable);
    long countByStatus(ContentStatus status);
    void deleteById(UUID id);
}

public class ContentRepositoryImpl implements ContentRepository {
    private final JdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;
    
    @Override
    public Content save(Content content) {
        if (content.getId() == null) {
            content.setId(UUID.randomUUID());
        }
        
        String sql = """
            INSERT INTO content (id, content_type, content_data, status, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?)
            """;
            
        jdbcTemplate.update(sql, 
            content.getId(),
            content.getContentType().name(),
            objectMapper.writeValueAsString(content.getContentData()),
            content.getStatus().name(),
            content.getCreatedAt(),
            content.getUpdatedAt()
        );
        
        return content;
    }
    
    @Override
    public Optional<Content> findById(UUID id) {
        String sql = "SELECT * FROM content WHERE id = ?";
        try {
            return Optional.of(jdbcTemplate.queryForObject(sql, new ContentRowMapper(), id));
        } catch (EmptyResultDataAccessException e) {
            return Optional.empty();
        }
    }
    
    @Override
    public List<Content> findByStatus(ContentStatus status, Pageable pageable) {
        String sql = """
            SELECT * FROM content 
            WHERE status = ? 
            ORDER BY created_at DESC 
            LIMIT ? OFFSET ?
            """;
            
        return jdbcTemplate.query(sql, new ContentRowMapper(), 
            status.name(), 
            pageable.getPageSize(), 
            pageable.getOffset()
        );
    }
    
    @Override
    public long countByStatus(ContentStatus status) {
        String sql = "SELECT COUNT(*) FROM content WHERE status = ?";
        return jdbcTemplate.queryForObject(sql, Long.class, status.name());
    }
    
    @Override
    public void deleteById(UUID id) {
        String sql = "DELETE FROM content WHERE id = ?";
        jdbcTemplate.update(sql, id);
    }
}

ORM vs. Raw Query Decisions

For this content moderation system, we use raw SQL queries instead of ORM for several reasons:

1. Performance: Raw queries provide better performance for complex content analysis operations
2. Control: Fine-grained control over query execution and optimization
3. Complexity: Content moderation requires complex joins and aggregations that are better expressed in SQL
4. Maintainability: Clear separation between business logic and data access

Read vs. Write Separation

We implement CQRS pattern for content moderation to separate read and write operations:

public interface ContentWriteRepository {
    Content save(Content content);
    void updateStatus(UUID id, ContentStatus status);
    void updateModerationResult(UUID id, ModerationResult result);
}

public interface ContentReadRepository {
    Optional<Content> findById(UUID id);
    List<Content> findByStatus(ContentStatus status, Pageable pageable);
    long countByStatus(ContentStatus status);
    List<Content> findRecentContent(int limit);
}

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

We implement eager loading for content relationships and use batch operations:

public class ContentBatchRepository {
    public List<Content> findWithModerationResults(List<UUID> contentIds) {
        // Single query to fetch all content with their moderation results
        String sql = """
            SELECT c.*, m.result, m.decision, m.created_at as moderation_time
            FROM content c
            LEFT JOIN moderation_results m ON c.id = m.content_id
            WHERE c.id IN (?)
            """;
            
        // Use batch processing to avoid N+1 queries
        return jdbcTemplate.query(sql, new ContentWithModerationRowMapper(), contentIds);
    }
}

Batch Operation Design

Batch operations are implemented for bulk content moderation:

public class ContentBatchService {
    public void processBatch(List<Content> contents) {
        String sql = """
            INSERT INTO content (id, content_type, content_data, status, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?)
            """;
            
        // Batch insert using JDBC batch operations
        jdbcTemplate.batchUpdate(sql, contents, contents.size(), (ps, content) -> {
            ps.setObject(1, content.getId());
            ps.setString(2, content.getContentType().name());
            ps.setString(3, objectMapper.writeValueAsString(content.getContentData()));
            ps.setString(4, content.getStatus().name());
            ps.setTimestamp(5, Timestamp.valueOf(content.getCreatedAt()));
            ps.setTimestamp(6, Timestamp.valueOf(content.getUpdatedAt()));
        });
    }
}

Transaction Management

All repository operations are wrapped in proper transaction boundaries:

@Service
@Transactional
public class ContentModerationService {
    @Autowired
    private ContentWriteRepository contentRepository;
    
    public Content processContent(Content content) {
        // Transaction boundary includes all related database operations
        return contentRepository.save(content);
    }
}

Data Access Layer Interface Contracts

public interface ContentRepository {
    Content save(Content content);
    Optional<Content> findById(UUID id);
    List<Content> findByStatus(ContentStatus status, Pageable pageable);
    long countByStatus(ContentStatus status);
    void deleteById(UUID id);
    
    // Additional content-specific operations
    List<Content> findByContentTypeAndStatus(ContentType type, ContentStatus status);
    List<Content> findContentForReview(Duration reviewWindow);
    void updateContentModerationMetadata(UUID id, Map<String, Object> metadata);
}

The repository layer provides a clean abstraction over database operations while maintaining performance and scalability requirements for content moderation workflows.

9. Asynchronous Processing & Messaging

Event-driven boundaries in this feature

The content moderation pipeline will use a hybrid synchronous/asynchronous approach to balance latency requirements with system throughput and resilience.

Synchronous flows:

• Content submission (user uploads content) - handled synchronously to provide immediate feedback to users
• Immediate policy checks (e.g., spam detection, profanity filter) - handled synchronously to prevent content from being published if it violates policies

Asynchronous flows:

• Deep content analysis (image recognition, video processing, text sentiment analysis) - handled asynchronously to avoid blocking user experience
• Moderation review queue management - handled asynchronously for scalability and load distribution
• Content re-evaluation after policy updates - handled asynchronously to process batch changes efficiently
• Notification and audit logging - handled asynchronously for better system decoupling

Message schema design with sample payload

{
  "version": "1.0",
  "event_type": "content.moderation.process",
  "timestamp": "2023-11-15T14:30:00Z",
  "content_id": "c7f8a9b2-1e4d-4f5a-9c3b-8d7e6f5a4b3c",
  "content_type": "image",
  "content_url": "https://storage.example.com/content/123456789.jpg",
  "user_id": "u987654321",
  "metadata": {
    "source": "mobile_app",
    "language": "en"
  },
  "policy_context": {
    "version": "v2.1",
    "categories": ["adult_content", "violence", "spam"],
    "thresholds": {
      "adult_content": 0.85,
      "violence": 0.90
    }
  }
}

Dead letter queue strategy and poison message handling

Poison messages will be handled through a dead-letter queue (DLQ) system with the following strategy:

1. Message reprocessing limit: Maximum 3 attempts before routing to DLQ
2. DLQ configuration: Messages that fail after max retries will be moved to a dedicated moderation.dlq topic
3. DLQ monitoring: DLQ messages will trigger alerts and require manual intervention by moderation team
4. DLQ cleanup: Messages in DLQ will be retained for 7 days before permanent deletion

At-least-once vs. exactly-once delivery

At-least-once delivery is selected for this feature because:

1. Content moderation is idempotent - reprocessing the same content won't cause harm
2. The system prioritizes correctness over strict uniqueness of processing
3. Content can be re-submitted by users, making exactly-once more complex to implement

Retry policy for consumers

public class ModerationRetryPolicy {
    private static final int MAX_RETRIES = 3;
    private static final long INITIAL_DELAY_MS = 1000;
    private static final double BACKOFF_MULTIPLIER = 2.0;
    private static final long MAX_DELAY_MS = 30000; // 30 seconds
    
    public RetryResult executeWithRetry(Supplier<CompletableFuture<Boolean>> operation) {
        int attempt = 0;
        long delayMs = INITIAL_DELAY_MS;
        
        while (attempt < MAX_RETRIES) {
            try {
                CompletableFuture<Boolean> result = operation.get();
                if (result != null && result.get()) {
                    return RetryResult.SUCCESS;
                }
            } catch (Exception e) {
                attempt++;
                if (attempt >= MAX_RETRIES) {
                    return RetryResult.FAILURE;
                }
                
                // Exponential backoff with jitter
                long jitter = ThreadLocalRandom.current().nextLong(0, delayMs / 2);
                long actualDelay = Math.min(delayMs + jitter, MAX_DELAY_MS);
                
                try {
                    Thread.sleep(actualDelay);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    return RetryResult.FAILURE;
                }
                
                delayMs = (long) (delayMs * BACKOFF_MULTIPLIER);
            }
        }
        
        return RetryResult.FAILURE;
    }
    
    enum RetryResult {
        SUCCESS, FAILURE
    }
}

Outbox pattern for reliable event publishing

The outbox pattern is implemented in the moderation pipeline to ensure message delivery reliability:

1. Transaction boundary: Content moderation events are written to an outbox table within the same database transaction as content processing
2. Background processor: A dedicated daemon process polls the outbox table and publishes messages to the message broker
3. Idempotency: Events are marked as published after successful delivery to prevent duplicate processing
4. Error handling: Failed events are retried according to the retry policy before being moved to DLQ

Sequence diagram for async flows

sequenceDiagram
    participant User
    participant API
    participant ContentService
    participant ModerationService
    participant OutboxRepo
    participant MessageBroker
    participant ContentProcessor
    participant DLQHandler
    
    User->>API: Submit content
    API->>ContentService: Process content
    ContentService->>ModerationService: Queue for moderation
    ModerationService->>OutboxRepo: Insert to outbox table
    OutboxRepo->>MessageBroker: Publish moderation event
    MessageBroker->>ContentProcessor: Deliver content analysis job
    ContentProcessor->>MessageBroker: Publish result event
    MessageBroker->>ModerationService: Deliver analysis result
    ModerationService->>API: Return moderation status
    
    loop Retry on failure
        MessageBroker->>ContentProcessor: Retry delivery
        ContentProcessor->>MessageBroker: Publish result event (retry)
    end
    
    MessageBroker->>DLQHandler: Deliver to DLQ on max retries

10. Error Handling Design

Error Classification Hierarchy

The error handling strategy follows a hierarchical classification to enable proper routing and response:

1. Validation Errors - Input data fails validation rules (400 Bad Request)
2. Domain Errors - Business rule violations (409 Conflict)
3. Infrastructure Errors - Database, cache, or external service failures (500 Internal Server Error)
4. Authorization Errors - Permission denied (403 Forbidden)
5. Rate Limiting Errors - API quota exceeded (429 Too Many Requests)

Error Propagation Strategy

• Domain Layer: Business logic exceptions are wrapped in custom domain exceptions and re-thrown
• Service Layer: Infrastructure and external service errors are wrapped in appropriate exception types and re-thrown
• Controller Layer: All exceptions are caught and mapped to standardized HTTP responses

Retry Strategy with Exponential Backoff and Jitter

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.5;

    public <T> T executeWithRetry(Supplier<T> operation) {
        Exception lastException = null;
        
        for (int attempt = 0; attempt <= MAX_RETRIES; attempt++) {
            try {
                return operation.get();
            } catch (Exception e) {
                lastException = e;
                
                if (attempt == MAX_RETRIES) {
                    throw new RuntimeException("Operation failed after " + MAX_RETRIES + " attempts", lastException);
                }
                
                // Calculate delay with jitter
                long delay = (long) (BASE_DELAY_MS * Math.pow(2, attempt) + 
                                    (Math.random() - 0.5) * BASE_DELAY_MS * JITTER_FACTOR);
                
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException("Retry interrupted", ie);
                }
            }
        }
        
        throw new RuntimeException("Unexpected execution path", lastException);
    }
}

Circuit Breaker Design

public class CircuitBreaker {
    private final int failureThreshold;
    private final long timeoutMs;
    private final int successThreshold;
    
    private volatile CircuitState state = CircuitState.CLOSED;
    private volatile int failureCount = 0;
    private volatile long lastFailureTime = 0;
    
    enum CircuitState {
        CLOSED, OPEN, HALF_OPEN
    }
    
    public boolean canExecute() {
        switch (state) {
            case CLOSED:
                return true;
            case HALF_OPEN:
                return failureCount < successThreshold;
            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) {
            failureCount = 0;
            state = CircuitState.CLOSED;
        }
    }
    
    public void recordFailure() {
        lastFailureTime = System.currentTimeMillis();
        
        switch (state) {
            case CLOSED:
                failureCount++;
                if (failureCount >= failureThreshold) {
                    state = CircuitState.OPEN;
                }
                break;
            case HALF_OPEN:
                failureCount++;
                if (failureCount >= failureThreshold) {
                    state = CircuitState.OPEN;
                } else {
                    state = CircuitState.CLOSED;
                }
                break;
            case OPEN:
                // No change in state
                break;
        }
    }
}

Fallback Behavior Design

public class ContentModerationService {
    private final ExternalModerationClient moderationClient;
    private final CircuitBreaker circuitBreaker = new CircuitBreaker();
    
    public ModerationResult moderateContent(ContentRequest request) {
        try {
            if (circuitBreaker.canExecute()) {
                return moderationClient.moderate(request);
            } else {
                // Fallback to basic content validation
                return fallbackModeration(request);
            }
        } catch (Exception e) {
            circuitBreaker.recordFailure();
            return fallbackModeration(request);
        }
    }
    
    private ModerationResult fallbackModeration(ContentRequest request) {
        // Basic content filtering logic as fallback
        return new ModerationResult(
            ModerationStatus.PENDING,
            "Fallback moderation applied",
            Collections.emptyList()
        );
    }
}

Error Logging Format and Alerting Thresholds

Structured Log Schema:

{
  "timestamp": "2023-12-01T10:30:45.123Z",
  "trace_id": "abc123def456",
  "span_id": "xyz789uvw012",
  "user_id": "user-12345",
  "error_code": "MODERATION_SERVICE_UNAVAILABLE",
  "error_message": "External moderation service timeout",
  "error_type": "InfrastructureError",
  "service_name": "ContentModerationService",
  "request_id": "req-98765",
  "stack_trace": "...",
  "severity": "ERROR"
}

Alerting Thresholds:

• Critical: > 100 errors per minute in moderation service
• Warning: > 50 errors per minute in moderation service
• Critical: > 5% of requests failing with infrastructure errors
• Warning: > 1% of requests failing with authorization errors

Error Response Format

All error responses follow a standardized JSON format:

{
  "error": {
    "code": "MODERATION_SERVICE_UNAVAILABLE",
    "message": "Content moderation service is temporarily unavailable",
    "details": {
      "retry_after_ms": 5000,
      "request_id": "req-98765"
    },
    "timestamp": "2023-12-01T10:30:45.123Z"
  }
}

Error Handling in External Service Integration

public class ExternalModerationClient {
    private final CircuitBreaker circuitBreaker = new CircuitBreaker();
    
    public ModerationResult moderate(ContentRequest request) {
        try {
            // Circuit breaker check
            if (!circuitBreaker.canExecute()) {
                throw new ServiceUnavailableException("External moderation service is unavailable");
            }
            
            // Retry with exponential backoff
            return retryableCall(() -> {
                // Make external API call
                return apiClient.callModerationService(request);
            });
            
        } catch (ExternalServiceException e) {
            circuitBreaker.recordFailure();
            throw new ModerationServiceException("External moderation service error", e);
        }
    }
}

Error Handling for Async Operations

For asynchronous content moderation tasks, error handling includes:

1. Dead Letter Queue (DLQ) Strategy: Failed messages are moved to DLQ after max retries
2. Poison Message Detection: Messages that consistently fail are flagged for manual review
3. Idempotency Handling: Retry logic ensures duplicate processing doesn't cause issues

Error Recovery Patterns

1. Graceful Degradation: Fallback to basic content filtering when external services fail
2. Circuit Breaking: Prevent cascading failures in the moderation pipeline
3. Retry with Exponential Backoff: Handle transient infrastructure issues
4. Error Categorization: Route errors to appropriate monitoring and alerting systems

This error handling design ensures robustness, observability, and graceful degradation while maintaining service availability under various failure conditions.

11. Security Design

Input Validation and Sanitization Strategy

All API inputs are validated using a combination of:

• Library: Apache Commons Validator for standard validation (email, URL, phone)
• Custom rules: Regex patterns for content format (e.g., max length, allowed characters)
• Sanitization: OWASP Java HTML Sanitizer for user-generated content to remove malicious tags and attributes

SQL Injection Prevention

• ORM usage: All database operations use JPA/Hibernate with parameterized queries to prevent SQL injection
• Raw query guardrails: Any raw SQL usage is strictly reviewed and validated through static analysis tools (SpotBugs, SonarQube)
• Input escaping: All user inputs are escaped before being used in dynamic queries

Authentication Middleware Design

• JWT-based: JWT tokens are used for authentication with 24-hour expiration and refresh token support
• Middleware: Custom Spring Security filter chain that validates JWT tokens, checks expiration, and extracts user claims
• Token storage: Refresh tokens stored in secure HTTP-only cookies with SameSite=Strict
• Token rotation: Access tokens are rotated on every request, refresh tokens are rotated after successful authentication

Authorization Enforcement

• RBAC model: Role-Based Access Control with predefined roles (Admin, Moderator, User)
• Permission checks: Service layer enforces permissions using @PreAuthorize annotations and custom permission evaluators
• Resource scope: Permissions are scoped to specific content types (e.g., MODERATE_POSTS, MODERATE_COMMENTS)
• Dynamic permissions: Support for role inheritance and permission overrides via configuration

Sensitive Data Encryption

• At rest: AES-256 encryption for sensitive data in database (user content, moderation logs)
• In transit: TLS 1.3 for all API communications with HSTS enabled
• Key management: AWS KMS for key rotation and lifecycle management

PII Handling and Masking Strategy

• Masking: User identifiers (email, phone) are masked in logs and audit trails using a consistent pattern (e.g., user***@domain.com)
• Storage: PII is stored in encrypted format and only accessible to authorized roles
• Deletion: PII is automatically deleted after 365 days of inactivity with soft-delete mechanism

Secrets Management

• Vault: AWS Secrets Manager for managing API keys, database credentials, and external service tokens
• Rotation: Secrets are rotated every 90 days with automated rotation workflows
• Access control: Least privilege principle enforced via IAM policies and secret ACLs

Audit Logging Design

• Events logged: Content moderation actions (review, approve, reject, flag), user actions, system events
• Format: Structured JSON with fields: timestamp, user_id, action, content_id, reason, ip_address
• Retention: 180 days for audit logs with automatic archival to S3 Glacier
• Access control: Audit logs are accessible only to Admin and Security team roles with read-only permissions

// Example of input validation in controller
@PostMapping("/moderate")
public ResponseEntity<ContentModerationResponse> moderateContent(
    @Valid @RequestBody ContentModerationRequest request) {
    // Validation handled by Bean Validation annotations
    // Sanitization of content field using OWASP sanitizer
    String sanitizedContent = htmlSanitizer.sanitize(request.getContent());
    
    // JWT validation in filter chain
    String userId = jwtTokenProvider.getUserIdFromToken();
    
    // Permission check using Spring Security
    if (!hasPermission(userId, "MODERATE_CONTENT")) {
        throw new AccessDeniedException("Insufficient permissions");
    }
    
    return ResponseEntity.ok(moderationService.processContent(request));
}

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. Focus on:

• Core moderation rules evaluation logic
• Content classification algorithms (text, image, video)
• Policy engine decision making
• State transition validation
• Error handling paths

Integration Testing Strategy

Database Integration Tests:

• Use testcontainers for PostgreSQL/MySQL with schema migration
• Test transaction boundaries and concurrent access patterns
• Validate data consistency under stress conditions

External Service Integration Tests:

• WireMock for third-party content analysis APIs (e.g., Google Vision, AWS Rekognition)
• Test failure scenarios and retry logic
• Validate payload format compatibility

Event System Integration Tests:

• Embedded Kafka or RabbitMQ for async flow validation
• Test message delivery and consumer handling under load

Contract Testing

Implement Pact-based contract testing between:

• Moderation service and content ingestion service
• Moderation service and notification service
• Moderation service and external analysis providers

Each contract will define:

• Request/response schemas for all endpoints
• Expected error responses and status codes
• Content format expectations (image/jpeg, text/plain, etc.)

Performance Testing Tool and Scenarios

Load Testing:

• Use k6 for load testing with 1000 concurrent content moderation requests
• Target 95th percentile response time < 200ms for content analysis
• Test under burst traffic (5000 concurrent requests)

Stress Testing:

• Gradually increase load to identify saturation points
• Monitor database connection pool exhaustion and cache eviction

Scalability Testing:

• Simulate 100,000 content items per hour
• Validate horizontal scaling with multiple service instances

Test Data Management Strategy

Test Fixtures:

• Create realistic content samples (text, images, videos) for testing
• Include edge cases: profanity, sensitive content, false positives/negatives

Test Data Factories:

• Factory pattern for generating test content objects with various attributes
• Support for generating large batches of test data

Database Seeding:

• Use Liquibase or Flyway for test database schema setup
• Seed with realistic moderation policy configurations and rule sets
• Ensure test data isolation through database transaction rollback

// Example of test data factory for content moderation
public class ContentModerationTestDataFactory {
    public static Content createTextContent(String text, String contentType) {
        return Content.builder()
                .id(UUID.randomUUID())
                .type(contentType)
                .data(text)
                .metadata(Map.of("language", "en"))
                .build();
    }
    
    public static Content createImageContent(String imageUrl) {
        return Content.builder()
                .id(UUID.randomUUID())
                .type("image/jpeg")
                .data(imageUrl)
                .metadata(Map.of("width", 1024, "height", 768))
                .build();
    }
}

13. Observability Design

Structured Logging Schema

All logs must follow a structured format with consistent fields to enable aggregation and analysis:

{
  "timestamp": "2023-10-27T14:30:00.000Z",
  "level": "INFO|WARN|ERROR",
  "service": "content-moderation-service",
  "trace_id": "123e4567-e89b-12d3-a456-426614174000",
  "span_id": "abcdef1234567890",
  "user_id": "user-123",
  "content_id": "content-456",
  "event_type": "moderation_decision",
  "decision": "approved|rejected|flagged",
  "reason_code": "spam|inappropriate|policy_violation",
  "duration_ms": 150,
  "error_code": "MOD_001",
  "error_message": "Moderation service timeout",
  "pii_masked": true
}

PII masking rules:

• User IDs and content IDs are masked in logs when not required for debugging
• All PII fields must be redacted or hashed in logs except for audit purposes

Distributed Tracing Instrumentation

Using OpenTelemetry with custom spans:

Key operations to trace:

1. Content ingestion pipeline: content_ingestion_pipeline
2. Moderation decision flow: moderation_decision_flow
3. External API calls: external_api_call
4. Async processing: async_moderation_processing

Span attributes to capture:

• content_type: image, text, video
• moderation_rule_set: default, custom, community_guidelines
• decision_time_ms: duration of moderation decision process
• external_service: AWS Rekognition, Google Cloud Vision, custom ML model
• model_version: v1.2, v2.0

Metrics Design

RED Method Metrics (Rate, Errors, Duration):

Metric Name	Type	Description	Labels
moderation_requests_total	Counter	Total moderation requests processed	status, content_type, decision
moderation_errors_total	Counter	Total errors in moderation pipeline	error_type, service, content_type
moderation_duration_seconds	Histogram	Time to process moderation request	content_type, decision, service
moderation_queue_size	Gauge	Current size of moderation queue	queue_type
moderation_api_response_time_seconds	Histogram	External API response times	api_service, status_code
moderation_decision_rate	Gauge	Rate of different moderation decisions	decision, content_type

Health Check Endpoint Design

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

Response format:

{
  "status": "healthy|unhealthy",
  "checks": {
    "database": {"status": "healthy", "details": "Connected to primary DB"},
    "cache": {"status": "healthy", "details": "Redis connected"},
    "external_services": {"status": "healthy", "details": "All external services reachable"},
    "queue": {"status": "healthy", "details": "Moderation queue size: 120"},
    "last_check": "2023-10-27T14:30:00.000Z"
  }
}

Health check criteria:

• /live: Basic service liveness (process running)
• /ready: Service ready to serve traffic (DB, cache connected)
• /health: Full health including external dependencies
• /startup: Service has completed startup sequence

Alerting Thresholds

Metric	Warning Level	Critical Level	Escalation Path
moderation_errors_total	> 5 errors/minute	> 20 errors/minute	On-call engineer, team lead
moderation_duration_seconds p95	> 500ms	> 2s	SRE team, engineering lead
moderation_queue_size	> 1000 items	> 5000 items	SRE team, engineering lead
moderation_api_response_time_seconds p95	> 1000ms	> 3s	SRE team, external service provider
moderation_requests_total rate drop	< 10 req/sec	< 5 req/sec	SRE team, engineering lead

Alerting rules:

• Alert thresholds use sliding window (5-minute intervals)
• Critical alerts trigger immediate on-call escalation
• Warning alerts generate team notifications but don't escalate immediately
• All alerts include relevant context (service, content type, error details)

14. Open Questions, Risks & Trade-offs

Decisions requiring product / engineering alignment

• Threshold sensitivity tuning: Should moderation rules be tuned per content type (text, image, video) or use a unified threshold? This requires product alignment on acceptable false positive rates vs. false negatives.
• Human review escalation policy: What percentage of flagged content should be escalated to human reviewers? This impacts both moderation quality and operational overhead.
• Content type classification strategy: Should content be classified by type (text, image, video) before moderation or should a unified pipeline handle all content types? This affects system complexity and performance.
• Moderation rule update frequency: How often should moderation rules be updated or retrained? This impacts system responsiveness to new threats and requires coordination between ML teams and product.

Known trade-offs and justifications

Trade-off	Option 1	Option 2	Justification
Real-time vs. batch moderation	Real-time processing	Batch processing	Real-time provides better user experience but increases latency and infrastructure cost. Batch processing reduces cost but introduces delay in content removal, potentially allowing harmful content to remain visible
Rule-based vs. ML-based moderation	Rule-based rules	Machine learning models	Rule-based systems are more predictable and interpretable but require manual maintenance. ML systems adapt to new threats automatically but can be opaque and require extensive training data
Centralized vs. distributed moderation	Single moderation service	Per-content-type services	Centralized approach reduces complexity but creates a single point of failure. Distributed approach improves scalability and fault isolation but increases operational overhead
Content caching vs. real-time fetching	Cache content for faster moderation	Fetch content on demand	Caching improves performance but increases memory usage and may serve stale content. On-demand fetching reduces memory overhead but increases latency for repeated moderation requests

Risks and mitigation strategies

Risk	Impact	Mitigation Strategy
High false positive rate	User experience degradation, potential content removal errors	Implement human review queue with clear escalation paths, provide user appeal mechanisms, continuously tune moderation thresholds
Moderation system being bypassed	Security vulnerabilities, policy violations	Implement rate limiting, input validation, and monitoring for suspicious patterns
ML model drift	Inability to detect new content patterns	Implement model monitoring, regular retraining schedules, and A/B testing for rule updates
High latency in moderation	User experience degradation, content visibility delays	Implement async processing with caching, prioritize critical content types, and provide user feedback mechanisms
Scalability limitations	System cannot handle peak traffic loads	Design horizontal scaling capabilities, implement efficient caching strategies, and use load balancing
Data privacy compliance	Regulatory penalties, data breaches	Implement strict data handling policies, encryption at rest and in transit, and regular compliance audits

Technical debt acknowledged

• Rule maintenance complexity: Current rule-based system requires manual updates and lacks version control, creating technical debt in rule management and rollback capabilities.
• ML model monitoring gaps: No automated system for detecting model performance degradation or drift, requiring manual intervention to maintain moderation quality.
• Limited content type support: The system currently only supports basic text and image content, with no extensible framework for supporting new content types.
• Inconsistent error handling: Different moderation components use varying error handling patterns, creating inconsistency in system behavior and making debugging difficult.
• No audit trail for moderation decisions: Lack of detailed logging for moderation actions makes it difficult to understand decision-making process or investigate false positives/negatives.
• Tight coupling between content and moderation: Content processing logic is tightly coupled with moderation logic, making system harder to extend or modify without affecting other components.