feat: add cdr processing worker and dlq routing

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()