Filtering Engine Design: Scalable Backend Architecture for Real-Time Data Processing

Comprehensive low-level design for a scalable filtering engine that processes real-time data streams with configurable rules and efficient query execution.

PublishedOctober 5, 2025
Length8,112 words
Reading time41 min read

Backend LLD

A filtering engine is a critical backend component that processes real-time data streams, applying configurable rules to filter and route data efficiently. This design document outlines a scalable architecture that handles high-throughput filtering operations while maintaining low latency and supporting complex rule configurations through a modular, extensible system.

1. Problem Statement & Requirements

Clarifying questions

What is the scale of data and queries we're targeting? Is this for a few thousand records or millions of records with complex filtering?
What's the expected read/write ratio for filter configurations? Are we optimizing more for query performance or configuration flexibility?
What consistency requirements exist? Should filter configurations be immediately available to all users after update, or eventual consistency is acceptable?
What's the expected latency for filter execution? Is sub-50ms p95 acceptable or is this a batch processing scenario?
What authentication model is in place? Will users have roles or will all access be through API keys?
Are there any external systems that need to integrate with this filtering engine? For example, analytics or reporting tools.
What's the expected lifecycle of filters? Do they expire, or are they static once created?

Functional requirements

Support creation, update, and deletion of filter configurations
Execute filters against data sets with various operators (equals, not equals, contains, greater than, etc.)
Support complex filter combinations using AND/OR logic
Allow filtering on multiple fields with different data types (string, numeric, date, boolean)
Provide a way to preview filter results before applying them
Support filtering with custom functions or expressions (e.g., "field > 100")
Allow users to save and reuse filter configurations
Support filtering with nested data structures (e.g., JSON fields)
Provide audit trail for filter configuration changes

Non-functional requirements

Latency targets:
- p50: < 20ms
- p95: < 100ms
- p99: < 500ms
Throughput:
- 10,000 filter executions per second (peak)
Availability:
- 99.9% uptime (28.8 minutes of downtime/month max)
Scalability:
- Support horizontal scaling to handle up to 100x current load

Constraints and assumptions

Filters will be defined using a DSL or JSON schema
Data sets to be filtered are stored in external systems (e.g., databases, search engines)
Filter engine will not store or manage the actual data to be filtered
Assumption: Filters are primarily used for read operations, not write
Assumption: Filter configurations will be managed by a separate admin service or UI
Assumption: Users will have pre-defined roles and permissions for filter access

Out of scope

Data storage or management (this is a filtering engine, not a data store)
Real-time data streaming or continuous filtering
Complex analytics or aggregation functions (e.g., grouping, windowing)
Multi-tenant data isolation (assumes single tenant or shared data model)
Native support for complex mathematical expressions beyond basic operators
Integration with specific external systems (e.g., Elasticsearch, MongoDB)
Filter versioning or rollback capabilities
UI/UX for filter creation/editing (assumes external admin interface)
Support for custom filter functions or plugins beyond basic operators

2. API Design

API Style Selection

REST is selected for this filtering engine due to its simplicity, wide adoption, and clear semantics. The filtering engine will primarily serve as a query execution layer that transforms filter criteria into database queries, making REST's resource-oriented approach well-suited for this use case.

API Endpoints

Method	Path	Description	Auth
POST	`/v1/filters`	Create a new filter configuration	JWT
GET	`/v1/filters/{id}`	Retrieve a specific filter by ID	JWT
PUT	`/v1/filters/{id}`	Update an existing filter configuration	JWT
DELETE	`/v1/filters/{id}`	Delete a filter configuration	JWT
POST	`/v1/filters/validate`	Validate filter criteria syntax	JWT
POST	`/v1/filters/execute`	Execute a filter query against data source	JWT
GET	`/v1/filters/{id}/results`	Retrieve results for a previously executed filter	JWT

Request and Response Schemas

Create Filter Request

{
  "name": "customer_segment_filter",
  "description": "Filter customers by segment and activity",
  "criteria": {
    "conditions": [
      {
        "field": "customer_segment",
        "operator": "IN",
        "values": ["VIP", "Premium"]
      },
      {
        "field": "last_activity_date",
        "operator": "GREATER_THAN",
        "value": "2023-01-01T00:00:00Z"
      }
    ],
    "logical_operator": "AND"
  },
  "data_source": {
    "type": "customer_db",
    "table": "customers"
  }
}

Create Filter Response

{
  "id": "uuid-12345",
  "name": "customer_segment_filter",
  "description": "Filter customers by segment and activity",
  "criteria": {
    "conditions": [
      {
        "field": "customer_segment",
        "operator": "IN",
        "values": ["VIP", "Premium"]
      },
      {
        "field": "last_activity_date",
        "operator": "GREATER_THAN",
        "value": "2023-01-01T00:00:00Z"
      }
    ],
    "logical_operator": "AND"
  },
  "data_source": {
    "type": "customer_db",
    "table": "customers"
  },
  "created_at": "2023-06-15T10:30:00Z",
  "updated_at": "2023-06-15T10:30:00Z"
}

Execute Filter Request

{
  "filter_id": "uuid-12345",
  "limit": 100,
  "offset": 0,
  "sort": [
    {
      "field": "created_at",
      "direction": "DESC"
    }
  ]
}

Execute Filter Response

{
  "results": [
    {
      "id": "customer-123",
      "name": "John Doe",
      "email": "john@example.com",
      "customer_segment": "VIP"
    }
  ],
  "total_count": 42,
  "has_more": true
}

HTTP Status Codes and Error Response Format

Status Code	Description
200	Success
201	Created
400	Bad Request - Invalid filter criteria
401	Unauthorized
403	Forbidden
404	Not Found
422	Unprocessable Entity - Validation errors
500	Internal Server Error

Standard error response format:

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Filter criteria validation failed",
    "details": [
      {
        "field": "criteria.conditions[0].field",
        "message": "Field 'customer_segment' is not valid for this data source"
      }
    ]
  }
}

API Versioning Strategy

Versioned using path prefix: /v1/filters/... with support for future versioning like /v2/filters/...

Pagination Design

Cursor-based pagination is selected because:

Provides better consistency for datasets that change during pagination
Handles duplicate records better than offset-based approaches
More efficient for large datasets

Cursor format: cursor=eyJmaWVsZCI6ImNyZWF0ZWRfYXQiLCJ2YWx1ZSI6IjIwMjMtMDYtMTVUMTA6MzA6MDBaIn0=

Authentication and Authorization

JWT tokens with scopes for filtering engine operations
Scope-based access control: filter:read, filter:write, filter:execute
JWT claims validation for user roles and permissions

Rate Limiting

Endpoint	Limit	Period	Enforcement
`/v1/filters/validate`	100 req/min	1 minute	In-memory token bucket
`/v1/filters/execute`	50 req/min	1 minute	Redis-based token bucket
`/v1/filters`	200 req/min	1 minute	In-memory token bucket

Idempotency Design

Mutation endpoints support idempotency using the Idempotency-Key header:

Key format: {user_id}:{operation_type}:{client_generated_id}
Store operation results for 24 hours
Return same result for duplicate requests within window

Example header: Idempotency-Key: user-123:filter_create:abc123

API Documentation

OpenAPI 3.0 specification will be maintained with:

Comprehensive parameter descriptions
Example payloads for all endpoints
Error code documentation
Rate limit information

3. Data Model Design

Entity-Relationship Diagram

erDiagram
    FILTER_CONFIG {
        string id PK
        string name
        string description
        jsonb conditions
        boolean is_active
        timestamp created_at
        timestamp updated_at
        string created_by
    }

    FILTER_CONDITION {
        string id PK
        string filter_config_id FK
        string field_name
        string operator
        jsonb value
        int priority
    }

    FILTER_RESULT {
        string id PK
        string filter_config_id FK
        string entity_id
        boolean is_match
        timestamp processed_at
    }

    FILTER_CONFIG ||--o{ FILTER_CONDITION : "has"
    FILTER_CONFIG ||--o{ FILTER_RESULT : "produces"

Table Schema Definitions

-- Filter Configuration table
CREATE TABLE filter_configs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    description TEXT,
    conditions JSONB NOT NULL,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    created_by VARCHAR(255) NOT NULL,
    
    INDEX idx_filter_configs_active (is_active),
    INDEX idx_filter_configs_created_by (created_by),
    INDEX idx_filter_configs_updated_at (updated_at)
);

-- Filter Conditions table
CREATE TABLE filter_conditions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    filter_config_id UUID NOT NULL REFERENCES filter_configs(id) ON DELETE CASCADE,
    field_name VARCHAR(255) NOT NULL,
    operator VARCHAR(50) NOT NULL, -- eq, neq, gt, lt, in, contains, etc.
    value JSONB NOT NULL,
    priority INT NOT NULL DEFAULT 0,
    
    INDEX idx_filter_conditions_config_id (filter_config_id),
    INDEX idx_filter_conditions_field_name (field_name),
    INDEX idx_filter_conditions_priority (priority),
    
    CONSTRAINT chk_operator_valid CHECK (operator IN ('eq', 'neq', 'gt', 'lt', 'gte', 'lte', 'in', 'not_in', 'contains', 'not_contains', 'exists', 'not_exists'))
);

-- Filter Results table
CREATE TABLE filter_results (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    filter_config_id UUID NOT NULL REFERENCES filter_configs(id) ON DELETE CASCADE,
    entity_id VARCHAR(255) NOT NULL,
    is_match BOOLEAN NOT NULL,
    processed_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_filter_results_config_id (filter_config_id),
    INDEX idx_filter_results_entity_id (entity_id),
    INDEX idx_filter_results_processed_at (processed_at),
    INDEX idx_filter_results_match (is_match),
    
    UNIQUE KEY uk_filter_results_config_entity (filter_config_id, entity_id)
);

Primary Key Strategy

UUID is used as the primary key for all tables because:

Enables distributed system compatibility without coordination
Prevents sequential ID prediction attacks
Allows for better sharding strategies based on UUID ranges
Supports cross-service entity references without ID conflicts

Index Design

Table	Index Name	Fields	Type	Purpose
filter_configs	idx_filter_configs_active	is_active	Standard	Fast filtering of active configs
filter_configs	idx_filter_configs_created_by	created_by	Standard	Audit and ownership queries
filter_configs	idx_filter_configs_updated_at	updated_at	Standard	Change tracking and cache invalidation
filter_conditions	idx_filter_conditions_config_id	filter_config_id	Standard	Fast lookup of conditions by config
filter_conditions	idx_filter_conditions_field_name	field_name	Standard	Field-based query optimization
filter_conditions	idx_filter_conditions_priority	priority	Standard	Priority-based condition evaluation
filter_results	idx_filter_results_config_id	filter_config_id	Standard	Fast lookup of results by config
filter_results	idx_filter_results_entity_id	entity_id	Standard	Fast lookup of specific entity match
filter_results	idx_filter_results_processed_at	processed_at	Standard	Time-based result queries
filter_results	idx_filter_results_match	is_match	Standard	Fast filtering of matching/non-matching results
filter_results	uk_filter_results_config_entity	filter_config_id, entity_id	Unique	Prevent duplicate processing

Data Partitioning / Sharding Strategy

Sharding by filter_config_id hash:

Hash-based sharding using consistent hashing for even distribution
Partition key: hash(filter_config_id) % NUMBER_OF_SHARDS
Each shard contains related filter configs and their results
Allows for horizontal scaling of filtering operations
Shard boundary changes are handled through consistent hashing to minimize data movement

Soft Delete vs. Hard Delete Decision

Soft delete is used for filter_configs and filter_conditions because:

Historical audit trail is required for compliance and debugging
Filter configurations may be re-enabled after temporary deactivation
Prevents data loss during accidental deletion or system errors
Allows for rollback capability in case of configuration errors

Audit Trail Design

All data modifications are tracked through:

created_at and updated_at timestamps
created_by field to identify the operator
updated_at is updated on every modification (including soft deletes)
Change tracking implemented via triggers or application-level logging

Data Archival Strategy and Retention Policy

Archival Strategy:

Results older than 90 days are archived to cold storage (S3/BigQuery)
Active filter configurations are retained indefinitely
Historical results are compressed and stored with reduced granularity

Retention Policy:

Data Type	Retention Period	Archive Location
Filter Results	90 days	S3 Glacier
Filter Configurations	Indefinite	Primary DB
Audit Logs	1 year	Log aggregation system

// Example of how the data model might be represented in Java
public class FilterConfig {
    private UUID id;
    private String name;
    private String description;
    private List<FilterCondition> conditions;
    private boolean isActive;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
    private String createdBy;
    
    // Getters and setters
}

public class FilterCondition {
    private UUID id;
    private UUID filterConfigId;
    private String fieldName;
    private String operator;
    private Object value;
    private Integer priority;
    
    // Getters and setters
}

public class FilterResult {
    private UUID id;
    private UUID filterConfigId;
    private String entityId;
    private boolean isMatch;
    private LocalDateTime processedAt;
    
    // Getters and setters
}

4. Class Design & Service Layer

classDiagram
    class FilterEngine {
        +applyFilters(List~Filter~, List~Record~) List~Record~
        +validateFilter(Filter) boolean
    }
    
    class Filter {
        +id: String
        +field: String
        +operator: Operator
        +value: Object
        +logicalOperator: LogicalOperator
    }
    
    class Record {
        +id: String
        +data: Map~String, Object~
    }
    
    class FilterProcessor {
        +processFilter(Filter, List~Record~) List~Record~
        +applyOperator(Operator, Object, Object) boolean
    }
    
    class Operator {
        +name: String
        +type: OperatorType
    }
    
    class FilterValidator {
        +validateFilter(Filter) ValidationResult
    }
    
    class FilterBuilder {
        +buildFilter(String, String, Object) Filter
    }
    
    class FilterEngineException {
        +message: String
        +errorCode: String
    }
    
    class ValidationResult {
        +isValid: boolean
        +errors: List~String~
    }
    
    FilterEngine --> FilterProcessor : uses
    FilterEngine --> FilterValidator : validates
    FilterProcessor --> Operator : applies
    FilterValidator --> ValidationResult : returns
    FilterEngine --> FilterBuilder : builds
    Filter --> Operator : has
    Filter --> LogicalOperator : has
    FilterEngine --> FilterEngineException : throws
    
    class OperatorType {
        +EQUALS
        +NOT_EQUALS
        +GREATER_THAN
        +LESS_THAN
        +CONTAINS
        +STARTS_WITH
        +ENDS_WITH
        +IN
        +BETWEEN
    }
    
    class LogicalOperator {
        +AND
        +OR
    }

stateDiagram-v2
    [*] --> Idle
    Idle --> Processing : applyFilters()
    Processing --> Completed : success
    Processing --> Failed : validation error
    Failed --> Idle : retry

Service Class Structure and Responsibilities

The filtering engine service is designed with a layered architecture that separates concerns:

FilterEngine: Main orchestrator that manages the filtering lifecycle
FilterProcessor: Core logic for applying filters to records
FilterValidator: Validation layer for filter integrity
Operator: Strategy pattern implementation for different comparison operations
FilterBuilder: Factory for creating filter objects

Interface / Abstract Class Design

public interface Filter {
    String getId();
    String getField();
    Operator getOperator();
    Object getValue();
    LogicalOperator getLogicalOperator();
}

public interface FilterProcessor {
    List<Record> processFilter(Filter filter, List<Record> records);
    boolean applyOperator(Operator operator, Object recordValue, Object filterValue);
}

public interface FilterValidator {
    ValidationResult validateFilter(Filter filter);
}

public enum OperatorType {
    EQUALS, NOT_EQUALS, GREATER_THAN, LESS_THAN, 
    CONTAINS, STARTS_WITH, ENDS_WITH, IN, BETWEEN
}

public enum LogicalOperator {
    AND, OR
}

Business Logic Encapsulation Patterns

The filtering engine implements the Strategy Pattern for operator handling:

public class OperatorStrategy {
    private final Map<OperatorType, OperatorHandler> handlers = new HashMap<>();
    
    public OperatorStrategy() {
        handlers.put(OperatorType.EQUALS, new EqualsHandler());
        handlers.put(OperatorType.NOT_EQUALS, new NotEqualsHandler());
        handlers.put(OperatorType.GREATER_THAN, new GreaterThanHandler());
        handlers.put(OperatorType.LESS_THAN, new LessThanHandler());
        handlers.put(OperatorType.CONTAINS, new ContainsHandler());
        handlers.put(OperatorType.STARTS_WITH, new StartsWithHandler());
        handlers.put(OperatorType.ENDS_WITH, new EndsWithHandler());
        handlers.put(OperatorType.IN, new InHandler());
        handlers.put(OperatorType.BETWEEN, new BetweenHandler());
    }
    
    public boolean apply(Operator operator, Object recordValue, Object filterValue) {
        OperatorHandler handler = handlers.get(operator.getType());
        if (handler == null) {
            throw new FilterEngineException("Unsupported operator: " + operator.getType());
        }
        return handler.apply(recordValue, filterValue);
    }
    
    interface OperatorHandler {
        boolean apply(Object recordValue, Object filterValue);
    }
    
    static class EqualsHandler implements OperatorHandler {
        @Override
        public boolean apply(Object recordValue, Object filterValue) {
            return Objects.equals(recordValue, filterValue);
        }
    }
    
    static class NotEqualsHandler implements OperatorHandler {
        @Override
        public boolean apply(Object recordValue, Object filterValue) {
            return !Objects.equals(recordValue, filterValue);
        }
    }
    
    static class GreaterThanHandler implements OperatorHandler {
        @Override
        public boolean apply(Object recordValue, Object filterValue) {
            if (recordValue instanceof Comparable && filterValue instanceof Comparable) {
                return ((Comparable) recordValue).compareTo((Comparable) filterValue) > 0;
            }
            return false;
        }
    }
    
    static class LessThanHandler implements OperatorHandler {
        @Override
        public boolean apply(Object recordValue, Object filterValue) {
            if (recordValue instanceof Comparable && filterValue instanceof Comparable) {
                return ((Comparable) recordValue).compareTo((Comparable) filterValue) < 0;
            }
            return false;
        }
    }
    
    static class ContainsHandler implements OperatorHandler {
        @Override
        public boolean apply(Object recordValue, Object filterValue) {
            if (recordValue instanceof String && filterValue instanceof String) {
                return ((String) recordValue).contains((String) filterValue);
            }
            return false;
        }
    }
    
    static class StartsWithHandler implements OperatorHandler {
        @Override
        public boolean apply(Object recordValue, Object filterValue) {
            if (recordValue instanceof String && filterValue instanceof String) {
                return ((String) recordValue).startsWith((String) filterValue);
            }
            return false;
        }
    }
    
    static class EndsWithHandler implements OperatorHandler {
        @Override
        public boolean apply(Object recordValue, Object filterValue) {
            if (recordValue instanceof String && filterValue instanceof String) {
                return ((String) recordValue).endsWith((String) filterValue);
            }
            return false;
        }
    }
    
    static class InHandler implements OperatorHandler {
        @Override
        public boolean apply(Object recordValue, Object filterValue) {
            if (filterValue instanceof Collection) {
                return ((Collection<?>) filterValue).contains(recordValue);
            }
            return false;
        }
    }
    
    static class BetweenHandler implements OperatorHandler {
        @Override
        public boolean apply(Object recordValue, Object filterValue) {
            if (filterValue instanceof List && ((List<?>) filterValue).size() == 2) {
                Object lower = ((List<?>) filterValue).get(0);
                Object upper = ((List<?>) filterValue).get(1);
                if (recordValue instanceof Comparable && lower instanceof Comparable && upper instanceof Comparable) {
                    return ((Comparable) recordValue).compareTo((Comparable) lower) >= 0 &&
                           ((Comparable) recordValue).compareTo((Comparable) upper) <= 0;
                }
            }
            return false;
        }
    }
}

Critical-Path Method Implementation

public class FilterEngine {
    private final FilterProcessor processor;
    private final FilterValidator validator;
    private final OperatorStrategy operatorStrategy;

    public FilterEngine(FilterProcessor processor, FilterValidator validator) {
        this.processor = processor;
        this.validator = validator;
        this.operatorStrategy = new OperatorStrategy();
    }

    public List<Record> applyFilters(List<Filter> filters, List<Record> records) {
        // Validate all filters before processing
        for (Filter filter : filters) {
            ValidationResult result = validator.validateFilter(filter);
            if (!result.isValid()) {
                throw new FilterEngineException("Invalid filter: " + result.getErrors());
            }
        }

        // Apply filters sequentially
        List<Record> filteredRecords = new ArrayList<>(records);
        
        for (Filter filter : filters) {
            try {
                filteredRecords = processor.processFilter(filter, filteredRecords);
            } catch (Exception e) {
                throw new FilterEngineException("Failed to process filter: " + filter.getId(), e);
            }
        }

        return filteredRecords;
    }
}

Transaction Boundary Design

For the filtering operation, a single transaction boundary encompasses:

All filter validations
Sequential application of filters to the input records
Result set generation

Service Interface Contracts

public interface FilterEngineService {
    List<Record> applyFilters(List<Filter> filters, List<Record> records);
    Filter createFilter(String field, OperatorType operator, Object value);
    List<Record> applyFilterChain(List<Filter> filters, List<Record> records);
}

Saga / Orchestration Pattern

For complex filtering scenarios involving external systems:

Simple filters: synchronous processing
Complex multi-step filters: asynchronous orchestration using event-driven architecture

Domain Object / Value Object Design

public class Filter implements Filter {
    private final String id;
    private final String field;
    private final Operator operator;
    private final Object value;
    private final LogicalOperator logicalOperator;

    public Filter(String id, String field, Operator operator, Object value, LogicalOperator logicalOperator) {
        this.id = id;
        this.field = field;
        this.operator = operator;
        this.value = value;
        this.logicalOperator = logicalOperator;
    }

    @Override
    public String getId() {
        return id;
    }

    @Override
    public String getField() {
        return field;
    }

    @Override
    public Operator getOperator() {
        return operator;
    }

    @Override
    public Object getValue() {
        return value;
    }

    @Override
    public LogicalOperator getLogicalOperator() {
        return logicalOperator;
    }
}

public class Record {
    private final String id;
    private final Map<String, Object> data;

    public Record(String id, Map<String, Object> data) {
        this.id = id;
        this.data = new HashMap<>(data);
    }

    public String getId() {
        return id;
    }

    public Object getFieldValue(String field) {
        return data.get(field);
    }

    public Map<String, Object> getData() {
        return new HashMap<>(data);
    }
}

public class FilterEngineException extends RuntimeException {
    private final String errorCode;
    
    public FilterEngineException(String message) {
        super(message);
        this.errorCode = "FILTER_ENGINE_ERROR";
    }
    
    public FilterEngineException(String message, Throwable cause) {
        super(message, cause);
        this.errorCode = "FILTER_ENGINE_ERROR";
    }
    
    public String getErrorCode() {
        return errorCode;
    }
}

5. Concurrency & Thread Safety

Concurrency Model

The filtering engine will operate in a multi-threaded environment with the following concurrency model:

Thread pool executor for handling concurrent filter requests using a fixed-size thread pool (e.g., 16-32 threads)
Async/Await pattern for non-blocking I/O operations (e.g., database reads, external API calls)
Immutable filter configurations to prevent race conditions during concurrent access
Thread-local storage for per-request context (e.g., trace IDs, user sessions)

Race Condition Prevention Strategy

Race conditions will be mitigated through:

Immutable filter definitions - once created, filter configurations are never modified
Synchronized access to shared mutable state (e.g., filter statistics counters)
Atomic operations for updating shared metrics
Read-Write locks for filter metadata that may be updated during runtime

Optimistic vs. Pessimistic Locking Selection

Optimistic locking is selected for:

Filter configuration updates (version-based conflict detection)
Statistics aggregation operations
Metadata synchronization between nodes in distributed environments

Pessimistic locking is used for:

Real-time filter execution that requires exclusive access to data sources
Critical sections where concurrent modification of shared resources could cause inconsistency

Deadlock Prevention Strategy

Deadlock prevention mechanisms:

Lock ordering - all threads acquire locks in a consistent order (e.g., by filter ID, then by resource type)
Timeout-based lock acquisition - maximum 500ms for lock acquisition with retry logic
Lock hierarchy - define clear lock levels (global, filter-group, individual filter)
Deadlock detection - periodic check for cycles in lock wait graphs with automatic recovery

Idempotency Implementation for Concurrent Duplicate Requests

For concurrent duplicate filter requests:

Request deduplication using a distributed cache with TTL (30 seconds)
Idempotency keys generated from filter parameters and timestamp
Concurrent request queuing with result caching for identical requests
Atomic operation pattern for updating filter statistics to prevent race conditions

public class FilterEngine {
    private final Map<String, CompletableFuture<FilterResult>> activeRequests = new ConcurrentHashMap<>();
    private final ReadWriteLock statsLock = new ReentrantReadWriteLock();
    
    public CompletableFuture<FilterResult> processFilterAsync(FilterRequest request) {
        String idempotencyKey = generateIdempotencyKey(request);
        
        // Check if same request is already being processed
        return activeRequests.computeIfAbsent(idempotencyKey, key -> {
            CompletableFuture<FilterResult> future = CompletableFuture.supplyAsync(() -> {
                try {
                    // Use optimistic locking for filter stats updates
                    return executeFilterWithOptimisticLocking(request);
                } catch (Exception e) {
                    throw new RuntimeException("Filter execution failed", e);
                }
            });
            
            // Remove from active requests when done
            future.whenComplete((result, error) -> 
                activeRequests.remove(idempotencyKey));
            
            return future;
        });
    }
    
    private String generateIdempotencyKey(FilterRequest request) {
        return Hashing.murmur3_128()
            .hashString(request.toString() + System.currentTimeMillis(), StandardCharsets.UTF_8)
            .toString();
    }
    
    private FilterResult executeFilterWithOptimisticLocking(FilterRequest request) {
        // Use read-write lock for filter statistics
        statsLock.writeLock().lock();
        try {
            // Execute filter logic with version checking for concurrent updates
            return executeFilter(request);
        } finally {
            statsLock.writeLock().unlock();
        }
    }
}

6. Sequence Diagrams for Key Flows

sequenceDiagram
    participant Client
    participant APIGateway
    participant FilterService
    participant Cache
    participant Database
    participant EventPublisher

    Client->>APIGateway: POST /filters
    APIGateway->>FilterService: Validate and parse filter request
    FilterService->>Cache: Check if result cached
    Cache-->>FilterService: Return cached result or miss
    alt Cache miss
        FilterService->>Database: Query filtered data
        Database-->>FilterService: Return filtered results
        FilterService->>Cache: Store result in cache
        Cache-->>FilterService: Cache set success
    end
    FilterService-->>APIGateway: Return filtered results
    APIGateway-->>Client: HTTP 200 with filtered data

sequenceDiagram
    participant Client
    participant APIGateway
    participant FilterService
    participant Database
    participant EventPublisher

    Client->>APIGateway: POST /filters?async=true
    APIGateway->>FilterService: Validate and queue filter job
    FilterService->>Database: Store filter request in queue table
    Database-->>FilterService: Job queued successfully
    FilterService->>EventPublisher: Publish filter job event
    EventPublisher-->>FilterService: Event published
    FilterService-->>APIGateway: Return job ID and status
    APIGateway-->>Client: HTTP 202 with job ID

    Note over FilterService,EventPublisher: Async processing happens here
    FilterService->>Database: Poll for pending filter jobs
    Database-->>FilterService: Return job details
    FilterService->>Database: Execute filter query
    Database-->>FilterService: Return filtered results
    FilterService->>Database: Update job status to completed
    FilterService->>EventPublisher: Publish completion event
    EventPublisher-->>FilterService: Event published

sequenceDiagram
    participant Client
    participant APIGateway
    participant FilterService
    participant Cache
    participant Database
    participant EventPublisher

    Client->>APIGateway: POST /filters?async=true
    APIGateway->>FilterService: Validate and queue filter job
    FilterService->>Database: Store filter request in queue table
    Database-->>FilterService: Job queued successfully
    FilterService->>EventPublisher: Publish filter job event
    EventPublisher-->>FilterService: Event published
    FilterService-->>APIGateway: Return job ID and status
    APIGateway-->>Client: HTTP 202 with job ID

    Note over FilterService,EventPublisher: Async processing fails
    FilterService->>Database: Poll for pending filter jobs
    Database-->>FilterService: Return job details
    FilterService->>Database: Execute filter query
    Database-->>FilterService: Query fails with timeout
    FilterService->>Database: Update job status to failed
    FilterService->>EventPublisher: Publish failure event
    EventPublisher-->>FilterService: Event published
    FilterService->>Database: Store error details in job table


## 7. Caching Design

### Cache Strategy Selection

For the filtering engine, we'll use a **cache-aside pattern** with **LRU eviction policy**. This approach provides the best balance between simplicity, performance, and cache hit ratio for our use case where filter results are frequently accessed but not always hot.

### Cache Key Design

Cache keys will follow a structured namespace pattern:

filter-engine:{filter_type}:{filter_params_hash}:{result_format}


Example key structure:

filter-engine:user-roles:3a5f2c1b4d6e8f9a:v1


### TTL Strategy

| Data Type | TTL | Reason |
|-----------|-----|--------|
| Filter configurations | 24h | Configuration changes are infrequent |
| Filter results (cached queries) | 5m | Results can become stale quickly with changing data |
| Filter metadata (counts, cardinality) | 1h | Metadata changes less frequently than query results |
| User-specific filter preferences | 7d | User preferences are relatively stable |

### Cache Eviction Policy

We'll implement **LRU (Least Recently Used)** eviction with a **time-based expiration**. This ensures that:
1. Hot data stays in cache longer
2. Cold data is evicted efficiently
3. Memory usage remains bounded

### Cache Stampede Prevention

To prevent cache stampedes:
1. **Probabilistic early expiry**: Set cache TTL to 75% of configured TTL (e.g., 3.75m for 5m TTL)
2. **Distributed locks**: Use Redis distributed locks when cache miss occurs to prevent multiple concurrent computations
3. **Cache warming**: Pre-warm popular filter combinations during low-traffic periods

### Distributed Cache Architecture

We'll use **Redis Cluster** with:
- 3 master nodes for data distribution
- 3 replica nodes per shard for high availability
- Redis Sentinel for automatic failover
- Memory policy set to `allkeys-lru` with maxmemory configured to 4GB per node

### Cache Invalidation Strategy

**Event-driven invalidation** will be used for cache consistency:
1. When underlying data changes (user roles, filter configurations), invalidate related cache entries
2. Use Redis Pub/Sub to broadcast invalidation events across service instances
3. Implement cache invalidation with a **time-based soft TTL** (5-10% grace period) to handle eventual consistency

### Cache Implementation Example

```java
@Component
public class FilterResultCache {
    private final RedisTemplate<String, Object> redisTemplate;
    private final ObjectMapper objectMapper;
    
    public FilterResultCache(RedisTemplate<String, Object> redisTemplate, 
                           ObjectMapper objectMapper) {
        this.redisTemplate = redisTemplate;
        this.objectMapper = objectMapper;
    }
    
    public Optional<FilterResult> getCachedResult(String filterKey) {
        String cacheKey = "filter-engine:" + filterKey;
        Object cached = redisTemplate.opsForValue().get(cacheKey);
        
        if (cached != null) {
            try {
                return Optional.of(objectMapper.readValue(
                    objectMapper.writeValueAsString(cached), FilterResult.class));
            } catch (Exception e) {
                // Log and continue with cache miss
                log.warn("Failed to deserialize cached result for key: {}", cacheKey, e);
            }
        }
        return Optional.empty();
    }
    
    public void cacheResult(String filterKey, FilterResult result, Duration ttl) {
        String cacheKey = "filter-engine:" + filterKey;
        try {
            String jsonResult = objectMapper.writeValueAsString(result);
            redisTemplate.opsForValue().set(cacheKey, jsonResult, ttl);
        } catch (Exception e) {
            log.error("Failed to cache result for key: {}", cacheKey, e);
        }
    }
    
    public void invalidateFilterCache(String filterType, String filterParams) {
        String pattern = "filter-engine:" + filterType + ":" + filterParams + ":*";
        Set<String> keys = redisTemplate.keys(pattern);
        if (!keys.isEmpty()) {
            redisTemplate.delete(keys);
        }
    }
}

8. Repository / Data Access Layer

Repository Pattern Implementation

The repository pattern is implemented to encapsulate data access logic and provide a clean abstraction layer between the service layer and database operations. Each entity will have its own repository interface and implementation.

public interface FilterRepository {
    Filter save(Filter filter);
    Optional<Filter> findById(UUID id);
    List<Filter> findAll();
    void deleteById(UUID id);
    Page<Filter> findAll(Pageable pageable);
}

@Repository
public class FilterRepositoryImpl implements FilterRepository {
    private final EntityManager entityManager;
    private final QueryFactory queryFactory;

    public FilterRepositoryImpl(EntityManager entityManager, QueryFactory queryFactory) {
        this.entityManager = entityManager;
        this.queryFactory = queryFactory;
    }

    @Override
    public Filter save(Filter filter) {
        if (filter.getId() == null) {
            entityManager.persist(filter);
        } else {
            entityManager.merge(filter);
        }
        return filter;
    }

    @Override
    public Optional<Filter> findById(UUID id) {
        return Optional.ofNullable(entityManager.find(Filter.class, id));
    }

    @Override
    public List<Filter> findAll() {
        return entityManager.createQuery("SELECT f FROM Filter f", Filter.class)
                           .getResultList();
    }

    @Override
    public void deleteById(UUID id) {
        Filter filter = entityManager.find(Filter.class, id);
        if (filter != null) {
            entityManager.remove(filter);
        }
    }

    @Override
    public Page<Filter> findAll(Pageable pageable) {
        TypedQuery<Filter> query = entityManager.createQuery(
            "SELECT f FROM Filter f ORDER BY f.createdAt DESC", Filter.class);
        query.setFirstResult(pageable.getOffset());
        query.setMaxResults(pageable.getPageSize());
        return new PageImpl<>(query.getResultList(), pageable, countAll());
    }

    private long countAll() {
        return entityManager.createQuery("SELECT COUNT(f) FROM Filter f", Long.class)
                           .getSingleResult();
    }
}

ORM vs Raw Query Decisions

For the filtering engine, we use JPA/Hibernate as the ORM for most operations due to its type safety, ease of maintenance, and integration with Spring ecosystem. However, for complex filtering queries that require high performance or database-specific features, raw SQL is used:

@Repository
public class FilterRepositoryImpl {
    @PersistenceContext
    private EntityManager entityManager;

    // Use raw SQL for complex filtering operations that benefit from database-specific optimizations
    public List<FilterResult> executeComplexFilterQuery(FilterCriteria criteria) {
        String sql = """
            SELECT f.id, f.name, f.description, f.created_at
            FROM filters f
            WHERE f.status = :status
            AND (:category IS NULL OR f.category = :category)
            AND (:minValue IS NULL OR f.value >= :minValue)
            ORDER BY f.created_at DESC
            LIMIT :limit OFFSET :offset
            """;
        
        Query query = entityManager.createNativeQuery(sql);
        query.setParameter("status", criteria.getStatus());
        query.setParameter("category", criteria.getCategory());
        query.setParameter("minValue", criteria.getMinValue());
        query.setParameter("limit", criteria.getLimit());
        query.setParameter("offset", criteria.getOffset());

        return query.getResultList().stream()
                   .map(this::mapToFilterResult)
                   .collect(Collectors.toList());
    }

    private FilterResult mapToFilterResult(Object[] row) {
        return new FilterResult(
            (UUID) row[0],
            (String) row[1],
            (String) row[2],
            (Timestamp) row[3]
        );
    }
}

Read vs Write Separation

The filtering engine uses CQRS (Command Query Responsibility Segregation) pattern for read and write operations:

// Write side - used for creating/updating filters
@Repository
public class FilterWriteRepository {
    private final EntityManager entityManager;

    public FilterWriteRepository(EntityManager entityManager) {
        this.entityManager = entityManager;
    }

    public Filter save(Filter filter) {
        if (filter.getId() == null) {
            entityManager.persist(filter);
        } else {
            entityManager.merge(filter);
        }
        return filter;
    }

    public void delete(UUID id) {
        Filter filter = entityManager.find(Filter.class, id);
        if (filter != null) {
            entityManager.remove(filter);
        }
    }
}

// Read side - optimized for querying filters with complex conditions
@Repository
public class FilterReadRepository {
    private final EntityManager entityManager;

    public FilterReadRepository(EntityManager entityManager) {
        this.entityManager = entityManager;
    }

    // Optimized for read-heavy operations with indexes and projections
    public List<FilterProjection> findActiveFiltersWithMetadata(
            String category, Integer minPriority) {
        String jpql = """
            SELECT new com.example.filter.dto.FilterProjection(
                f.id, f.name, f.category, f.priority, f.metadata
            )
            FROM Filter f
            WHERE f.status = 'ACTIVE'
            AND (:category IS NULL OR f.category = :category)
            AND (:minPriority IS NULL OR f.priority >= :minPriority)
            ORDER BY f.priority DESC
            """;

        TypedQuery<FilterProjection> query = entityManager.createQuery(jpql, FilterProjection.class);
        query.setParameter("category", category);
        query.setParameter("minPriority", minPriority);

        return query.getResultList();
    }
}

Connection Pool Configuration

Connection pool is configured with HikariCP for optimal performance:

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

N+1 Query Prevention Strategy

To prevent N+1 query problems, we use eager loading and batch fetching:

@Entity
@Table(name = "filters")
@NamedEntityGraph(
    name = "Filter.withMetadata",
    attributeNodes = @NamedAttributeNode(value = "metadata", fetchType = FetchType.EAGER)
)
public class Filter {
    @Id
    private UUID id;

    @OneToMany(mappedBy = "filter", fetch = FetchType.LAZY, cascade = CascadeType.ALL)
    @BatchSize(size = 20) // Batch fetch to reduce queries
    private List<FilterMetadata> metadata;

    // Other fields...
}

// In service layer, explicitly fetch with entity graph
public List<Filter> getFiltersWithMetadata(List<UUID> filterIds) {
    String jpql = "SELECT f FROM Filter f FETCH ALL JOINS WHERE f.id IN :ids";
    TypedQuery<Filter> query = entityManager.createQuery(jpql, Filter.class);
    query.setParameter("ids", filterIds);
    return query.getResultList();
}

Batch Operation Design

Batch operations are implemented for bulk filter creation and updates:

@Repository
public class BatchFilterRepository {
    private final EntityManager entityManager;

    public BatchFilterRepository(EntityManager entityManager) {
        this.entityManager = entityManager;
    }

    @Transactional
    public void batchCreateFilters(List<Filter> filters) {
        int batchSize = 1000;
        for (int i = 0; i < filters.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, filters.size());
            List<Filter> batch = filters.subList(i, endIndex);
            
            // Use bulk insert with native query for better performance
            String sql = "INSERT INTO filters (id, name, category, status) VALUES ";
            // Build parameterized SQL string...
            
            Query query = entityManager.createNativeQuery(sql);
            // Set parameters for batch...
            
            query.executeUpdate();
        }
    }

    @Transactional
    public void batchUpdateFilterStatus(List<UUID> filterIds, String newStatus) {
        String jpql = "UPDATE Filter f SET f.status = :status WHERE f.id IN :ids";
        Query query = entityManager.createQuery(jpql);
        query.setParameter("status", newStatus);
        query.setParameter("ids", filterIds);
        query.executeUpdate();
    }
}

9. Asynchronous Processing & Messaging

Event-driven boundaries in this feature

The filtering engine is designed to process large volumes of data efficiently, with some operations being computationally intensive and time-consuming. The system will use a hybrid synchronous/asynchronous approach where:

Synchronous flows: Real-time filtering requests that must return results within strict latency SLAs (e.g., <200ms for simple filters)
Asynchronous flows: Heavy computation filtering operations (e.g., complex multi-dimensional aggregations, pattern matching) that can be queued and processed in the background

The key trigger for async processing is when a filter operation exceeds a configurable threshold (e.g., >500ms) or involves complex multi-step computations that can be parallelized.

Message schema design with sample payload

{
  "version": "1.0",
  "event_type": "filtering_job_created",
  "job_id": "f1a2b3c4-d5e6-f7g8-h9i0-j1k2l3m4n5o6",
  "filter_spec": {
    "name": "user_segmentation_filter",
    "conditions": [
      {
        "field": "age",
        "operator": "between",
        "values": [25, 40]
      },
      {
        "field": "location",
        "operator": "in",
        "values": ["NYC", "LA", "Chicago"]
      }
    ],
    "aggregations": [
      {
        "field": "purchase_amount",
        "function": "sum"
      }
    ]
  },
  "source_data": {
    "dataset_id": "user_events_2024",
    "partition_key": "2024-01-15"
  },
  "created_at": "2024-01-15T10:30:00Z",
  "priority": "high"
}

Dead letter queue strategy and poison message handling

For failed async filter jobs, we implement a dead-letter queue (DLQ) strategy:

Retry mechanism: Each failed job will be retried up to 3 times with exponential backoff (1s, 4s, 9s)
DLQ routing: Jobs that fail after max retries are moved to a DLQ topic with a 24-hour TTL
Manual intervention: Failed jobs in DLQ are flagged for manual review and can be reprocessed via admin UI
Monitoring: DLQ queue size and failure rate are monitored with alerts

At-least-once vs. exactly-once delivery

We choose at-least-once delivery for the filtering engine because:

The filtering system is idempotent - reprocessing a filter job with same inputs will produce identical results
The system uses job IDs to deduplicate processing
Performance overhead of exactly-once is not justified for this use case

Retry policy for consumers

public class FilterJobRetryPolicy {
    private static final int MAX_RETRIES = 3;
    private static final long BASE_DELAY_MS = 1000;
    private static final double BACKOFF_MULTIPLIER = 4.0;
    
    public boolean shouldRetry(int attempt, Exception error) {
        if (attempt >= MAX_RETRIES) return false;
        
        // Don't retry on certain non-recoverable errors
        if (error instanceof ValidationException || 
            error instanceof IllegalArgumentException) {
            return false;
        }
        
        return true;
    }
    
    public long getDelay(int attempt) {
        return (long) Math.pow(BACKOFF_MULTIPLIER, attempt) * BASE_DELAY_MS;
    }
}

Outbox pattern for reliable event publishing

We implement the outbox pattern to ensure filtering job events are published reliably:

public class FilterJobOutbox {
    private final EventRepository eventRepository;
    private final MessagePublisher messagePublisher;
    
    public void publishFilterJobEvent(FilterJob job) {
        // 1. Persist event to outbox table
        OutboxEvent outboxEvent = new OutboxEvent(
            job.getId(), 
            "filtering_job_created", 
            serialize(job),
            System.currentTimeMillis()
        );
        
        eventRepository.save(outboxEvent);
        
        // 2. Publish to message broker (idempotent operation)
        try {
            messagePublisher.publish(outboxEvent);
            
            // 3. Mark as published in outbox
            eventRepository.markAsPublished(outboxEvent.getId());
        } catch (Exception e) {
            // Handle publish failure - retry logic or DLQ
            eventRepository.markAsFailed(outboxEvent.getId());
        }
    }
}

Sequence diagram for async flows

sequenceDiagram
    participant Client
    participant FilterEngine
    participant JobQueue
    participant FilterWorker
    participant EventPublisher
    
    Client->>FilterEngine: POST /filter/async
    FilterEngine->>JobQueue: Queue FilterJob
    JobQueue-->>FilterEngine: Acknowledge
    FilterEngine-->>Client: 202 Accepted, job_id
    
    loop Processing
        JobQueue->>FilterWorker: Dequeue FilterJob
        FilterWorker->>FilterWorker: Process filter logic
        FilterWorker->>EventPublisher: Publish job_completed event
    end
    
    EventPublisher-->>Client: Notify completion via webhook

10. Error Handling Design

Error Classification Hierarchy

The filtering engine implements a hierarchical error classification system to enable proper handling and observability:

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

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

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

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

public class FilterNotFoundException extends BaseException {
    public FilterNotFoundException(String filterId) {
        super("FILTER_NOT_FOUND", 404, "Filter with ID " + filterId + " not found");
    }
}

public class InternalServerException extends BaseException {
    public InternalServerException(String message, Throwable cause) {
        super("INTERNAL_ERROR", 500, message);
    }
}

public class ServiceUnavailableException extends BaseException {
    public ServiceUnavailableException(String message) {
        super("SERVICE_UNAVAILABLE", 503, message);
    }
}

Error Propagation Strategy

Errors follow a layered propagation model:

Domain layer: Throws specific domain exceptions (e.g., ValidationException)
Service layer: Wraps domain errors in service-specific exceptions or rethrows as InternalServerException
Controller layer: Maps to appropriate HTTP responses with standardized error format

Retry Strategy

For transient infrastructure issues, a retry mechanism is implemented:

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

    public <T> T executeWithRetry(Supplier<T> operation) {
        int attempts = 0;
        while (true) {
            try {
                return operation.get();
            } catch (Exception e) {
                attempts++;
                if (attempts >= MAX_RETRIES || !isRetryable(e)) {
                    throw new InternalServerException("Operation failed after " + attempts + " attempts", e);
                }
                
                long delay = (long) (BASE_DELAY_MS * Math.pow(2, attempts - 1));
                delay += (long) (delay * JITTER_FACTOR * (Math.random() - 0.5));
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new InternalServerException("Retry interrupted", ie);
                }
            }
        }
    }

    private boolean isRetryable(Exception e) {
        return e instanceof ServiceUnavailableException || 
               e instanceof InternalServerException ||
               (e.getCause() != null && isRetryable(e.getCause()));
    }
}

Circuit Breaker Design

For external dependencies like database or caching systems, a circuit breaker is implemented:

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

    public enum CircuitState {
        CLOSED, OPEN, HALF_OPEN
    }

    public <T> T execute(Supplier<T> operation) throws Exception {
        switch (state) {
            case CLOSED:
                return executeClosed(operation);
            case OPEN:
                if (System.currentTimeMillis() - lastFailureTime > timeoutMs) {
                    state = CircuitState.HALF_OPEN;
                    return executeHalfOpen(operation);
                }
                throw new ServiceUnavailableException("Circuit breaker is OPEN");
            case HALF_OPEN:
                return executeHalfOpen(operation);
        }
        throw new InternalServerException("Unexpected circuit state");
    }

    private <T> T executeClosed(Supplier<T> operation) throws Exception {
        try {
            T result = operation.get();
            failureCount.set(0);
            return result;
        } catch (Exception e) {
            failureCount.incrementAndGet();
            if (failureCount.get() >= failureThreshold) {
                state = CircuitState.OPEN;
                lastFailureTime = System.currentTimeMillis();
            }
            throw e;
        }
    }

    private <T> T executeHalfOpen(Supplier<T> operation) throws Exception {
        try {
            T result = operation.get();
            state = CircuitState.CLOSED;
            failureCount.set(0);
            return result;
        } catch (Exception e) {
            state = CircuitState.OPEN;
            lastFailureTime = System.currentTimeMillis();
            throw e;
        }
    }
}

Fallback Behavior Design

For critical operations, fallbacks are provided:

public class FilterService {
    private final CircuitBreaker circuitBreaker = new CircuitBreaker(5, 30000);
    
    public FilterResult applyFilter(String filterId, FilterContext context) {
        try {
            return circuitBreaker.execute(() -> {
                Filter filter = filterRepository.findById(filterId);
                if (filter == null) {
                    throw new FilterNotFoundException(filterId);
                }
                return filter.apply(context);
            });
        } catch (FilterNotFoundException e) {
            // Fallback to default filter behavior or return empty result
            return new FilterResult(Collections.emptyList(), Collections.emptyMap());
        } catch (Exception e) {
            // Log and return error response
            logger.error("Failed to apply filter: " + filterId, e);
            throw new InternalServerException("Filter processing failed", e);
        }
    }
}

Error Logging Format

Structured logging with trace context:

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

    public void logError(String operation, String errorCode, String message, 
                        Map<String, Object> context, Throwable cause) {
        Map<String, Object> logContext = new HashMap<>(context);
        logContext.put("operation", operation);
        logContext.put("errorCode", errorCode);
        logContext.put("message", message);
        logContext.put("timestamp", Instant.now().toString());
        
        if (cause != null) {
            logContext.put("errorType", cause.getClass().getSimpleName());
            logContext.put("stackTrace", cause.getStackTrace());
        }
        
        logger.error("Filter engine error: {}", logContext);
    }
}

Alerting Thresholds

Critical errors trigger alerts:

Error Type	Alert Threshold	Alert Channel
FilterNotFoundException	10 occurrences in 5 minutes	On-call
InternalServerException	5 occurrences in 1 minute	Slack
ServiceUnavailableException	3 occurrences in 2 minutes	PagerDuty

Error Response Format

Standardized JSON error response:

{
  "error": {
    "code": "FILTER_NOT_FOUND",
    "message": "Filter with ID abc123 not found",
    "httpStatus": 404,
    "timestamp": "2023-10-15T10:30:00Z",
    "traceId": "abc123def456"
  }
}

11. Security Design

Input Validation and Sanitization Strategy

All API inputs are validated using a combination of:

Custom validation annotations (e.g., @NotBlank, @Min, @Max) for basic constraints
Apache Commons Validator for complex patterns (email, phone, URL)
OWASP Java HTML Sanitizer for any user-provided content that may be rendered in UI
Custom regex patterns for domain-specific constraints (e.g., filter expression syntax)

public class FilterRequest {
    @NotBlank
    @Size(max = 1000)
    private String filterExpression;
    
    @Min(0)
    @Max(10000)
    private Integer limit;
    
    // getters/setters
}

SQL Injection Prevention

Prepared statements for all database queries using parameterized queries
Whitelisting of filter fields - only predefined columns can be used in filters
No dynamic SQL construction from user input
Query plan validation to detect unexpected query patterns

Authentication Middleware Design

JWT-based authentication with 24-hour expiration for API tokens
OAuth2 integration for third-party service authentication
API key validation for client applications with rate limiting per key
Authentication middleware that intercepts all requests and validates token signature

Authorization Enforcement

Role-Based Access Control (RBAC) with hierarchical permissions:
- FILTER_READ: Read access to filter definitions
- FILTER_WRITE: Create/update/delete filters
- FILTER_EXECUTE: Execute filters on data sets
Permission checking at service layer before processing:

public boolean hasPermission(String userId, String requiredPermission) {
    return userPermissionService.hasPermission(userId, requiredPermission);
}

Sensitive Data Encryption

At-rest encryption: AES-256 GCM for filter definitions and execution logs
In-transit encryption: TLS 1.3 for all API communications
Key rotation: 90-day automatic key rotation with backup keys

PII Handling and Masking Strategy

PII data fields in filter expressions are masked in logs and audit trails
Redaction policy: Any field containing SSN, email, or phone numbers are replaced with *** in logs
Data classification tags applied to filter definitions to indicate PII content

Secrets Management

AWS Secrets Manager integration for storing:
- Database credentials
- Third-party API keys
- Encryption keys (rotated every 90 days)
Secrets rotation handled via Lambda functions with CloudWatch triggers
No hardcoded secrets in source code or configuration files

Audit Logging Design

Audit log entries capture:
- User ID (masked for PII)
- Timestamp
- Operation type (create, update, delete, execute)
- Filter ID and expression
- IP address (masked for privacy)
- Response status code and duration

{
  "timestamp": "2023-12-01T10:30:45.123Z",
  "userId": "user_***",
  "operation": "FILTER_EXECUTE",
  "filterId": "filter_abc123",
  "expression": "status:active AND age:>25",
  "ipAddress": "10.0.0.***",
  "statusCode": 200,
  "durationMs": 45
}

Retention policy: 90 days for audit logs with daily backups to secure S3 bucket
Log aggregation: Centralized logging via CloudWatch with alerting for suspicious patterns

12. Testing Strategy

Unit Testing Scope and Coverage Targets (%)

Service Layer: 85% code coverage with focus on business logic, validation, and error paths
Repository Layer: 90% coverage with all CRUD operations and query execution paths
Domain Objects: 100% coverage for value objects, builders, and validation logic
Event Handlers: 80% coverage for async processing and error handling
API Layer: 75% coverage for request/response mapping, validation, and error responses
Core Filtering Logic: 95% coverage including all filter types, combinations, and edge cases

Integration Testing Strategy

Database Integration: Use Testcontainers with PostgreSQL 13+ for real database testing with schema migrations
External Service Mocks: WireMock for external API integrations (e.g., user service, analytics)
Event Broker: Embedded Kafka or Redis Streams for async flow testing
Caching Layer: In-memory Redis mock for cache behavior verification
Authentication Integration: Mock OAuth2 server with test tokens for auth flow validation

Contract Testing

API Contracts: Use Pact for consumer-driven contract testing between filtering engine and client services
Event Contracts: Define Avro schemas with schema registry integration for event contract validation
External API Contracts: Mock external service contracts using WireMock or similar tools

Performance Testing Tool

Load Testing: k6 for concurrent request testing with realistic user scenarios (1000+ concurrent users)
Stress Testing: JMeter for long-running stress scenarios (10k+ requests over 24h)
Key Performance Scenarios:
- 500 concurrent filter requests with 100ms p95 latency target
- 5000+ requests per minute with 99% success rate
- Memory usage under 2GB for 10k concurrent filters

Test Data Management Strategy

Fixtures: YAML-based fixtures for common filter configurations and test data sets
Factories: Factory classes using Builder pattern to generate realistic test data with various filter combinations
Seeding: Database seed scripts for initial state setup including sample entities and filter configurations
Data Isolation: Per-test database schema or transaction rollback for clean test environments

// Example unit test for filter validation logic
@Test
public void testFilterValidation_InvalidOperator() {
    FilterRequest request = new FilterRequest();
    FilterCondition condition = new FilterCondition();
    condition.setField("status");
    condition.setOperator("INVALID_OP");
    condition.setValue("ACTIVE");
    
    assertThrows(InvalidFilterException.class, () -> {
        filterService.validateFilter(request);
    });
}

// Example integration test for database query
@Test
public void testFilterQueryWithDatabase() {
    // Given: Test data in database
    User user = new User("user1", "ACTIVE");
    userRepository.save(user);
    
    // When: Filter query executed
    List<User> result = filterService.applyFilter(
        new FilterRequest()
            .withCondition(new FilterCondition("status", "EQUALS", "ACTIVE"))
    );
    
    // Then: Correct results returned
    assertThat(result).hasSize(1);
    assertThat(result.get(0).getStatus()).isEqualTo("ACTIVE");
}

13. Observability Design

Structured Logging Schema

All logs must follow a structured JSON format with consistent fields to enable programmatic parsing and filtering. The schema includes:

{
  "timestamp": "2023-10-27T14:30:45.123Z",
  "level": "INFO",
  "service": "filtering-engine",
  "trace_id": "12345678-1234-1234-1234-123456789012",
  "span_id": "abcdef12-3456-7890-abcd-ef1234567890",
  "user_id": "user-123",
  "request_id": "req-456",
  "operation": "filterItems",
  "entity_id": "filter-789",
  "duration_ms": 150,
  "error_code": "FILTERING_ERROR",
  "error_message": "Invalid filter condition",
  "stack_trace": "...",
  "payload": {
    "filters": {
      "category": "electronics"
    }
  },
  "metadata": {
    "source": "api",
    "version": "v1"
  }
}

PII masking rules:

User IDs and session tokens are masked in logs
Email addresses are truncated (e.g., "user@example.com" → "u***@example.com")
Credit card numbers are fully redacted

Distributed Tracing Instrumentation

Using OpenTelemetry, we instrument key operations with custom spans:

Span Types:

filtering_engine.process_request - Root span for incoming filter requests
filtering_engine.apply_filters - Span for applying individual filters
filtering_engine.fetch_data - Span for database queries
filtering_engine.cache_hit - Span for cache hits vs misses

Key Attributes Captured:

Span currentSpan = Span.current();
currentSpan.setAttribute("filtering_engine.filter_type", "category");
currentSpan.setAttribute("filtering_engine.filter_value", "electronics");
currentSpan.setAttribute("filtering_engine.result_count", 42);
currentSpan.setAttribute("filtering_engine.cache_hit", true);

Metrics Design

RED Method Metrics:

Metric Name	Type	Labels	Description
`filtering_engine_requests_total`	Counter	`operation`, `status`, `error_code`	Total filter requests
`filtering_engine_request_duration_seconds`	Histogram	`operation`, `status`	Request processing time
`filtering_engine_cache_hits_total`	Counter	`type`, `source`	Cache hit counts
`filtering_engine_cache_misses_total`	Counter	`type`, `source`	Cache miss counts
`filtering_engine_errors_total`	Counter	`error_type`, `operation`	Error counts by type

Example Histogram Configuration:

Histogram requestDuration = OpenTelemetry.get().getMeter("filtering-engine")
    .histogramBuilder("filtering_engine_request_duration_seconds")
    .setDescription("Duration of filtering requests in seconds")
    .setUnit("s")
    .build();

Health Check Endpoint Design

Endpoints:

/health - Application liveness check (basic service status)
/ready - Readiness check (dependencies and caches ready)
/live - Health status including database connectivity
/startup - Startup completion verification

Sample Response:

{
  "status": "healthy",
  "timestamp": "2023-10-27T14:30:45.123Z",
  "components": {
    "database": {
      "status": "healthy",
      "latency_ms": 15
    },
    "cache": {
      "status": "healthy",
      "latency_ms": 2
    }
  },
  "version": "1.0.0"
}

Alerting Thresholds

Critical Alerts (escalation to on-call):

filtering_engine_requests_total with status="error" > 100 requests/minute
filtering_engine_request_duration_seconds p99 > 5 seconds
filtering_engine_cache_misses_total rate > 1000/minute

Warning Alerts:

filtering_engine_request_duration_seconds p95 > 2 seconds
filtering_engine_cache_hit_ratio < 0.8 (80%)
Database connection pool usage > 90%

Escalation Path:

Slack channel for warnings
PagerDuty for critical alerts (5-minute escalation window)
On-call engineer must acknowledge within 10 minutes

14. Open Questions, Risks & Trade-offs

Decisions requiring product / engineering alignment

Filter complexity vs. performance trade-off: Should we support complex nested filters with AND/OR logic, or limit to simple field-value comparisons? Complex filters increase query processing time and memory usage but provide better user experience.
Real-time vs. batch filtering: Should filtering results be computed in real-time or pre-computed and cached? Real-time provides fresh data but has higher latency; batch processing offers better performance but introduces staleness.
Filter configuration persistence: Should filter configurations be stored in database or managed via external configuration service? Database storage provides versioning and audit trail but adds complexity; external service offers flexibility but introduces dependency.

Known trade-offs and justifications

Trade-off	Option 1	Option 2	Justification
Filter expression parsing	AST-based parser	Regex-based	AST provides better extensibility and error handling for complex expressions, while regex is simpler but harder to maintain
Filter caching strategy	In-memory LRU cache	Distributed Redis cache	In-memory is faster but limited to single instance; Redis provides horizontal scaling but introduces network latency
Filter validation approach	Client-side validation + server-side	Server-side only	Client-side improves UX but can be bypassed; server-side ensures data integrity
Filter result aggregation	In-memory merge	Database-level grouping	In-memory is faster for small datasets but memory-intensive; database grouping scales better but requires complex queries

Risks and mitigation strategies

Performance degradation under high load: Complex filter expressions can cause CPU and memory spikes. Mitigation: Implement query timeout limits, circuit breaker patterns, and resource monitoring.
Security vulnerabilities from dynamic filter generation: User-provided filter expressions can lead to injection attacks. Mitigation: Implement strict input validation, sanitize all user inputs, and use whitelisted filter operators.
Cache invalidation complexity: Filters that depend on multiple data sources can cause stale cache issues. Mitigation: Implement event-driven cache invalidation with proper TTLs and audit trails.
Scalability limitations: Monolithic filter engine can become a bottleneck. Mitigation: Design modular architecture with horizontal scaling capabilities and microservices pattern.

Technical debt acknowledged

Hardcoded filter operator support: Current implementation only supports basic operators (equals, not equals, greater than, etc.). Future work needed to support custom operator plugins and extensible operator registration.
Limited filter expression nesting: Expression nesting depth is currently capped at 3 levels, which may not be sufficient for complex business logic. This limitation should be revisited once production usage patterns are understood.
No filter composition support: Filters cannot currently be composed or reused across different queries. This is a major limitation for complex filtering scenarios and requires architectural refactoring to support filter templates or composition patterns.
Inconsistent error handling: Different filter types return different error formats, making client-side error handling inconsistent. Standardized error response format should be implemented for better developer experience.
Lack of filter performance monitoring: No built-in metrics to track filter execution times or resource usage. This makes it difficult to identify performance bottlenecks and optimize filter expressions at scale.