feat(worker): complete Phase 3 - processing worker with normalization, micro-batching, and at-least-once offset management

app/worker/__init__.py

@@ -0,0 +1 @@
+# app/worker package

app/worker/clickhouse.py

@@ -0,0 +1,182 @@
+import time
+from abc import ABC, abstractmethod
+from typing import Any, List, Optional
+
+from app.common.config import config
+from app.common.logging import setup_logger
+from app.worker.normalizer import NormalizedRecord
+
+logger = setup_logger("processing-worker")
+
+COLUMN_NAMES = [
+    "cdr_id",
+    "raw_id",
+    "msisdn",
+    "imsi",
+    "service_type",
+    "radio_access_type",
+    "cell_id",
+    "bytes_uploaded",
+    "bytes_downloaded",
+    "total_bytes",
+    "duration_seconds",
+    "session_start_time",
+    "session_end_time",
+    "source_ip",
+]
+
+
+class BaseClickHouseClient(ABC):
+    @abstractmethod
+    def insert_batch(self, records: List[NormalizedRecord]) -> bool:
+        pass
+
+    @abstractmethod
+    def is_healthy(self) -> bool:
+        pass
+
+    @abstractmethod
+    def close(self) -> None:
+        pass
+
+
+class ClickHouseService(BaseClickHouseClient):
+    """
+    ClickHouse client using clickhouse-connect with exponential retry backoff.
+    """
+
+    def __init__(
+        self,
+        host: Optional[str] = None,
+        port: Optional[int] = None,
+        user: Optional[str] = None,
+        password: Optional[str] = None,
+        database: Optional[str] = None,
+        table: Optional[str] = None,
+        max_attempts: Optional[int] = None,
+        initial_delay_ms: Optional[int] = None,
+        backoff_factor: Optional[float] = None,
+    ):
+        try:
+            import clickhouse_connect
+            self._cc = clickhouse_connect
+        except ImportError:
+            raise ImportError(
+                "clickhouse-connect is not installed. Run 'pip install clickhouse-connect'."
+            )
+
+        self.host = host or config.clickhouse.host
+        self.port = port or config.clickhouse.port
+        self.user = user or config.clickhouse.user
+        self.password = password or config.clickhouse.password
+        self.database = database or config.clickhouse.database
+        self.table = table or config.clickhouse.table
+
+        self.max_attempts = max_attempts or config.retry.max_attempts
+        self.initial_delay_ms = initial_delay_ms or config.retry.initial_delay_ms
+        self.backoff_factor = backoff_factor or config.retry.backoff_factor
+
+        self._client = None
+        self._connect()
+
+    def _connect(self) -> None:
+        try:
+            self._client = self._cc.get_client(
+                host=self.host,
+                port=self.port,
+                username=self.user,
+                password=self.password,
+                database=self.database,
+                connect_timeout=5,
+                send_receive_timeout=10,
+            )
+        except Exception as e:
+            logger.error(
+                f"ClickHouse connection failed: {e}",
+                extra={"event": "clickhouse_connection_failed"},
+            )
+            self._client = None
+
+    def is_healthy(self) -> bool:
+        if self._client is None:
+            self._connect()
+        if self._client is None:
+            return False
+        try:
+            return self._client.ping()
+        except Exception:
+            return False
+
+    def insert_batch(self, records: List[NormalizedRecord]) -> bool:
+        """
+        Inserts a micro-batch of NormalizedRecord into ClickHouse with exponential backoff retry.
+        Returns True if successful, raises exception if all retries fail.
+        """
+        if not records:
+            return True
+
+        rows = [record.to_row() for record in records]
+        last_exception = None
+
+        for attempt in range(1, self.max_attempts + 1):
+            try:
+                if self._client is None:
+                    self._connect()
+                if self._client is None:
+                    raise ConnectionError("ClickHouse client is not connected")
+
+                self._client.insert(
+                    table=self.table,
+                    data=rows,
+                    column_names=COLUMN_NAMES,
+                )
+                return True
+            except Exception as e:
+                last_exception = e
+                delay = (self.initial_delay_ms / 1000.0) * (
+                    self.backoff_factor ** (attempt - 1)
+                )
+                logger.warning(
+                    f"ClickHouse insert attempt {attempt}/{self.max_attempts} failed: {e}. "
+                    f"Retrying in {delay:.2f}s...",
+                    extra={"event": "clickhouse_insert_retry"},
+                )
+                time.sleep(delay)
+
+        logger.error(
+            f"All {self.max_attempts} ClickHouse insert attempts failed: {last_exception}",
+            extra={"event": "clickhouse_batch_failed"},
+        )
+        raise last_exception or RuntimeError("ClickHouse insert failed")
+
+    def close(self) -> None:
+        if self._client:
+            self._client.close()
+            self._client = None
+
+
+class MockClickHouseClient(BaseClickHouseClient):
+    """
+    In-memory ClickHouse mock client for offline tests.
+    """
+
+    def __init__(self, fail_attempts: int = 0):
+        self.rows: List[Any] = []
+        self.fail_attempts = fail_attempts
+        self.attempt_count = 0
+        self._healthy = True
+
+    def is_healthy(self) -> bool:
+        return self._healthy
+
+    def insert_batch(self, records: List[NormalizedRecord]) -> bool:
+        self.attempt_count += 1
+        if self.attempt_count <= self.fail_attempts:
+            raise TimeoutError("ClickHouse insert timed out (simulated)")
+
+        for record in records:
+            self.rows.append(record.to_row())
+        return True
+
+    def close(self) -> None:
+        pass

app/worker/consumer.py

@@ -0,0 +1,266 @@
+import signal
+import time
+from datetime import datetime, timezone
+from typing import Any, List, Optional, Tuple
+
+from app.api.producer import ConfluentKafkaProducer, MessageProducer
+from app.common.config import config
+from app.common.logging import setup_logger
+from app.worker.clickhouse import BaseClickHouseClient, ClickHouseService
+from app.worker.dlq import route_to_dlq
+from app.worker.internal_server import WorkerInternalServer
+from app.worker.metrics import worker_metrics
+from app.worker.normalizer import NormalizedRecord, normalize_cdr_payload
+
+logger = setup_logger("processing-worker")
+
+
+class ProcessingWorker:
+    """
+    Consumes raw CDRs from Kafka, normalizes schema, micro-batches into ClickHouse,
+    routes malformed payloads to DLQ, and manages offsets with at-least-once semantics.
+    """
+
+    def __init__(
+        self,
+        consumer: Optional[Any] = None,
+        dlq_producer: Optional[MessageProducer] = None,
+        clickhouse: Optional[BaseClickHouseClient] = None,
+        batch_size: Optional[int] = None,
+        flush_interval: Optional[float] = None,
+        internal_server: Optional[WorkerInternalServer] = None,
+    ):
+        self.batch_size = batch_size or config.batch.size
+        self.flush_interval = flush_interval or config.batch.flush_interval_seconds
+
+        # Kafka Consumer
+        self._consumer = consumer
+        # Kafka DLQ Producer
+        self.dlq_producer = dlq_producer or ConfluentKafkaProducer()
+        self.dlq_topic = config.kafka.dlq_topic
+
+        # ClickHouse Client
+        self.clickhouse = clickhouse or ClickHouseService()
+
+        # Internal HTTP Server for metrics & /stats
+        self.internal_server = internal_server or WorkerInternalServer()
+
+        # Micro-batch buffer: list of (NormalizedRecord, message)
+        self.buffer: List[Tuple[NormalizedRecord, Any]] = []
+        self.last_flush_time = time.time()
+        self.running = False
+
+    def _init_consumer(self) -> None:
+        if self._consumer is None:
+            try:
+                import confluent_kafka
+
+                conf = {
+                    "bootstrap.servers": config.kafka.bootstrap_servers,
+                    "group.id": config.kafka.consumer_group,
+                    "auto.offset.reset": config.kafka.auto_offset_reset,
+                    "enable.auto.commit": False,  # Manual offset commit post ClickHouse ACK
+                }
+                self._consumer = confluent_kafka.Consumer(conf)
+            except ImportError:
+                raise ImportError(
+                    "confluent-kafka is required. Run 'pip install confluent-kafka'."
+                )
+
+    def _handle_signal(self, signum: int, frame: Any) -> None:
+        sig_name = "SIGTERM" if signum == signal.SIGTERM else "SIGINT"
+        logger.info(
+            f"Received {sig_name}, initiating graceful shutdown...",
+            extra={"event": "shutdown_signal_received"},
+        )
+        self.running = False
+
+    def flush_buffer(self) -> None:
+        """
+        Flushes buffered normalized records to ClickHouse and commits Kafka offsets.
+        If ClickHouse write fails all retries, routes batch to DLQ.
+        """
+        if not self.buffer:
+            self.last_flush_time = time.time()
+            return
+
+        batch_records = [item[0] for item in self.buffer]
+        batch_messages = [item[1] for item in self.buffer]
+        batch_count = len(batch_records)
+
+        start_time = time.time()
+        try:
+            self.clickhouse.insert_batch(batch_records)
+            flush_duration = time.time() - start_time
+
+            # Commit offsets only after ClickHouse ACK
+            if self._consumer and hasattr(self._consumer, "commit"):
+                try:
+                    self._consumer.commit(asynchronous=False)
+                except Exception as e:
+                    logger.error(
+                        f"Kafka offset commit failed after ClickHouse ACK: {e}",
+                        extra={"event": "kafka_commit_failed"},
+                    )
+
+            # Record metrics
+            now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+            worker_metrics.set_last_flush_at(now_iso)
+            worker_metrics.batch_flush_duration.observe(flush_duration)
+            worker_metrics.batch_size_at_flush.observe(batch_count)
+
+            for rec in batch_records:
+                worker_metrics.record_processed(
+                    rec.service_type, rec.radio_access_type, 1
+                )
+
+            logger.info(
+                f"Successfully flushed {batch_count} records to ClickHouse in {flush_duration:.3f}s",
+                extra={"event": "clickhouse_batch_flushed"},
+            )
+
+        except Exception as e:
+            logger.error(
+                f"Batch insert to ClickHouse failed after retries: {e}. Routing batch to DLQ.",
+                extra={"event": "ch_batch_fatal_error"},
+            )
+            # Route all records in batch to DLQ to prevent pipeline deadlock
+            for rec, msg in self.buffer:
+                route_to_dlq(
+                    producer=self.dlq_producer,
+                    topic=self.dlq_topic,
+                    original_payload=vars(rec),
+                    failure_reason="ch_insert_failed",
+                    failed_field=None,
+                    raw_id=rec.raw_id,
+                )
+
+            # Commit offsets to prevent infinite replay stall
+            if self._consumer and hasattr(self._consumer, "commit"):
+                try:
+                    self._consumer.commit(asynchronous=False)
+                except Exception as ce:
+                    logger.error(f"Commit after DLQ fallback failed: {ce}")
+
+        finally:
+            self.buffer.clear()
+            worker_metrics.set_in_flight_buffer(0)
+            self.last_flush_time = time.time()
+
+    def process_single_message(self, message: Any) -> None:
+        """
+        Parses, validates, and either buffers or routes message to DLQ.
+        """
+        value_bytes = message.value() if hasattr(message, "value") else message
+        if value_bytes is None:
+            return
+
+        record, failure_reason, failed_field, raw_id = normalize_cdr_payload(
+            value_bytes
+        )
+
+        if failure_reason:
+            payload_str = (
+                value_bytes.decode("utf-8", errors="replace")
+                if isinstance(value_bytes, bytes)
+                else str(value_bytes)
+            )
+            route_to_dlq(
+                producer=self.dlq_producer,
+                topic=self.dlq_topic,
+                original_payload=payload_str,
+                failure_reason=failure_reason,
+                failed_field=failed_field,
+                raw_id=raw_id,
+            )
+            # Commit DLQ record offset so it is not reprocessed
+            if self._consumer and hasattr(self._consumer, "commit"):
+                try:
+                    self._consumer.commit(message=message, asynchronous=False)
+                except Exception:
+                    pass
+            return
+
+        if record:
+            self.buffer.append((record, message))
+            worker_metrics.set_in_flight_buffer(len(self.buffer))
+
+            # Flush if batch size threshold reached
+            if len(self.buffer) >= self.batch_size:
+                self.flush_buffer()
+
+    def start(self) -> None:
+        """
+        Main worker execution loop with signal handling and micro-batching timer.
+        """
+        self._init_consumer()
+        self.internal_server.start()
+
+        # Register signals for graceful shutdown
+        signal.signal(signal.SIGTERM, self._handle_signal)
+        signal.signal(signal.SIGINT, self._handle_signal)
+
+        self._consumer.subscribe([config.kafka.raw_topic])
+        self.running = True
+        logger.info(
+            f"Processing Worker started. Subscribed to '{config.kafka.raw_topic}'. "
+            f"Batch size: {self.batch_size}, Interval: {self.flush_interval}s",
+            extra={"event": "worker_started"},
+        )
+
+        try:
+            while self.running:
+                # Poll with short 100ms timeout
+                msg = self._consumer.poll(timeout=0.1)
+
+                if msg is None:
+                    # Check flush timer on idle / interval
+                    if (
+                        self.buffer
+                        and (time.time() - self.last_flush_time) >= self.flush_interval
+                    ):
+                        self.flush_buffer()
+                    continue
+
+                if msg.error():
+                    logger.error(
+                        f"Consumer error: {msg.error()}",
+                        extra={"event": "kafka_consumer_error"},
+                    )
+                    continue
+
+                self.process_single_message(msg)
+
+                # Check flush interval during message stream
+                if (
+                    self.buffer
+                    and (time.time() - self.last_flush_time) >= self.flush_interval
+                ):
+                    self.flush_buffer()
+
+        finally:
+            self.stop()
+
+    def stop(self) -> None:
+        """
+        Graceful shutdown: flush any in-memory records and commit offsets.
+        """
+        logger.info(
+            f"Stopping worker: flushing {len(self.buffer)} remaining in-memory records...",
+            extra={"event": "graceful_shutdown_flushing"},
+        )
+        if self.buffer:
+            self.flush_buffer()
+
+        if self._consumer and hasattr(self._consumer, "close"):
+            self._consumer.close()
+
+        self.dlq_producer.close()
+        self.clickhouse.close()
+        self.internal_server.stop()
+        logger.info("Processing Worker stopped cleanly", extra={"event": "worker_stopped"})
+
+
+if __name__ == "__main__":
+    worker = ProcessingWorker()
+    worker.start()

app/worker/dlq.py

@@ -0,0 +1,66 @@
+import json
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+
+from app.api.producer import MessageProducer
+from app.common.config import config
+from app.common.logging import setup_logger
+from app.worker.metrics import worker_metrics
+
+logger = setup_logger("processing-worker")
+
+ALLOWED_FAILURE_REASONS = {
+    "missing_required_field",
+    "invalid_rat_value",
+    "invalid_epoch",
+    "malformed_json",
+    "normalization_error",
+    "ch_insert_failed",
+}
+
+
+def build_dlq_envelope(
+    original_payload: Any,
+    failure_reason: str,
+    failed_field: Optional[str] = None,
+    worker_instance: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Constructs a standardized DLQ envelope according to specification.
+    """
+    return {
+        "original_payload": original_payload,
+        "failure_reason": failure_reason,
+        "failed_field": failed_field,
+        "failed_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "worker_instance": worker_instance or config.worker.instance_id,
+    }
+
+
+def route_to_dlq(
+    producer: MessageProducer,
+    topic: str,
+    original_payload: Any,
+    failure_reason: str,
+    failed_field: Optional[str] = None,
+    raw_id: Optional[str] = None,
+) -> bool:
+    """
+    Wraps unprocessable record in a DLQ envelope, updates DLQ metric,
+    and publishes to the DLQ topic.
+    """
+    envelope = build_dlq_envelope(
+        original_payload=original_payload,
+        failure_reason=failure_reason,
+        failed_field=failed_field,
+    )
+    envelope_bytes = json.dumps(envelope, default=str).encode("utf-8")
+
+    success = producer.produce(topic=topic, key=raw_id, value=envelope_bytes)
+    worker_metrics.record_dlq(failure_reason, 1)
+
+    logger.warning(
+        f"Record routed to DLQ: reason={failure_reason}, field={failed_field}",
+        extra={"raw_id": raw_id, "event": "routed_to_dlq"},
+    )
+    return success

app/worker/internal_server.py

@@ -0,0 +1,95 @@
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Optional
+
+from app.common.config import config
+from app.common.logging import setup_logger
+from app.worker.metrics import worker_metrics
+
+logger = setup_logger("processing-worker")
+
+
+class WorkerHTTPRequestHandler(BaseHTTPRequestHandler):
+    """
+    Lightweight HTTP request handler for worker internal metrics and stats.
+    """
+
+    def log_message(self, format: str, *args) -> None:
+        # Suppress default noisy stderr logging
+        pass
+
+    def do_GET(self) -> None:
+        if self.path == "/internal/stats":
+            stats = worker_metrics.get_internal_stats()
+            body = json.dumps(stats, indent=2).encode("utf-8")
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+            return
+
+        if self.path == "/metrics":
+            metrics_text = worker_metrics.get_prometheus_metrics()
+            body = metrics_text.encode("utf-8")
+            self.send_response(200)
+            self.send_header(
+                "Content-Type", "text/plain; version=0.0.4; charset=utf-8"
+            )
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+            return
+
+        if self.path in ("/health", "/healthz", "/live"):
+            body = b'{"status": "ok"}'
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+            return
+
+        self.send_response(404)
+        self.send_header("Content-Type", "application/json")
+        body = b'{"error": "not_found"}'
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+
+class WorkerInternalServer:
+    """
+    Background HTTP server serving /internal/stats and /metrics on worker port (8001).
+    """
+
+    def __init__(self, host: Optional[str] = None, port: Optional[int] = None):
+        self.host = host or config.worker.stats_host
+        self.port = port or config.worker.stats_port
+        self._server: Optional[ThreadingHTTPServer] = None
+        self._thread: Optional[threading.Thread] = None
+
+    def start(self) -> None:
+        self._server = ThreadingHTTPServer(
+            (self.host, self.port), WorkerHTTPRequestHandler
+        )
+        self._thread = threading.Thread(
+            target=self._server.serve_forever, daemon=True, name="WorkerInternalHTTP"
+        )
+        self._thread.start()
+        logger.info(
+            f"Worker internal HTTP server started on http://{self.host}:{self.port}",
+            extra={"event": "worker_http_started"},
+        )
+
+    def stop(self) -> None:
+        if self._server:
+            self._server.shutdown()
+            self._server.server_close()
+            if self._thread:
+                self._thread.join(timeout=2.0)
+            logger.info(
+                "Worker internal HTTP server stopped cleanly",
+                extra={"event": "worker_http_stopped"},
+            )

app/worker/metrics.py

@@ -0,0 +1,165 @@
+import threading
+from collections import defaultdict
+from typing import Any, Dict, List, Optional
+
+
+class SimpleHistogram:
+    """
+    Lightweight, thread-safe Prometheus-compatible histogram.
+    """
+
+    def __init__(self, name: str, help_text: str, buckets: List[float]):
+        self.name = name
+        self.help_text = help_text
+        self.buckets = sorted(buckets)
+        self._counts: Dict[float, int] = {b: 0 for b in self.buckets}
+        self._count = 0
+        self._sum = 0.0
+        self._lock = threading.Lock()
+
+    def observe(self, value: float) -> None:
+        with self._lock:
+            self._count += 1
+            self._sum += value
+            for b in self.buckets:
+                if value <= b:
+                    self._counts[b] += 1
+
+    def to_prometheus(self) -> str:
+        lines = [
+            f"# HELP {self.name} {self.help_text}",
+            f"# TYPE {self.name} histogram",
+        ]
+        with self._lock:
+            for b in self.buckets:
+                lines.append(f'{self.name}_bucket{{le="{b}"}} {self._counts[b]}')
+            lines.append(f'{self.name}_bucket{{le="+Inf"}} {self._count}')
+            lines.append(f"{self.name}_count {self._count}")
+            lines.append(f"{self.name}_sum {self._sum:.4f}")
+        return "\n".join(lines) + "\n"
+
+
+class WorkerMetrics:
+    """
+    Thread-safe operational metrics collector for the Processing Worker.
+    Provides data for /metrics (Prometheus) and /internal/stats (Ingestion API aggregator).
+    """
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._processed_total: Dict[tuple, int] = defaultdict(int)
+        self._dlq_total: Dict[str, int] = defaultdict(int)
+        self._in_flight_buffer_size: int = 0
+        self._last_flush_at: Optional[str] = None
+        self._consumer_lag: Dict[str, int] = defaultdict(int)
+
+        # Histograms
+        self.kafka_to_ch_latency = SimpleHistogram(
+            name="cdr_kafka_to_clickhouse_latency_seconds",
+            help_text="End-to-end latency from Kafka produce to ClickHouse write",
+            buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
+        )
+        self.batch_flush_duration = SimpleHistogram(
+            name="cdr_batch_flush_duration_seconds",
+            help_text="ClickHouse batch flush duration in seconds",
+            buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0],
+        )
+        self.batch_size_at_flush = SimpleHistogram(
+            name="cdr_batch_size_at_flush",
+            help_text="Batch sizes at flush time",
+            buckets=[10, 50, 100, 250, 500, 1000],
+        )
+
+    def record_processed(
+        self, service_type: str, radio_access_type: str, count: int = 1
+    ) -> None:
+        with self._lock:
+            self._processed_total[(service_type, radio_access_type)] += count
+
+    def record_dlq(self, reason: str, count: int = 1) -> None:
+        with self._lock:
+            self._dlq_total[reason] += count
+
+    def set_in_flight_buffer(self, size: int) -> None:
+        with self._lock:
+            self._in_flight_buffer_size = size
+
+    def set_last_flush_at(self, timestamp_iso: str) -> None:
+        with self._lock:
+            self._last_flush_at = timestamp_iso
+
+    def set_consumer_lag(self, partition: int, lag: int) -> None:
+        with self._lock:
+            self._consumer_lag[str(partition)] = lag
+
+    def get_internal_stats(self) -> Dict[str, Any]:
+        """
+        Returns structured dictionary for GET /internal/stats.
+        """
+        with self._lock:
+            total_processed = sum(self._processed_total.values())
+            total_dlq = sum(self._dlq_total.values())
+            in_flight = self._in_flight_buffer_size
+            last_flush = self._last_flush_at
+
+            # Breakdowns
+            by_service: Dict[str, int] = defaultdict(int)
+            by_rat: Dict[str, int] = defaultdict(int)
+            for (svc, rat), cnt in self._processed_total.items():
+                by_service[svc] += cnt
+                by_rat[rat] += cnt
+
+            dlq_reasons = dict(self._dlq_total)
+
+        return {
+            "pipeline": {
+                "total_processed": total_processed,
+                "total_dlq": total_dlq,
+                "in_flight_buffer": in_flight,
+                "last_flush_at": last_flush,
+            },
+            "breakdown": {
+                "by_service_type": dict(by_service),
+                "by_radio_access_type": dict(by_rat),
+                "dlq_by_reason": dlq_reasons,
+            },
+        }
+
+    def get_prometheus_metrics(self) -> str:
+        """
+        Formats all worker metrics into standard Prometheus exposition format.
+        """
+        lines = []
+
+        # 1. cdr_processed_total
+        lines.append("# HELP cdr_processed_total Total CDRs successfully written to ClickHouse")
+        lines.append("# TYPE cdr_processed_total counter")
+        with self._lock:
+            for (svc, rat), count in self._processed_total.items():
+                lines.append(f'cdr_processed_total{{service_type="{svc}",radio_access_type="{rat}"}} {count}')
+
+            # 2. cdr_dlq_total
+            lines.append("# HELP cdr_dlq_total Total CDRs routed to DLQ by reason")
+            lines.append("# TYPE cdr_dlq_total counter")
+            for reason, count in self._dlq_total.items():
+                lines.append(f'cdr_dlq_total{{reason="{reason}"}} {count}')
+
+            # 3. Gauges
+            lines.append("# HELP cdr_in_flight_buffer_size Current in-memory buffer size")
+            lines.append("# TYPE cdr_in_flight_buffer_size gauge")
+            lines.append(f"cdr_in_flight_buffer_size {self._in_flight_buffer_size}")
+
+            lines.append("# HELP cdr_kafka_consumer_lag Kafka consumer lag per partition")
+            lines.append("# TYPE cdr_kafka_consumer_lag gauge")
+            for part, lag in self._consumer_lag.items():
+                lines.append(f'cdr_kafka_consumer_lag{{partition="{part}"}} {lag}')
+
+        # 4. Histograms
+        lines.append(self.kafka_to_ch_latency.to_prometheus().strip())
+        lines.append(self.batch_flush_duration.to_prometheus().strip())
+        lines.append(self.batch_size_at_flush.to_prometheus().strip())
+
+        return "\n".join(lines) + "\n"
+
+
+worker_metrics = WorkerMetrics()

app/worker/normalizer.py

@@ -0,0 +1,200 @@
+import json
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional, Tuple
+
+import phonenumbers
+
+RAT_MAP = {
+    1: "3G",
+    6: "LTE_4G",
+    9: "5G_NR",
+}
+
+REQUIRED_FIELDS = [
+    "raw_id",
+    "device_id",
+    "caller_num",
+    "svc",
+    "rat",
+    "epoch_start",
+    "epoch_end",
+    "up_bytes",
+    "down_bytes",
+    "lac",
+    "ci",
+]
+
+
+@dataclass(frozen=True)
+class NormalizedRecord:
+    cdr_id: str
+    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: str
+    session_end_time: str
+    source_ip: str
+
+    def to_row(self) -> Tuple[Any, ...]:
+        """Returns ordered tuple matching ClickHouse cdr_records schema columns."""
+        return (
+            uuid.UUID(self.cdr_id),
+            self.raw_id,
+            self.msisdn,
+            self.imsi,
+            self.service_type,
+            self.radio_access_type,
+            self.cell_id,
+            self.bytes_uploaded,
+            self.bytes_downloaded,
+            self.total_bytes,
+            self.duration_seconds,
+            datetime.strptime(self.session_start_time, "%Y-%m-%d %H:%M:%S"),
+            datetime.strptime(self.session_end_time, "%Y-%m-%d %H:%M:%S"),
+            self.source_ip,
+        )
+
+
+def normalize_phone_number(raw_num: str) -> Optional[str]:
+    """
+    Normalizes telephone number into E.164 format:
+    - Replaces leading '00' or '011' with '+'
+    - Preserves existing '+'
+    - Validates with phonenumbers library
+    """
+    if not isinstance(raw_num, str):
+        return None
+
+    cleaned = raw_num.strip()
+    if cleaned.startswith("00"):
+        cleaned = "+" + cleaned[2:]
+    elif cleaned.startswith("011"):
+        cleaned = "+" + cleaned[3:]
+    elif not cleaned.startswith("+"):
+        cleaned = "+" + cleaned
+
+    try:
+        parsed = phonenumbers.parse(cleaned, None)
+        if not (
+            phonenumbers.is_valid_number(parsed)
+            or phonenumbers.is_possible_number(parsed)
+        ):
+            return None
+        return phonenumbers.format_number(
+            parsed, phonenumbers.PhoneNumberFormat.E164
+        )
+    except Exception:
+        return None
+
+
+def normalize_cdr_payload(
+    raw_payload_bytes: bytes,
+) -> Tuple[Optional[NormalizedRecord], Optional[str], Optional[str], Optional[str]]:
+    """
+    Validates and normalizes raw CDR payload.
+    Returns:
+      (record, failure_reason, failed_field, raw_id)
+    If record is None, failure_reason describes why it should be routed to DLQ.
+    """
+    # 1. Parse JSON
+    try:
+        if isinstance(raw_payload_bytes, bytes):
+            payload_str = raw_payload_bytes.decode("utf-8")
+        else:
+            payload_str = str(raw_payload_bytes)
+        payload: Dict[str, Any] = json.loads(payload_str)
+    except Exception:
+        return None, "malformed_json", None, None
+
+    if not isinstance(payload, dict):
+        return None, "malformed_json", None, None
+
+    raw_id = payload.get("raw_id")
+    raw_id_str = str(raw_id) if raw_id is not None else None
+
+    # 2. Check required fields
+    for field in REQUIRED_FIELDS:
+        if field not in payload or payload[field] is None:
+            return None, "missing_required_field", field, raw_id_str
+
+    # 3. Check timestamps (epoch_start <= epoch_end)
+    try:
+        epoch_start = int(payload["epoch_start"])
+        epoch_end = int(payload["epoch_end"])
+        if epoch_start > epoch_end:
+            return None, "invalid_epoch", "epoch_start", raw_id_str
+    except (ValueError, TypeError):
+        return None, "invalid_epoch", "epoch_start", raw_id_str
+
+    # 4. Check bytes (must be >= 0)
+    try:
+        up_bytes = int(payload["up_bytes"])
+        down_bytes = int(payload["down_bytes"])
+        if up_bytes < 0:
+            return None, "normalization_error", "up_bytes", raw_id_str
+        if down_bytes < 0:
+            return None, "normalization_error", "down_bytes", raw_id_str
+    except (ValueError, TypeError):
+        return None, "normalization_error", "bytes", raw_id_str
+
+    # 5. Radio Access Type mapping
+    rat_raw = payload.get("rat")
+    try:
+        rat_int = int(rat_raw)
+        if rat_int not in RAT_MAP:
+            return None, "invalid_rat_value", "rat", raw_id_str
+        radio_access_type = RAT_MAP[rat_int]
+    except (ValueError, TypeError):
+        return None, "invalid_rat_value", "rat", raw_id_str
+
+    # 6. E.164 phone normalization
+    caller_num = str(payload["caller_num"])
+    msisdn = normalize_phone_number(caller_num)
+    if not msisdn:
+        return None, "normalization_error", "caller_num", raw_id_str
+
+    # 7. Field mappings and derivations
+    imsi = str(payload["device_id"])
+    service_type = str(payload["svc"])
+    cell_id = f"{payload['lac']}_{payload['ci']}"
+    total_bytes = up_bytes + down_bytes
+    duration_seconds = epoch_end - epoch_start
+
+    start_dt = datetime.fromtimestamp(epoch_start, tz=timezone.utc).strftime(
+        "%Y-%m-%d %H:%M:%S"
+    )
+    end_dt = datetime.fromtimestamp(epoch_end, tz=timezone.utc).strftime(
+        "%Y-%m-%d %H:%M:%S"
+    )
+
+    src_ip = str(payload.get("src_ip") or "0.0.0.0")
+
+    # Deterministic UUIDv5 cdr_id from raw_id
+    cdr_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, raw_id_str))
+
+    record = NormalizedRecord(
+        cdr_id=cdr_id,
+        raw_id=raw_id_str,
+        msisdn=msisdn,
+        imsi=imsi,
+        service_type=service_type,
+        radio_access_type=radio_access_type,
+        cell_id=cell_id,
+        bytes_uploaded=up_bytes,
+        bytes_downloaded=down_bytes,
+        total_bytes=total_bytes,
+        duration_seconds=duration_seconds,
+        session_start_time=start_dt,
+        session_end_time=end_dt,
+        source_ip=src_ip,
+    )
+    return record, None, None, raw_id_str