feat: add cdr processing worker and dlq routing

.env.sample

@@ -4,7 +4,11 @@
 
 REDPANDA_VERSION=v26.2.3
 
-KAFKA_BOOTSTRAP_SERVERS=redpanda:9092
+# Used by Python applications running on the host.
+KAFKA_BOOTSTRAP_SERVERS=localhost:19092
+
+# Used by containers communicating inside the Docker network.
+KAFKA_INTERNAL_BOOTSTRAP_SERVERS=redpanda:9092
 
 # Raw CDR topic
 KAFKA_RAW_TOPIC=cdr.raw
@@ -19,9 +23,35 @@ KAFKA_DLQ_TOPIC_PARTITIONS=1
 KAFKA_DLQ_TOPIC_REPLICATION_FACTOR=1
 KAFKA_DLQ_TOPIC_RETENTION_MS=604800000
 
+
+# =========================
+# Processing Worker
+# =========================
+
+WORKER_INSTANCE_ID=worker-1
+
+KAFKA_AUTO_OFFSET_RESET=earliest
+KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS=1
+
+KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS=5
+
 KAFKA_CONSUMER_GROUP=cdr-processing-workers
 
 
+# =========================
+# API ingestion
+# =========================
+
+API_QUEUE_SIZE=10000
+
+KAFKA_DELIVERY_TIMEOUT_SECONDS=4
+KAFKA_MESSAGE_TIMEOUT_MS=3000
+
+KAFKA_LINGER_MS=5
+KAFKA_BATCH_SIZE=131072
+KAFKA_PRODUCER_MAX_MESSAGES=100000
+
+
 # =========================
 # ClickHouse
 # =========================

README.md

@@ -4,19 +4,192 @@ High-throughput telecom Call Detail Record ingestion and analytics pipeline.
 
 ## Architecture
 
-The system will consist of:
+The system consists of:
 
-1. FastAPI ingestion API
-2. Kafka-compatible Redpanda message broker
-3. Processing and normalization worker
-4. ClickHouse analytics database
+* FastAPI ingestion API
+* Redpanda as a Kafka-compatible message broker
+* Processing and normalization worker
+* ClickHouse analytical storage
+* Dead Letter Queue for failed records
 
-Data flow:
+Current data flow:
 
-Network Element → Ingestion API → Redpanda → Worker → ClickHouse
+```text
+Network Element
+    ↓
+FastAPI API
+    ↓
+Validation
+    ↓
+Bounded Async Queue
+    ↓
+Redpanda / cdr.raw
+    ↓
+Processing Worker
+    ↓
+Validation + Normalization
+    ├── valid → processed by worker
+    └── invalid → cdr.raw.dlq
+```
 
-Malformed or unprocessable records are routed by the worker to a Dead Letter Queue.
+Infrastructure already available:
 
-## Development Status
+```text
+Redpanda
+├── cdr.raw
+└── cdr.raw.dlq
 
-Currently under development.
+ClickHouse
+└── telecom.cdr_records
+```
+
+worker processing is not connected yet.
+
+## Current Features
+
+* Dockerized Redpanda and ClickHouse
+* Automatic and idempotent Kafka topic provisioning
+* Automatic ClickHouse schema initialization
+* Persistent Redpanda and ClickHouse volumes
+* Raw and normalized CDR domain models
+* Required-field and type validation
+* RAT validation and normalization
+* E.164 phone-number normalization
+* Deterministic UUIDv5 `cdr_id` generation
+* Epoch and traffic-value validation
+* FastAPI `POST /api/v1/cdr/raw`
+* FastAPI `GET /health`
+* Structured JSON error responses
+* Unit and API integration tests
+* Bounded asynchronous ingestion queue for backpressure
+* Long-lived asynchronous Kafka producer
+* Kafka delivery acknowledgement before `202 Accepted`
+* `429 Too Many Requests` on ingestion queue saturation
+* `503 Service Unavailable` on Kafka delivery failure
+* Kafka consumer with manual offset management
+* Worker-side JSON decoding and validation
+* Worker-side CDR normalization
+* DLQ routing for malformed and unprocessable Kafka records
+* DLQ records with failure reason, failed field, timestamp, and worker instance
+* Kafka offsets committed only after the current processing outcome succeeds
+
+## Project Structure
+
+```text
+telecom-cdr-pipeline/
+├── infrastructure/
+│   └── clickhouse/
+│       └── init.sql
+│
+├── scripts/
+│   ├── sample-valid-cdr.json
+│   └── sample-invalid-rat.json
+│
+├── src/
+│   └── core/
+│       ├── __init__.py
+│       ├── config.py
+│       │
+│       ├── api/
+│       │   ├── __init__.py
+│       │   ├── kafka_publisher.py
+│       │   ├── main.py
+│       │   ├── publisher.py
+│       │   └── routes.py
+│       │
+│       ├── domain/
+│       │   ├── __init__.py
+│       │   ├── errors.py
+│       │   ├── models.py
+│       │   ├── normalization.py
+│       │   └── validation.py
+│       │
+│       └── worker/
+│           ├── __init__.py
+│           ├── consumer.py
+│           ├── dlq.py
+│           ├── main.py
+│           └── processor.py
+│
+├── tests/
+│   ├── integration/
+│   │   └── test_ingestion_api.py
+│   │
+│   └── unit/
+│       ├── test_dlq.py
+│       ├── test_normalization.py
+│       ├── test_project_setup.py
+│       ├── test_validation.py
+│       └── test_worker_processor.py
+│
+├── .env.example
+├── .gitignore
+├── docker-compose.yml
+├── pyproject.toml
+└── README.md
+```
+
+## Configuration
+
+Configuration is provided through environment variables documented in `.env.example`.
+
+This includes:
+
+* Redpanda version and topic settings
+* Kafka broker and consumer configuration
+* ClickHouse connection settings
+* Worker batch size and flush interval
+
+## Running Infrastructure
+
+```bash
+docker compose up -d
+```
+
+Verify services:
+
+```bash
+docker compose ps -a
+```
+
+Expected:
+
+```text
+Redpanda       healthy
+redpanda-init  exited successfully
+ClickHouse     healthy
+```
+
+## Running the API
+
+```bash
+uvicorn core.api.main:app --reload --host 0.0.0.0 --port 8000
+```
+
+Available endpoints:
+
+```text
+GET  /health
+POST /api/v1/cdr/raw
+```
+
+A valid CDR currently returns:
+
+```json
+{
+  "status": "accepted",
+  "raw_id": "raw-test-001"
+}
+```
+
+Invalid payloads return structured `400` responses.
+
+## Testing
+
+Run:
+
+```bash
+pytest -v
+```
+
+The current test suite covers domain validation, normalization, deterministic ID generation, and API request validation.

docker-compose.yml

@@ -56,7 +56,7 @@ services:
         condition: service_healthy
 
     environment:
-      KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_BOOTSTRAP_SERVERS}
+      KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS}
 
       KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC}
       KAFKA_RAW_TOPIC_PARTITIONS: ${KAFKA_RAW_TOPIC_PARTITIONS}

pyproject.toml

@@ -3,11 +3,18 @@ name = "telecom-cdr-pipeline"
 version = "0.1.0"
 description = "High-throughput telecom CDR ingestion and analytics pipeline"
 requires-python = ">=3.11"
-dependencies = []
+dependencies = [
+    "phonenumbers>=8.13",
+    "fastapi>=0.115,<1.0",
+    "uvicorn[standard]>=0.30,<1.0",
+    "confluent-kafka>=2.15,<3.0",
+    "pydantic-settings>=2.0,<3.0",
+]
 
 [project.optional-dependencies]
 dev = [
     "pytest>=8.0",
+    "httpx2>=2.4,<3.0",
 ]
 
 [build-system]

scripts/sample-valid-cdr.json

@@ -0,0 +1,14 @@
+{
+  "raw_id": "raw-test-001",
+  "src_ip": "10.240.12.1",
+  "device_id": "310410998877665",
+  "caller_num": "0019875551234",
+  "svc": "4G_DATA",
+  "up_bytes": 1048576,
+  "down_bytes": 4194304,
+  "rat": 6,
+  "lac": "4021",
+  "ci": "8812",
+  "epoch_start": 1789900000,
+  "epoch_end": 1789900300
+}

src/core/api/kafka_publisher.py

@@ -0,0 +1,184 @@
+import asyncio
+import json
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from confluent_kafka import KafkaException, Producer
+
+from core.api.publisher import (
+    IngestionQueueFull,
+    KafkaPublishError,
+)
+from core.config import Settings
+
+
+@dataclass(slots=True)
+class PendingMessage:
+    key: str
+    value: bytes
+    result: asyncio.Future[None]
+
+
+class KafkaPublisher:
+    def __init__(self, settings: Settings) -> None:
+        self._settings = settings
+
+        self._queue: asyncio.Queue[PendingMessage] = asyncio.Queue(
+            maxsize=settings.api_queue_size,
+        )
+
+        self._producer = Producer(
+            {
+                "bootstrap.servers": settings.kafka_bootstrap_servers,
+
+                "client.id": "cdr-ingestion-api",
+
+                "acks": "all",
+                "enable.idempotence": True,
+
+                "linger.ms": settings.kafka_linger_ms,
+                "batch.size": settings.kafka_batch_size,
+
+                "compression.type": "lz4",
+
+                "queue.buffering.max.messages":
+                    settings.kafka_producer_max_messages,
+
+                "message.timeout.ms":
+                    settings.kafka_message_timeout_ms,
+            }
+        )
+
+        self._task: asyncio.Task[None] | None = None
+        self._stopping = False
+
+    async def start(self) -> None:
+        if self._task is not None:
+            return
+
+        self._stopping = False
+
+        self._task = asyncio.create_task(
+            self._run(),
+            name="kafka-publisher",
+        )
+
+    async def publish(
+        self,
+        payload: Mapping[str, Any],
+        key: str,
+    ) -> None:
+        loop = asyncio.get_running_loop()
+
+        result: asyncio.Future[None] = loop.create_future()
+
+        value = json.dumps(
+            payload,
+            separators=(",", ":"),
+            ensure_ascii=False,
+        ).encode("utf-8")
+
+        message = PendingMessage(
+            key=key,
+            value=value,
+            result=result,
+        )
+
+        try:
+            self._queue.put_nowait(message)
+
+        except asyncio.QueueFull as exc:
+            raise IngestionQueueFull(
+                "Ingestion queue is full."
+            ) from exc
+
+        try:
+            await asyncio.wait_for(
+                result,
+                timeout=self._settings.kafka_delivery_timeout_seconds,
+            )
+
+        except TimeoutError as exc:
+            raise KafkaPublishError(
+                "Kafka delivery acknowledgement timed out."
+            ) from exc
+
+    def _send_to_kafka(
+        self,
+        pending: PendingMessage,
+    ) -> None:
+        def on_delivery(error, message) -> None:
+            if pending.result.done():
+                return
+
+            if error is not None:
+                pending.result.set_exception(
+                    KafkaPublishError(str(error))
+                )
+                return
+
+            pending.result.set_result(None)
+
+        try:
+            self._producer.produce(
+                topic=self._settings.kafka_raw_topic,
+                key=pending.key,
+                value=pending.value,
+                on_delivery=on_delivery,
+            )
+
+        except BufferError as exc:
+            if not pending.result.done():
+                pending.result.set_exception(
+                    KafkaPublishError(
+                        "Kafka producer queue is full."
+                    )
+                )
+
+        except KafkaException as exc:
+            if not pending.result.done():
+                pending.result.set_exception(
+                    KafkaPublishError(str(exc))
+                )
+
+    async def _run(self) -> None:
+        while (
+            not self._stopping
+            or not self._queue.empty()
+            or len(self._producer) > 0
+        ):
+            pending: PendingMessage | None = None
+
+            try:
+                pending = await asyncio.wait_for(
+                    self._queue.get(),
+                    timeout=0.05,
+                )
+
+            except TimeoutError:
+                pass
+
+            if pending is not None:
+                self._send_to_kafka(pending)
+                self._queue.task_done()
+
+            # Triggers Kafka delivery callbacks.
+            self._producer.poll(0)
+
+            # Give the event loop a chance to handle HTTP requests.
+            await asyncio.sleep(0)
+
+    async def stop(self) -> None:
+        if self._task is None:
+            return
+
+        self._stopping = True
+
+        await self._task
+
+        self._task = None
+
+    @property
+    def queue_size(self) -> int:
+        return self._queue.qsize()

src/core/api/main.py

@@ -0,0 +1,53 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from core.api.kafka_publisher import KafkaPublisher
+from core.api.publisher import Publisher
+from core.api.routes import router
+from core.config import get_settings
+
+
+def create_app(
+    publisher: Publisher | None = None,
+) -> FastAPI:
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        active_publisher = publisher
+
+        if active_publisher is None:
+            active_publisher = KafkaPublisher(
+                get_settings()
+            )
+
+        app.state.publisher = active_publisher
+
+        await active_publisher.start()
+
+        try:
+            yield
+
+        finally:
+            await active_publisher.stop()
+
+    app = FastAPI(
+        title="Telecom CDR Ingestion API",
+        version="0.1.0",
+        lifespan=lifespan,
+    )
+
+    app.include_router(
+        router,
+        prefix="/api/v1",
+    )
+
+    @app.get("/health")
+    async def health() -> dict[str, str]:
+        return {
+            "status": "ok",
+        }
+
+    return app
+
+
+app = create_app()

src/core/api/publisher.py

@@ -0,0 +1,25 @@
+from collections.abc import Mapping
+from typing import Any, Protocol
+
+
+class IngestionQueueFull(Exception):
+    pass
+
+
+class KafkaPublishError(Exception):
+    pass
+
+
+class Publisher(Protocol):
+    async def start(self) -> None:
+        ...
+
+    async def stop(self) -> None:
+        ...
+
+    async def publish(
+        self,
+        payload: Mapping[str, Any],
+        key: str,
+    ) -> None:
+        ...

src/core/api/routes.py

@@ -0,0 +1,83 @@
+import json
+
+from fastapi import APIRouter, Request, status
+from fastapi.responses import JSONResponse
+
+from core.api.publisher import IngestionQueueFull, KafkaPublishError
+from core.domain.errors import CDRProcessingError
+from core.domain.validation import parse_raw_cdr
+
+
+router = APIRouter()
+
+
+@router.post(
+    "/cdr/raw",
+    status_code=status.HTTP_202_ACCEPTED,
+)
+async def ingest_raw_cdr(request: Request) -> JSONResponse:
+    try:
+        payload = await request.json()
+
+    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+        return JSONResponse(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            content={
+                "error": "invalid_json",
+                "detail": str(exc),
+            },
+        )
+
+    if not isinstance(payload, dict):
+        return JSONResponse(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            content={
+                "error": "invalid_payload",
+                "detail": "JSON payload must be an object.",
+            },
+        )
+
+    try:
+        raw_cdr = parse_raw_cdr(payload)
+
+    except CDRProcessingError as exc:
+        return JSONResponse(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            content={
+                "error": exc.reason.value,
+                "detail": exc.detail,
+                "failed_field": exc.failed_field,
+            },
+        )
+
+    try:
+        await request.app.state.publisher.publish(
+            payload,
+            key=raw_cdr.raw_id,
+        )
+
+    except IngestionQueueFull:
+        return JSONResponse(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            content={
+                "error": "backpressure",
+                "detail": "Ingestion queue is full.",
+            },
+        )
+
+    except KafkaPublishError as exc:
+        return JSONResponse(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            content={
+                "error": "kafka_unavailable",
+                "detail": str(exc),
+            },
+        )
+
+    return JSONResponse(
+        status_code=status.HTTP_202_ACCEPTED,
+        content={
+            "status": "accepted",
+            "raw_id": raw_cdr.raw_id,
+        },
+    )

src/core/config.py

@@ -0,0 +1,36 @@
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    kafka_bootstrap_servers: str = "localhost:19092"
+    kafka_raw_topic: str = "cdr.raw"
+
+    api_queue_size: int = 10_000
+
+    kafka_delivery_timeout_seconds: float = 4.0
+    kafka_message_timeout_ms: int = 3_000
+
+    kafka_linger_ms: int = 5
+    kafka_batch_size: int = 131_072
+    kafka_producer_max_messages: int = 100_000
+
+    kafka_dlq_topic: str = "cdr.raw.dlq"
+    kafka_consumer_group: str = "cdr-processing-workers"
+    kafka_auto_offset_reset: str = "earliest"
+
+    kafka_consumer_poll_timeout_seconds: float = 1.0
+    kafka_dlq_delivery_timeout_seconds: float = 5.0
+
+    worker_instance_id: str = "worker-1"
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()

src/core/domain/errors.py

@@ -0,0 +1,23 @@
+from enum import StrEnum
+
+
+class FailureReason(StrEnum):
+    MISSING_REQUIRED_FIELD = "missing_required_field"
+    INVALID_RAT_VALUE = "invalid_rat_value"
+    INVALID_EPOCH = "invalid_epoch"
+    MALFORMED_JSON = "malformed_json"
+    NORMALIZATION_ERROR = "normalization_error"
+
+
+class CDRProcessingError(Exception):
+    def __init__(
+        self,
+        reason: FailureReason,
+        detail: str,
+        failed_field: str | None = None,
+    ) -> None:
+        super().__init__(detail)
+
+        self.reason = reason
+        self.detail = detail
+        self.failed_field = failed_field

src/core/domain/models.py

@@ -0,0 +1,43 @@
+from dataclasses import dataclass
+from datetime import datetime
+from ipaddress import IPv4Address
+from uuid import UUID
+
+
+@dataclass(frozen=True, slots=True)
+class RawCDR:
+    raw_id: str
+    device_id: str
+    caller_num: str
+    svc: str
+    rat: int
+    epoch_start: int
+    epoch_end: int
+    up_bytes: int
+    down_bytes: int
+    lac: str
+    ci: str
+    src_ip: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class NormalizedCDR:
+    cdr_id: UUID
+    raw_id: str
+
+    msisdn: str
+    imsi: str
+
+    service_type: str
+    radio_access_type: str
+    cell_id: str
+
+    bytes_uploaded: int
+    bytes_downloaded: int
+    total_bytes: int
+    duration_seconds: int
+
+    session_start_time: datetime
+    session_end_time: datetime
+
+    source_ip: IPv4Address

src/core/domain/normalization.py

@@ -0,0 +1,121 @@
+import uuid
+from datetime import datetime, timezone
+from ipaddress import IPv4Address, AddressValueError
+
+import phonenumbers
+from phonenumbers import PhoneNumberFormat
+
+from core.domain.errors import CDRProcessingError, FailureReason
+from core.domain.models import NormalizedCDR, RawCDR
+
+
+RAT_MAPPING = {
+    1: "3G",
+    6: "LTE_4G",
+    9: "5G_NR",
+}
+
+def normalize_rat(rat: int) -> str:
+    try:
+        return RAT_MAPPING[rat]
+    except KeyError as exc:
+        raise CDRProcessingError(
+            reason=FailureReason.INVALID_RAT_VALUE,
+            detail=f"Unsupported RAT value: {rat}.",
+            failed_field="rat",
+        ) from exc
+
+def _normalize_international_prefix(number: str) -> str:
+    number = number.strip()
+
+    if number.startswith("+"):
+        return number
+
+    if number.startswith("00"):
+        return f"+{number[2:]}"
+
+    if number.startswith("011"):
+        return f"+{number[3:]}"
+
+    return number
+
+def normalize_msisdn(number: str) -> str:
+    prepared_number = _normalize_international_prefix(number)
+
+    try:
+        parsed = phonenumbers.parse(
+            prepared_number,
+            None,
+        )
+    except phonenumbers.NumberParseException as exc:
+        raise CDRProcessingError(
+            reason=FailureReason.NORMALIZATION_ERROR,
+            detail=f"Unable to parse caller_num: {number}.",
+            failed_field="caller_num",
+        ) from exc
+
+    if not phonenumbers.is_possible_number(parsed):
+        raise CDRProcessingError(
+            reason=FailureReason.NORMALIZATION_ERROR,
+            detail=f"caller_num is not a possible phone number: {number}.",
+            failed_field="caller_num",
+        )
+
+    return phonenumbers.format_number(
+        parsed,
+        PhoneNumberFormat.E164,
+    )
+
+def generate_cdr_id(raw_id: str) -> uuid.UUID:
+    return uuid.uuid5(
+        uuid.NAMESPACE_DNS,
+        raw_id,
+    )
+
+def epoch_to_datetime(epoch: int) -> datetime:
+    try:
+        return datetime.fromtimestamp(
+            epoch,
+            tz=timezone.utc,
+        )
+    except (OverflowError, OSError, ValueError) as exc:
+        raise CDRProcessingError(
+            reason=FailureReason.INVALID_EPOCH,
+            detail=f"Invalid Unix epoch value: {epoch}.",
+        ) from exc
+
+def normalize_source_ip(source_ip: str | None) -> IPv4Address:
+    if source_ip is None:
+        return IPv4Address("0.0.0.0")
+
+    try:
+        return IPv4Address(source_ip)
+    except AddressValueError as exc:
+        raise CDRProcessingError(
+            reason=FailureReason.NORMALIZATION_ERROR,
+            detail=f"Invalid IPv4 address: {source_ip}.",
+            failed_field="src_ip",
+        ) from exc
+
+def normalize_cdr(raw: RawCDR) -> NormalizedCDR:
+    return NormalizedCDR(
+        cdr_id=generate_cdr_id(raw.raw_id),
+        raw_id=raw.raw_id,
+
+        msisdn=normalize_msisdn(raw.caller_num),
+        imsi=raw.device_id,
+
+        service_type=raw.svc,
+        radio_access_type=normalize_rat(raw.rat),
+        cell_id=f"{raw.lac}_{raw.ci}",
+
+        bytes_uploaded=raw.up_bytes,
+        bytes_downloaded=raw.down_bytes,
+        total_bytes=raw.up_bytes + raw.down_bytes,
+        duration_seconds=raw.epoch_end - raw.epoch_start,
+
+        session_start_time=epoch_to_datetime(raw.epoch_start),
+        session_end_time=epoch_to_datetime(raw.epoch_end),
+
+        source_ip=normalize_source_ip(raw.src_ip),
+    )

src/core/domain/validation.py

@@ -0,0 +1,139 @@
+from collections.abc import Mapping
+from typing import Any
+
+from core.domain.errors import CDRProcessingError, FailureReason
+from core.domain.models import RawCDR
+
+
+REQUIRED_FIELDS = (
+    "raw_id",
+    "device_id",
+    "caller_num",
+    "svc",
+    "rat",
+    "epoch_start",
+    "epoch_end",
+    "up_bytes",
+    "down_bytes",
+    "lac",
+    "ci",
+)
+
+
+SUPPORTED_RAT_VALUES = {
+    1,
+    6,
+    9,
+}
+
+def _validate_required_fields(payload: Mapping[str, Any]) -> None:
+    for field in REQUIRED_FIELDS:
+        if field not in payload or payload[field] is None:
+            raise CDRProcessingError(
+                reason=FailureReason.MISSING_REQUIRED_FIELD,
+                detail=f"Required field '{field}' is missing or null.",
+                failed_field=field,
+            )
+
+def _validate_type(
+    payload: Mapping[str, Any],
+    field: str,
+    expected_type: type,
+) -> None:
+    value = payload[field]
+
+    # bool is technically a subclass of int in Python.
+    # We do not want True/False accepted as integer CDR fields.
+    if expected_type is int and isinstance(value, bool):
+        valid = False
+    else:
+        valid = isinstance(value, expected_type)
+
+    if not valid:
+        raise CDRProcessingError(
+            reason=FailureReason.NORMALIZATION_ERROR,
+            detail=(
+                f"Field '{field}' must be of type "
+                f"{expected_type.__name__}."
+            ),
+            failed_field=field,
+        )
+
+def parse_raw_cdr(payload: Mapping[str, Any]) -> RawCDR:
+    _validate_required_fields(payload)
+
+    string_fields = (
+        "raw_id",
+        "device_id",
+        "caller_num",
+        "svc",
+        "lac",
+        "ci",
+    )
+
+    integer_fields = (
+        "rat",
+        "epoch_start",
+        "epoch_end",
+        "up_bytes",
+        "down_bytes",
+    )
+
+    for field in string_fields:
+        _validate_type(payload, field, str)
+
+    for field in integer_fields:
+        _validate_type(payload, field, int)
+
+    src_ip = payload.get("src_ip")
+
+    if src_ip is not None:
+        _validate_type(payload, "src_ip", str)
+
+    raw_cdr = RawCDR(
+        raw_id=payload["raw_id"],
+        device_id=payload["device_id"],
+        caller_num=payload["caller_num"],
+        svc=payload["svc"],
+        rat=payload["rat"],
+        epoch_start=payload["epoch_start"],
+        epoch_end=payload["epoch_end"],
+        up_bytes=payload["up_bytes"],
+        down_bytes=payload["down_bytes"],
+        lac=payload["lac"],
+        ci=payload["ci"],
+        src_ip=src_ip,
+    )
+
+    validate_raw_cdr(raw_cdr)
+
+    return raw_cdr
+
+def validate_raw_cdr(raw: RawCDR) -> None:
+    if raw.epoch_start > raw.epoch_end:
+        raise CDRProcessingError(
+            reason=FailureReason.INVALID_EPOCH,
+            detail="epoch_start must be less than or equal to epoch_end.",
+            failed_field="epoch_start",
+        )
+
+    if raw.up_bytes < 0:
+        raise CDRProcessingError(
+            reason=FailureReason.NORMALIZATION_ERROR,
+            detail="up_bytes must be greater than or equal to zero.",
+            failed_field="up_bytes",
+        )
+
+    if raw.down_bytes < 0:
+        raise CDRProcessingError(
+            reason=FailureReason.NORMALIZATION_ERROR,
+            detail="down_bytes must be greater than or equal to zero.",
+            failed_field="down_bytes",
+        )
+
+    if raw.rat not in SUPPORTED_RAT_VALUES:
+        raise CDRProcessingError(
+            reason=FailureReason.INVALID_RAT_VALUE,
+            detail=f"Unsupported RAT value: {raw.rat}.",
+            failed_field="rat",
+        )

src/core/worker/consumer.py

@@ -0,0 +1,215 @@
+import logging
+from typing import Any
+
+from confluent_kafka import Consumer, KafkaException, Message
+
+from core.config import Settings
+from core.domain.errors import (
+    CDRProcessingError,
+    FailureReason,
+)
+from core.worker.dlq import (
+    DLQPublishError,
+    DLQPublisher,
+)
+from core.worker.processor import (
+    decode_payload,
+    process_payload,
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+class ProcessingWorker:
+    def __init__(
+        self,
+        settings: Settings,
+    ) -> None:
+        self._settings = settings
+
+        self._consumer = Consumer(
+            {
+                "bootstrap.servers":
+                    settings.kafka_bootstrap_servers,
+
+                "group.id":
+                    settings.kafka_consumer_group,
+
+                "client.id":
+                    settings.worker_instance_id,
+
+                "auto.offset.reset":
+                    settings.kafka_auto_offset_reset,
+
+                "enable.auto.commit": False,
+            }
+        )
+
+        self._dlq = DLQPublisher(settings)
+
+        self._running = False
+
+    def run(self) -> None:
+        self._running = True
+
+        self._consumer.subscribe(
+            [self._settings.kafka_raw_topic]
+        )
+
+        logger.info(
+            "Worker started. topic=%s group=%s instance=%s",
+            self._settings.kafka_raw_topic,
+            self._settings.kafka_consumer_group,
+            self._settings.worker_instance_id,
+        )
+
+        try:
+            while self._running:
+                message = self._consumer.poll(
+                    self._settings.kafka_consumer_poll_timeout_seconds
+                )
+
+                if message is None:
+                    continue
+
+                if message.error():
+                    logger.error(
+                        "Kafka consumer error: %s",
+                        message.error(),
+                    )
+                    continue
+
+                self._handle_message(message)
+
+        finally:
+            self._shutdown()
+
+    def _handle_message(
+        self,
+        message: Message,
+    ) -> None:
+        value = message.value() or b""
+
+        original_payload: Any = {
+            "raw": value.decode(
+                "utf-8",
+                errors="replace",
+            )
+        }
+
+        try:
+            payload = decode_payload(value)
+
+            original_payload = payload
+
+            normalized = process_payload(payload)
+
+        except CDRProcessingError as exc:
+            self._route_to_dlq(
+                message=message,
+                original_payload=original_payload,
+                error=exc,
+            )
+
+            return
+
+        except Exception as exc:
+            logger.exception(
+                "Unexpected CDR processing error."
+            )
+
+            error = CDRProcessingError(
+                reason=FailureReason.NORMALIZATION_ERROR,
+                detail=f"Unexpected processing error: {exc}",
+            )
+
+            self._route_to_dlq(
+                message=message,
+                original_payload=original_payload,
+                error=error,
+            )
+
+            return
+
+        logger.info(
+            (
+                "Normalized CDR. "
+                "raw_id=%s cdr_id=%s "
+                "service_type=%s rat=%s"
+            ),
+            normalized.raw_id,
+            normalized.cdr_id,
+            normalized.service_type,
+            normalized.radio_access_type,
+        )
+
+        # Temporary until ClickHouse batching is implemented.
+        self._commit(message)
+
+    def _route_to_dlq(
+        self,
+        message: Message,
+        original_payload: Any,
+        error: CDRProcessingError,
+    ) -> None:
+        logger.warning(
+            (
+                "Routing CDR to DLQ. "
+                "reason=%s field=%s "
+                "partition=%s offset=%s"
+            ),
+            error.reason.value,
+            error.failed_field,
+            message.partition(),
+            message.offset(),
+        )
+
+        try:
+            self._dlq.publish(
+                original_payload=original_payload,
+                error=error,
+            )
+
+        except DLQPublishError:
+            logger.exception(
+                "DLQ publishing failed. "
+                "Original Kafka offset will not be committed."
+            )
+
+            raise
+
+        self._commit(message)
+
+    def _commit(
+        self,
+        message: Message,
+    ) -> None:
+        committed = self._consumer.commit(
+            message=message,
+            asynchronous=False,
+        )
+
+        for partition in committed or []:
+            if partition.error is not None:
+                raise KafkaException(
+                    partition.error
+                )
+
+        logger.debug(
+            "Committed Kafka offset. partition=%s offset=%s",
+            message.partition(),
+            message.offset(),
+        )
+
+    def stop(self) -> None:
+        self._running = False
+
+    def _shutdown(self) -> None:
+        logger.info("Worker shutting down.")
+
+        try:
+            self._dlq.close()
+
+        finally:
+            self._consumer.close()

src/core/worker/dlq.py

@@ -0,0 +1,124 @@
+import json
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+from confluent_kafka import KafkaException, Producer
+
+from core.config import Settings
+from core.domain.errors import CDRProcessingError
+
+
+class DLQPublishError(Exception):
+    pass
+
+
+def build_dlq_envelope(
+    original_payload: Any,
+    error: CDRProcessingError,
+    worker_instance: str,
+) -> dict[str, Any]:
+    return {
+        "original_payload": original_payload,
+        "failure_reason": error.reason.value,
+        "failed_field": error.failed_field,
+        "failed_at": (
+            datetime.now(timezone.utc)
+            .isoformat()
+            .replace("+00:00", "Z")
+        ),
+        "worker_instance": worker_instance,
+    }
+
+
+class DLQPublisher:
+    def __init__(
+        self,
+        settings: Settings,
+    ) -> None:
+        self._settings = settings
+
+        self._producer = Producer(
+            {
+                "bootstrap.servers":
+                    settings.kafka_bootstrap_servers,
+
+                "client.id":
+                    f"{settings.worker_instance_id}-dlq",
+
+                "acks": "all",
+                "enable.idempotence": True,
+
+                "compression.type": "lz4",
+
+                "message.timeout.ms":
+                    settings.kafka_message_timeout_ms,
+            }
+        )
+
+    def publish(
+        self,
+        original_payload: Any,
+        error: CDRProcessingError,
+    ) -> None:
+        envelope = build_dlq_envelope(
+            original_payload=original_payload,
+            error=error,
+            worker_instance=self._settings.worker_instance_id,
+        )
+
+        value = json.dumps(
+            envelope,
+            separators=(",", ":"),
+            ensure_ascii=False,
+        ).encode("utf-8")
+
+        delivered = False
+        delivery_error = None
+
+        def on_delivery(err, message) -> None:
+            nonlocal delivered
+            nonlocal delivery_error
+
+            delivery_error = err
+            delivered = True
+
+        try:
+            self._producer.produce(
+                topic=self._settings.kafka_dlq_topic,
+                value=value,
+                on_delivery=on_delivery,
+            )
+
+        except (BufferError, KafkaException) as exc:
+            raise DLQPublishError(
+                f"Unable to queue DLQ message: {exc}"
+            ) from exc
+
+        deadline = (
+            time.monotonic()
+            + self._settings.kafka_dlq_delivery_timeout_seconds
+        )
+
+        while not delivered:
+            self._producer.poll(0.1)
+
+            if time.monotonic() >= deadline:
+                raise DLQPublishError(
+                    "DLQ Kafka delivery timed out."
+                )
+
+        if delivery_error is not None:
+            raise DLQPublishError(
+                f"DLQ delivery failed: {delivery_error}"
+            )
+
+    def close(self) -> None:
+        remaining = self._producer.flush(
+            self._settings.kafka_dlq_delivery_timeout_seconds
+        )
+
+        if remaining > 0:
+            raise DLQPublishError(
+                f"{remaining} DLQ message(s) were not delivered."
+            )

src/core/worker/main.py

@@ -0,0 +1,51 @@
+import logging
+import signal
+
+from core.config import get_settings
+from core.worker.consumer import ProcessingWorker
+
+
+logging.basicConfig(
+    level=logging.INFO,
+    format=(
+        "%(asctime)s "
+        "%(levelname)s "
+        "%(name)s "
+        "%(message)s"
+    ),
+)
+
+
+def main() -> None:
+    settings = get_settings()
+
+    worker = ProcessingWorker(
+        settings=settings,
+    )
+
+    def handle_shutdown(
+        signum,
+        frame,
+    ) -> None:
+        logging.getLogger(__name__).info(
+            "Shutdown signal received: %s",
+            signum,
+        )
+
+        worker.stop()
+
+    signal.signal(
+        signal.SIGINT,
+        handle_shutdown,
+    )
+
+    signal.signal(
+        signal.SIGTERM,
+        handle_shutdown,
+    )
+
+    worker.run()
+
+
+if __name__ == "__main__":
+    main()

src/core/worker/processor.py

@@ -0,0 +1,38 @@
+import json
+from typing import Any
+
+from core.domain.errors import (
+    CDRProcessingError,
+    FailureReason,
+)
+from core.domain.models import NormalizedCDR
+from core.domain.normalization import normalize_cdr
+from core.domain.validation import parse_raw_cdr
+
+
+def decode_payload(value: bytes) -> dict[str, Any]:
+    try:
+        text = value.decode("utf-8")
+        payload = json.loads(text)
+
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise CDRProcessingError(
+            reason=FailureReason.MALFORMED_JSON,
+            detail=f"Unable to decode Kafka message as JSON: {exc}",
+        ) from exc
+
+    if not isinstance(payload, dict):
+        raise CDRProcessingError(
+            reason=FailureReason.NORMALIZATION_ERROR,
+            detail="Kafka JSON payload must be an object.",
+        )
+
+    return payload
+
+
+def process_payload(
+    payload: dict[str, Any],
+) -> NormalizedCDR:
+    raw_cdr = parse_raw_cdr(payload)
+
+    return normalize_cdr(raw_cdr)

tests/integration/test_ingestion_api.py

@@ -0,0 +1,228 @@
+from collections.abc import Mapping
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from core.api.main import create_app
+from core.api.publisher import (
+    IngestionQueueFull,
+    KafkaPublishError,
+)
+
+
+class FakePublisher:
+    def __init__(self) -> None:
+        self.messages: list[
+            tuple[str, Mapping[str, Any]]
+        ] = []
+
+        self.failure: Exception | None = None
+
+    async def start(self) -> None:
+        pass
+
+    async def stop(self) -> None:
+        pass
+
+    async def publish(
+        self,
+        payload: Mapping[str, Any],
+        key: str,
+    ) -> None:
+        if self.failure is not None:
+            raise self.failure
+
+        self.messages.append(
+            (key, payload)
+        )
+
+
+@pytest.fixture
+def publisher() -> FakePublisher:
+    return FakePublisher()
+
+
+@pytest.fixture
+def client(
+    publisher: FakePublisher,
+):
+    app = create_app(
+        publisher=publisher,
+    )
+
+    with TestClient(app) as test_client:
+        yield test_client
+
+
+def valid_payload() -> dict:
+    return {
+        "raw_id": "raw-test-001",
+        "src_ip": "10.240.12.1",
+        "device_id": "310410998877665",
+        "caller_num": "0019875551234",
+        "svc": "4G_DATA",
+        "up_bytes": 1048576,
+        "down_bytes": 4194304,
+        "rat": 6,
+        "lac": "4021",
+        "ci": "8812",
+        "epoch_start": 1789900000,
+        "epoch_end": 1789900300,
+    }
+
+
+def test_health_endpoint(client: TestClient) -> None:
+    response = client.get("/health")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "status": "ok",
+    }
+
+
+def test_valid_cdr_returns_202(
+    client: TestClient,
+    publisher: FakePublisher,
+) -> None:
+    payload = valid_payload()
+
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=payload,
+    )
+
+    assert response.status_code == 202
+
+    assert response.json() == {
+        "status": "accepted",
+        "raw_id": "raw-test-001",
+    }
+
+    assert len(publisher.messages) == 1
+
+    key, published_payload = publisher.messages[0]
+
+    assert key == "raw-test-001"
+    assert published_payload == payload
+
+
+def test_missing_required_field_returns_400(client: TestClient) -> None:
+    payload = valid_payload()
+    del payload["caller_num"]
+
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=payload,
+    )
+
+    assert response.status_code == 400
+
+    body = response.json()
+
+    assert body["error"] == "missing_required_field"
+    assert body["failed_field"] == "caller_num"
+
+
+def test_invalid_rat_returns_400(client: TestClient) -> None:
+    payload = valid_payload()
+    payload["rat"] = 99
+
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=payload,
+    )
+
+    assert response.status_code == 400
+
+    body = response.json()
+
+    assert body["error"] == "invalid_rat_value"
+    assert body["failed_field"] == "rat"
+
+
+def test_invalid_epoch_returns_400(client: TestClient) -> None:
+    payload = valid_payload()
+
+    payload["epoch_start"] = 200
+    payload["epoch_end"] = 100
+
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=payload,
+    )
+
+    assert response.status_code == 400
+
+    body = response.json()
+
+    assert body["error"] == "invalid_epoch"
+
+
+def test_malformed_json_returns_400(client: TestClient) -> None:
+    response = client.post(
+        "/api/v1/cdr/raw",
+        content='{"raw_id": invalid}',
+        headers={
+            "Content-Type": "application/json",
+        },
+    )
+
+    assert response.status_code == 400
+
+    body = response.json()
+
+    assert body["error"] == "invalid_json"
+
+
+def test_non_object_json_returns_400(client: TestClient) -> None:
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=[
+            "not",
+            "a",
+            "cdr",
+        ],
+    )
+
+    assert response.status_code == 400
+
+    body = response.json()
+
+    assert body["error"] == "invalid_payload"
+
+
+def test_queue_full_returns_429(
+    client: TestClient,
+    publisher: FakePublisher,
+) -> None:
+    publisher.failure = IngestionQueueFull(
+        "Ingestion queue is full."
+    )
+
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=valid_payload(),
+    )
+
+    assert response.status_code == 429
+
+    assert response.json()["error"] == "backpressure"
+
+
+def test_kafka_failure_returns_503(
+    client: TestClient,
+    publisher: FakePublisher,
+) -> None:
+    publisher.failure = KafkaPublishError(
+        "Kafka unavailable."
+    )
+
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=valid_payload(),
+    )
+
+    assert response.status_code == 503
+
+    assert response.json()["error"] == "kafka_unavailable"

tests/unit/test_dlq.py

@@ -0,0 +1,36 @@
+from core.domain.errors import (
+    CDRProcessingError,
+    FailureReason,
+)
+from core.worker.dlq import build_dlq_envelope
+
+
+def test_build_dlq_envelope() -> None:
+    original = {
+        "raw_id": "raw-invalid-001",
+        "rat": 99,
+    }
+
+    error = CDRProcessingError(
+        reason=FailureReason.INVALID_RAT_VALUE,
+        detail="Unsupported RAT value: 99.",
+        failed_field="rat",
+    )
+
+    envelope = build_dlq_envelope(
+        original_payload=original,
+        error=error,
+        worker_instance="worker-test",
+    )
+
+    assert envelope["original_payload"] == original
+
+    assert (
+        envelope["failure_reason"]
+        == "invalid_rat_value"
+    )
+
+    assert envelope["failed_field"] == "rat"
+    assert envelope["worker_instance"] == "worker-test"
+
+    assert envelope["failed_at"].endswith("Z")

tests/unit/test_normalization.py

@@ -0,0 +1,77 @@
+from core.domain.normalization import generate_cdr_id, normalize_cdr, normalize_rat
+from core.domain.validation import parse_raw_cdr
+
+
+def test_normalize_valid_cdr() -> None:
+    payload = {
+        "raw_id": "raw-us-east-9988221",
+        "src_ip": "10.240.12.1",
+        "device_id": "310410998877665",
+        "caller_num": "0019875551234",
+        "svc": "4G_DATA",
+        "up_bytes": 1048576,
+        "down_bytes": 4194304,
+        "rat": 6,
+        "lac": "4021",
+        "ci": "8812",
+        "epoch_start": 1726070000,
+        "epoch_end": 1726070300,
+    }
+
+    raw = parse_raw_cdr(payload)
+
+    normalized = normalize_cdr(raw)
+
+    assert normalized.raw_id == "raw-us-east-9988221"
+    assert normalized.imsi == "310410998877665"
+    assert normalized.msisdn == "+19875551234"
+
+    assert normalized.service_type == "4G_DATA"
+    assert normalized.radio_access_type == "LTE_4G"
+
+    assert normalized.cell_id == "4021_8812"
+
+    assert normalized.bytes_uploaded == 1048576
+    assert normalized.bytes_downloaded == 4194304
+    assert normalized.total_bytes == 5242880
+
+    assert normalized.duration_seconds == 300
+
+    assert str(normalized.source_ip) == "10.240.12.1"
+
+def test_cdr_id_is_deterministic() -> None:
+    first = generate_cdr_id("raw-123")
+    second = generate_cdr_id("raw-123")
+
+    assert first == second
+
+def test_different_raw_ids_generate_different_cdr_ids() -> None:
+    first = generate_cdr_id("raw-123")
+    second = generate_cdr_id("raw-456")
+
+    assert first != second
+
+def test_rat_mapping() -> None:
+    assert normalize_rat(1) == "3G"
+    assert normalize_rat(6) == "LTE_4G"
+    assert normalize_rat(9) == "5G_NR"
+
+def test_missing_source_ip_defaults_to_zero_address() -> None:
+    payload = {
+        "raw_id": "raw-001",
+        "device_id": "310410998877665",
+        "caller_num": "0019875551234",
+        "svc": "4G_DATA",
+        "up_bytes": 100,
+        "down_bytes": 200,
+        "rat": 6,
+        "lac": "4021",
+        "ci": "8812",
+        "epoch_start": 1726070000,
+        "epoch_end": 1726070300,
+    }
+
+    raw = parse_raw_cdr(payload)
+    normalized = normalize_cdr(raw)
+
+    assert str(normalized.source_ip) == "0.0.0.0"

tests/unit/test_validation.py

@@ -0,0 +1,71 @@
+import pytest
+
+from core.domain.errors import CDRProcessingError, FailureReason
+from core.domain.validation import parse_raw_cdr
+
+
+def valid_payload() -> dict:
+    return {
+        "raw_id": "raw-001",
+        "src_ip": "10.240.12.1",
+        "device_id": "310410998877665",
+        "caller_num": "0019875551234",
+        "svc": "4G_DATA",
+        "up_bytes": 1048576,
+        "down_bytes": 4194304,
+        "rat": 6,
+        "lac": "4021",
+        "ci": "8812",
+        "epoch_start": 1726070000,
+        "epoch_end": 1726070300,
+    }
+
+def test_missing_required_field() -> None:
+    payload = valid_payload()
+    del payload["caller_num"]
+
+    with pytest.raises(CDRProcessingError) as exc_info:
+        parse_raw_cdr(payload)
+
+    error = exc_info.value
+
+    assert error.reason == FailureReason.MISSING_REQUIRED_FIELD
+    assert error.failed_field == "caller_num"
+
+def test_invalid_rat() -> None:
+    payload = valid_payload()
+    payload["rat"] = 99
+
+    with pytest.raises(CDRProcessingError) as exc_info:
+        parse_raw_cdr(payload)
+
+    error = exc_info.value
+
+    assert error.reason == FailureReason.INVALID_RAT_VALUE
+    assert error.failed_field == "rat"
+
+def test_epoch_start_after_end_is_invalid() -> None:
+    payload = valid_payload()
+
+    payload["epoch_start"] = 200
+    payload["epoch_end"] = 100
+
+    with pytest.raises(CDRProcessingError) as exc_info:
+        parse_raw_cdr(payload)
+
+    error = exc_info.value
+
+    assert error.reason == FailureReason.INVALID_EPOCH
+
+def test_negative_uploaded_bytes_is_invalid() -> None:
+    payload = valid_payload()
+
+    payload["up_bytes"] = -1
+
+    with pytest.raises(CDRProcessingError) as exc_info:
+        parse_raw_cdr(payload)
+
+    error = exc_info.value
+
+    assert error.reason == FailureReason.NORMALIZATION_ERROR
+    assert error.failed_field == "up_bytes"

tests/unit/test_worker_processor.py

@@ -0,0 +1,78 @@
+import json
+
+import pytest
+
+from core.domain.errors import (
+    CDRProcessingError,
+    FailureReason,
+)
+from core.worker.processor import (
+    decode_payload,
+    process_payload,
+)
+
+
+def valid_payload() -> dict:
+    return {
+        "raw_id": "raw-worker-001",
+        "src_ip": "10.240.12.1",
+        "device_id": "310410998877665",
+        "caller_num": "0019875551234",
+        "svc": "4G_DATA",
+        "up_bytes": 100,
+        "down_bytes": 200,
+        "rat": 6,
+        "lac": "4021",
+        "ci": "8812",
+        "epoch_start": 1789900000,
+        "epoch_end": 1789900300,
+    }
+
+
+def test_decode_valid_json_object() -> None:
+    value = json.dumps(
+        valid_payload()
+    ).encode("utf-8")
+
+    payload = decode_payload(value)
+
+    assert payload == valid_payload()
+
+
+def test_malformed_json() -> None:
+    with pytest.raises(
+        CDRProcessingError
+    ) as exc_info:
+        decode_payload(
+            b'{"raw_id": invalid}'
+        )
+
+    assert (
+        exc_info.value.reason
+        == FailureReason.MALFORMED_JSON
+    )
+
+
+def test_json_array_is_rejected() -> None:
+    with pytest.raises(
+        CDRProcessingError
+    ) as exc_info:
+        decode_payload(
+            b'["not", "a", "cdr"]'
+        )
+
+    assert (
+        exc_info.value.reason
+        == FailureReason.NORMALIZATION_ERROR
+    )
+
+
+def test_worker_normalizes_valid_payload() -> None:
+    normalized = process_payload(
+        valid_payload()
+    )
+
+    assert normalized.raw_id == "raw-worker-001"
+    assert normalized.radio_access_type == "LTE_4G"
+    assert normalized.total_bytes == 300
+    assert normalized.duration_seconds == 300