Production-Ready Event Ingestion Handler: Validation, Idempotent Writes, DLQ, Metrics

Problem Statement

You are building an event ingestion service for a content platform that receives document interaction events from upstream streaming sources (e.g., Kafka or webhooks). Due to network retries, upstream producers frequently send duplicate payloads or malformed data.

Implement an EventIngestionHandler that processes incoming raw event payloads safely, writes valid records to a database, and handles failures according to production standards.

Requirements

1. Validation:

   • event_id: non-empty str (strict type; e.g., numbers or other types must fail).
   • user_id: positive int (user_id > 0). Strings like "123" must be rejected to prevent type coercion bugs in downstream relational schemas.
   • event_type: str belonging to an allowed set: {"view", "bookmark", "highlight", "download"}.
   • timestamp: positive int representing epoch milliseconds (timestamp > 0).
   • payload: must be a dict (can be empty).
   • If any validation rule fails, record a metric, publish the raw event to the DeadLetterQueue with an error reason, and return status "REJECTED".

2. Idempotency:

   • If an event_id has already been written to the database, do not write it again. Emit a duplicate metric and return status "DUPLICATE".

3. Persistence and Retries:

   • Call database.insert(event). If a transient DatabaseError is raised, retry up to max_retries (default 2 retries, total 3 attempts). If all attempts fail, log the error, emit a failure metric, publish to the DeadLetterQueue, and return status "FAILED".

4. Metrics:

   • Emit metric counts for: "events.received", "events.success", "events.duplicate", "events.validation_failed", and "events.dlq_routed".

Interfaces

class Database:
    def exists(self, event_id: str) -> bool: ...
    def insert(self, event: dict) -> None: ...  # May raise DatabaseError

class DeadLetterQueue:
    def send(self, raw_event: dict, reason: str) -> None: ...

class MetricsClient:
    def increment(self, metric_name: str, count: int = 1) -> None: ...

class DatabaseError(Exception):
    pass

class EventIngestionHandler:
    def __init__(self, db: Database, dlq: DeadLetterQueue, metrics: MetricsClient, max_retries: int = 2):
        ...

    def handle(self, raw_event: dict) -> str:
        """
        Processes raw_event and returns one of: 'SUCCESS', 'DUPLICATE', 'REJECTED', 'FAILED'
        """
        ...

Example 1

event = {
    "event_id": "evt_001",
    "user_id": 42,
    "event_type": "view",
    "timestamp": 1700000000000,
    "payload": {"document_id": "doc_999"}
}
# Result: 'SUCCESS'
# Database receives insert, metrics 'events.received' and 'events.success' incremented.

Example 2

event = {
    "event_id": "evt_002",
    "user_id": "42",  # String instead of int
    "event_type": "bookmark",
    "timestamp": 1700000000000,
    "payload": {}
}
# Result: 'REJECTED'
# DLQ receives payload with reason 'invalid user_id: expected int'.
# Metrics 'events.received', 'events.validation_failed', and 'events.dlq_routed' incremented.