feat(api): complete Phase 2 - framework-less asyncio Ingestion API with backpressure and metrics

.gitattributes

@@ -21,3 +21,5 @@
 *.gif binary
 *.ico binary
 *.tar.gz binary
+*.zip binary
+*.pdf binary

app/api/__init__.py

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

app/api/metrics.py

@@ -0,0 +1,37 @@
+import threading
+
+
+class APIMetrics:
+    """
+    In-memory metrics tracker for the Ingestion API.
+    Thread-safe and async-safe counter for Prometheus scraping.
+    """
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._received_total: int = 0
+
+    def increment_received(self, count: int = 1) -> None:
+        with self._lock:
+            self._received_total += count
+
+    @property
+    def total_received(self) -> int:
+        with self._lock:
+            return self._received_total
+
+    def get_prometheus_metrics(self) -> str:
+        """
+        Formats metrics into standard Prometheus text format.
+        """
+        with self._lock:
+            val = self._received_total
+
+        return (
+            "# HELP cdr_api_received_total Total CDRs received by the API\n"
+            "# TYPE cdr_api_received_total counter\n"
+            f"cdr_api_received_total {val}\n"
+        )
+
+
+api_metrics = APIMetrics()

app/api/producer.py

@@ -0,0 +1,185 @@
+import asyncio
+import logging
+from abc import ABC, abstractmethod
+from typing import Any, Callable, Dict, Optional
+
+from app.common.config import config
+from app.common.logging import setup_logger
+
+logger = setup_logger("ingestion-api")
+
+
+class MessageProducer(ABC):
+    @abstractmethod
+    def produce(self, topic: str, key: Optional[str], value: bytes) -> bool:
+        """Produce a message synchronously or enqueue in broker buffer."""
+        pass
+
+    @abstractmethod
+    def is_healthy(self) -> bool:
+        """Check if producer is healthy and connected to broker."""
+        pass
+
+    @abstractmethod
+    def flush(self, timeout: float = 5.0) -> int:
+        """Flush any pending messages."""
+        pass
+
+    @abstractmethod
+    def close(self) -> None:
+        """Close producer connection."""
+        pass
+
+
+class ConfluentKafkaProducer(MessageProducer):
+    """
+    Production Kafka producer implementation wrapping confluent-kafka.
+    Uses 'raw_id' as partition key and LZ4 compression.
+    """
+
+    def __init__(self, bootstrap_servers: Optional[str] = None):
+        try:
+            import confluent_kafka
+            self._ck = confluent_kafka
+        except ImportError:
+            raise ImportError(
+                "confluent-kafka library is not installed. Run 'pip install confluent-kafka'."
+            )
+
+        self.bootstrap_servers = bootstrap_servers or config.kafka.bootstrap_servers
+        self._producer_config = {
+            "bootstrap.servers": self.bootstrap_servers,
+            "compression.type": "lz4",
+            "queue.buffering.max.messages": 100000,
+            "queue.buffering.max.ms": 5,
+            "acks": "all",
+            "retries": 3,
+        }
+        self._producer = self._ck.Producer(self._producer_config)
+        self._healthy = True
+
+    def _delivery_report(self, err: Any, msg: Any) -> None:
+        if err is not None:
+            self._healthy = False
+            raw_id = msg.key().decode("utf-8") if msg and msg.key() else None
+            logger.error(
+                f"Kafka delivery failed: {err}",
+                extra={"raw_id": raw_id, "event": "kafka_produce_failed"},
+            )
+        else:
+            self._healthy = True
+
+    def produce(self, topic: str, key: Optional[str], value: bytes) -> bool:
+        try:
+            key_bytes = key.encode("utf-8") if key else None
+            self._producer.produce(
+                topic=topic,
+                key=key_bytes,
+                value=value,
+                on_delivery=self._delivery_report,
+            )
+            # Serve delivery callback queue non-blockingly
+            self._producer.poll(0)
+            return True
+        except BufferError:
+            logger.warning(
+                "Kafka local producer buffer full",
+                extra={"raw_id": key, "event": "kafka_buffer_full"},
+            )
+            self._producer.poll(0.1)
+            return False
+        except Exception as e:
+            self._healthy = False
+            logger.error(
+                f"Kafka produce error: {e}",
+                extra={"raw_id": key, "event": "kafka_error"},
+            )
+            return False
+
+    def is_healthy(self) -> bool:
+        return self._healthy
+
+    def flush(self, timeout: float = 5.0) -> int:
+        return self._producer.flush(timeout)
+
+    def close(self) -> None:
+        self.flush()
+
+
+class MockMessageProducer(MessageProducer):
+    """
+    In-memory mock producer for offline testing and verification.
+    """
+
+    def __init__(self):
+        self.messages = []
+        self._healthy = True
+
+    def produce(self, topic: str, key: Optional[str], value: bytes) -> bool:
+        if not self._healthy:
+            return False
+        self.messages.append({"topic": topic, "key": key, "value": value})
+        return True
+
+    def is_healthy(self) -> bool:
+        return self._healthy
+
+    def set_healthy(self, healthy: bool) -> None:
+        self._healthy = healthy
+
+    def flush(self, timeout: float = 5.0) -> int:
+        return 0
+
+    def close(self) -> None:
+        pass
+
+
+class AsyncProducerWorker:
+    """
+    Background worker consuming from the bounded async queue and writing to the Kafka producer.
+    """
+
+    def __init__(self, queue: asyncio.Queue, producer: MessageProducer, topic: str):
+        self.queue = queue
+        self.producer = producer
+        self.topic = topic
+        self._running = False
+        self._task: Optional[asyncio.Task] = None
+
+    async def start(self) -> None:
+        self._running = True
+        self._task = asyncio.create_task(self._run_loop())
+
+    async def _run_loop(self) -> None:
+        while self._running:
+            try:
+                item = await self.queue.get()
+                if item is None:  # Sentinel to stop
+                    self.queue.task_done()
+                    break
+
+                key, value = item
+                success = self.producer.produce(self.topic, key, value)
+                if not success:
+                    logger.error(
+                        "Failed to publish message from async queue to Kafka",
+                        extra={"raw_id": key, "event": "queue_dispatch_failed"},
+                    )
+                self.queue.task_done()
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error(
+                    f"Unexpected error in async producer worker: {e}",
+                    extra={"event": "worker_unexpected_error"},
+                )
+
+    async def stop(self) -> None:
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self.producer.flush()

app/api/server.py

@@ -0,0 +1,287 @@
+import asyncio
+import json
+import logging
+from typing import Optional, Tuple
+
+from app.api.metrics import api_metrics
+from app.api.producer import (
+    AsyncProducerWorker,
+    ConfluentKafkaProducer,
+    MessageProducer,
+    MockMessageProducer,
+)
+from app.api.stats import get_aggregated_pipeline_stats
+from app.common.config import config
+from app.common.logging import setup_logger
+
+logger = setup_logger("ingestion-api")
+
+
+class IngestionAPIServer:
+    """
+    Framework-less HTTP/1.1 Ingestion API server built on standard library asyncio.
+    Enforces bounded async queue backpressure and dispatches to Kafka.
+    """
+
+    def __init__(
+        self,
+        host: Optional[str] = None,
+        port: Optional[int] = None,
+        queue_max_size: Optional[int] = None,
+        producer: Optional[MessageProducer] = None,
+        worker_stats_url: Optional[str] = None,
+    ):
+        self.host = host or config.api.host
+        self.port = port or config.api.port
+        self.queue_max_size = queue_max_size or config.api.queue_max_size
+        self.worker_stats_url = worker_stats_url or config.api.worker_stats_url
+
+        self.queue: asyncio.Queue = asyncio.Queue(maxsize=self.queue_max_size)
+        self.producer = producer or ConfluentKafkaProducer()
+        self.dispatcher = AsyncProducerWorker(
+            queue=self.queue,
+            producer=self.producer,
+            topic=config.kafka.raw_topic,
+        )
+
+        self._server: Optional[asyncio.Server] = None
+        self._running = False
+
+    async def start(self) -> None:
+        """Start the async producer dispatcher and HTTP server."""
+        await self.dispatcher.start()
+        self._server = await asyncio.start_server(
+            self._handle_client, self.host, self.port
+        )
+        self._running = True
+        logger.info(
+            f"Ingestion API listening on http://{self.host}:{self.port} (queue capacity: {self.queue_max_size})",
+            extra={"event": "server_started"},
+        )
+
+    async def stop(self) -> None:
+        """Stop the HTTP server and flush dispatcher queue."""
+        self._running = False
+        if self._server:
+            self._server.close()
+            await self._server.wait_closed()
+        await self.dispatcher.stop()
+        self.producer.close()
+        logger.info("Ingestion API stopped cleanly", extra={"event": "server_stopped"})
+
+    async def _handle_client(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        try:
+            # Read request line
+            request_line = await reader.readline()
+            if not request_line:
+                writer.close()
+                await writer.wait_closed()
+                return
+
+            line_str = request_line.decode("utf-8", errors="replace").strip()
+            parts = line_str.split(" ")
+            if len(parts) < 2:
+                await self._send_response(
+                    writer, 400, "application/json", b'{"error": "bad_request"}'
+                )
+                return
+
+            method = parts[0].upper()
+            path = parts[1]
+
+            # Read headers
+            headers = {}
+            while True:
+                header_line = await reader.readline()
+                if not header_line or header_line == b"\r\n" or header_line == b"\n":
+                    break
+                header_str = header_line.decode("utf-8", errors="replace").strip()
+                if ":" in header_str:
+                    key, val = header_str.split(":", 1)
+                    headers[key.strip().lower()] = val.strip()
+
+            # Read body based on Content-Length
+            content_length = int(headers.get("content-length", 0))
+            body = b""
+            if content_length > 0:
+                body = await reader.readexactly(content_length)
+
+            # Route dispatch
+            await self._dispatch_route(writer, method, path, headers, body)
+
+        except asyncio.IncompleteReadError:
+            pass
+        except Exception as e:
+            logger.error(
+                f"Error handling HTTP request: {e}",
+                extra={"event": "request_handler_error"},
+            )
+            try:
+                err_body = json.dumps(
+                    {"error": "internal_error", "detail": str(e)}
+                ).encode("utf-8")
+                await self._send_response(writer, 500, "application/json", err_body)
+            except Exception:
+                pass
+        finally:
+            try:
+                writer.close()
+                await writer.wait_closed()
+            except Exception:
+                pass
+
+    async def _dispatch_route(
+        self,
+        writer: asyncio.StreamWriter,
+        method: str,
+        path: str,
+        headers: dict,
+        body: bytes,
+    ) -> None:
+        # 1. Health check
+        if method == "GET" and path in ("/health", "/healthz", "/live"):
+            res = json.dumps(
+                {"status": "ok", "queue_size": self.queue.qsize()}
+            ).encode("utf-8")
+            await self._send_response(writer, 200, "application/json", res)
+            return
+
+        # 2. Prometheus metrics
+        if method == "GET" and path == "/metrics":
+            res = api_metrics.get_prometheus_metrics().encode("utf-8")
+            await self._send_response(
+                writer, 200, "text/plain; version=0.0.4; charset=utf-8", res
+            )
+            return
+
+        # 3. Public Stats summary
+        if method == "GET" and path == "/stats":
+            stats = await get_aggregated_pipeline_stats(self.worker_stats_url)
+            res = json.dumps(stats, indent=2).encode("utf-8")
+            await self._send_response(writer, 200, "application/json", res)
+            return
+
+        # 4. Raw CDR Ingestion endpoint
+        if method == "POST" and path == "/api/v1/cdr/raw":
+            await self._handle_raw_cdr_ingestion(writer, body)
+            return
+
+        # 5. Not found
+        err = json.dumps(
+            {"error": "not_found", "detail": f"Cannot {method} {path}"}
+        ).encode("utf-8")
+        await self._send_response(writer, 404, "application/json", err)
+
+    async def _handle_raw_cdr_ingestion(
+        self, writer: asyncio.StreamWriter, body: bytes
+    ) -> None:
+        # Check broker reachability
+        if not self.producer.is_healthy():
+            err = json.dumps(
+                {
+                    "error": "broker_unavailable",
+                    "detail": "Kafka broker is unreachable",
+                }
+            ).encode("utf-8")
+            await self._send_response(writer, 503, "application/json", err)
+            return
+
+        # Check queue capacity (Backpressure)
+        if self.queue.full():
+            err = json.dumps(
+                {
+                    "error": "queue_full",
+                    "detail": "Internal async buffer is saturated (backpressure)",
+                }
+            ).encode("utf-8")
+            await self._send_response(writer, 429, "application/json", err)
+            return
+
+        # Validate JSON format
+        try:
+            body_text = body.decode("utf-8")
+            payload = json.loads(body_text)
+        except json.JSONDecodeError as e:
+            err = json.dumps(
+                {
+                    "error": "invalid_json",
+                    "detail": f"Unexpected token at position {e.pos}: {e.msg}",
+                }
+            ).encode("utf-8")
+            await self._send_response(writer, 400, "application/json", err)
+            return
+
+        if not isinstance(payload, dict):
+            err = json.dumps(
+                {"error": "invalid_payload", "detail": "Expected a JSON object"}
+            ).encode("utf-8")
+            await self._send_response(writer, 400, "application/json", err)
+            return
+
+        raw_id = payload.get("raw_id")
+
+        # Put in bounded queue without blocking
+        try:
+            self.queue.put_nowait((raw_id, body))
+        except asyncio.QueueFull:
+            err = json.dumps(
+                {
+                    "error": "queue_full",
+                    "detail": "Internal async buffer is saturated (backpressure)",
+                }
+            ).encode("utf-8")
+            await self._send_response(writer, 429, "application/json", err)
+            return
+
+        # Increment metric
+        api_metrics.increment_received(1)
+
+        # 202 Accepted
+        res = json.dumps(
+            {"status": "accepted", "raw_id": raw_id, "queued": True}
+        ).encode("utf-8")
+        await self._send_response(writer, 202, "application/json", res)
+
+    async def _send_response(
+        self,
+        writer: asyncio.StreamWriter,
+        status_code: int,
+        content_type: str,
+        body: bytes,
+    ) -> None:
+        status_reasons = {
+            200: "OK",
+            202: "Accepted",
+            400: "Bad Request",
+            404: "Not Found",
+            429: "Too Many Requests",
+            500: "Internal Server Error",
+            503: "Service Unavailable",
+        }
+        reason = status_reasons.get(status_code, "Unknown")
+        headers = (
+            f"HTTP/1.1 {status_code} {reason}\r\n"
+            f"Content-Type: {content_type}\r\n"
+            f"Content-Length: {len(body)}\r\n"
+            f"Connection: close\r\n"
+            f"\r\n"
+        )
+        writer.write(headers.encode("utf-8") + body)
+        await writer.drain()
+
+
+async def main():
+    server = IngestionAPIServer()
+    await server.start()
+    try:
+        # Keep running until cancelled
+        while True:
+            await asyncio.sleep(3600)
+    except (asyncio.CancelledError, KeyboardInterrupt):
+        await server.stop()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

app/api/stats.py

@@ -0,0 +1,72 @@
+import asyncio
+import json
+import urllib.request
+import urllib.error
+from typing import Any, Dict
+
+from app.api.metrics import api_metrics
+from app.common.logging import setup_logger
+
+logger = setup_logger("ingestion-api")
+
+
+def _sync_fetch_worker_stats(url: str, timeout: float = 2.0) -> Dict[str, Any]:
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    with urllib.request.urlopen(req, timeout=timeout) as response:
+        if response.status == 200:
+            content = response.read().decode("utf-8")
+            return json.loads(content)
+        raise ValueError(f"Worker returned status {response.status}")
+
+
+async def get_aggregated_pipeline_stats(worker_stats_url: str) -> Dict[str, Any]:
+    """
+    Combines local Ingestion API statistics with remote runtime statistics
+    from the Processing Worker.
+    """
+    total_received = api_metrics.total_received
+
+    try:
+        loop = asyncio.get_running_loop()
+        worker_stats = await loop.run_in_executor(
+            None, _sync_fetch_worker_stats, worker_stats_url
+        )
+
+        pipeline_data = worker_stats.get("pipeline", {})
+        breakdown_data = worker_stats.get("breakdown", {})
+
+        return {
+            "pipeline": {
+                "total_received": total_received,
+                "total_processed": pipeline_data.get("total_processed", 0),
+                "total_dlq": pipeline_data.get("total_dlq", 0),
+                "in_flight_buffer": pipeline_data.get("in_flight_buffer", 0),
+                "last_flush_at": pipeline_data.get("last_flush_at"),
+            },
+            "breakdown": {
+                "by_service_type": breakdown_data.get("by_service_type", {}),
+                "by_radio_access_type": breakdown_data.get("by_radio_access_type", {}),
+                "dlq_by_reason": breakdown_data.get("dlq_by_reason", {}),
+            },
+        }
+    except Exception as e:
+        logger.warning(
+            f"Unable to fetch worker stats from {worker_stats_url}: {e}",
+            extra={"event": "worker_stats_fetch_failed"},
+        )
+        # Fallback when worker is temporarily unreachable
+        return {
+            "pipeline": {
+                "total_received": total_received,
+                "total_processed": 0,
+                "total_dlq": 0,
+                "in_flight_buffer": 0,
+                "last_flush_at": None,
+                "worker_status": "unreachable",
+            },
+            "breakdown": {
+                "by_service_type": {},
+                "by_radio_access_type": {},
+                "dlq_by_reason": {},
+            },
+        }

app/common/__init__.py

@@ -1,2 +1 @@
 # app/common package
-

app/common/config.py

@@ -81,4 +81,3 @@ class AppConfig:
 
 
 config = AppConfig()
-

app/common/logging.py

@@ -85,4 +85,3 @@ def setup_logger(
         logger.addHandler(handler)
 
     return logger
-

docker/clickhouse/init.sql

@@ -39,4 +39,3 @@ PARTITION BY toYYYYMM(session_start_time)
 ORDER BY (service_type, radio_access_type, cell_id, session_start_time, msisdn)
 TTL session_start_time + INTERVAL 90 DAY
 SETTINGS index_granularity = 8192;
-

openspec/changes/telecom-cdr-pipeline/design.md

@@ -69,4 +69,3 @@ The architecture is constrained to pure Python without web frameworks (FastAPI,
 - **[Risk] ClickHouse outage stalls worker** → **[Mitigation]** Retries are bounded to 3 attempts with exponential backoff. If ClickHouse remains unreachable, batch is offloaded to DLQ and offsets committed to prevent infinite consumer lag loops.
 - **[Risk] Worker crashes between ClickHouse ACK and Kafka commit** → **[Mitigation]** Re-consumed batch is re-inserted; `ReplacingMergeTree(processed_at)` deduplicates rows sharing the order key `(service_type, radio_access_type, cell_id, session_start_time, msisdn)`.
 - **[Risk] Phone number parsing performance overhead** → **[Mitigation]** Use `phonenumbers` library with fast-path prefix checks (`00`, `011`, `+`) before full parsing.
-

openspec/changes/telecom-cdr-pipeline/proposal.md

@@ -30,4 +30,3 @@ Telecom network elements generate high-volume, heterogeneous Call Detail Records
 - **Dependencies**: `confluent-kafka`, `clickhouse-connect`, `phonenumbers` (no web frameworks).
 - **Infrastructure**: Kafka/Redpanda broker and ClickHouse database managed via Docker Compose.
 - **Data Flow**: Network Element → HTTP API → Kafka (`cdr.raw`) → Worker → ClickHouse (`cdr_records`) / DLQ (`cdr.raw.dlq`).
-

openspec/changes/telecom-cdr-pipeline/specs/clickhouse-storage/spec.md

@@ -24,4 +24,3 @@ The storage layer and worker client SHALL store `source_ip` using the ClickHouse
 #### Scenario: IPv4 string inserted into table
 - **WHEN** a normalized record with an IPv4 address string (e.g. `"10.240.12.1"`) is inserted
 - **THEN** the value is inserted as a valid IPv4 representation compatible with ClickHouse's internal `IPv4` (`UInt32`) structure using `toIPv4()` casting or native client IP types
-

openspec/changes/telecom-cdr-pipeline/specs/ingestion-api/spec.md

@@ -53,4 +53,3 @@ The Ingestion API SHALL expose a human-readable JSON status endpoint at `GET /st
 #### Scenario: Unified stats requested
 - **WHEN** an operator issues a `GET /stats` request
 - **THEN** the API queries the Worker's internal stats endpoint, aggregates `total_received` with worker metrics (`total_processed`, `total_dlq`, `in_flight_buffer`, `last_flush_at`, and breakdown maps), and returns HTTP status `200 OK` with structured JSON
-

openspec/changes/telecom-cdr-pipeline/specs/message-broker/spec.md

@@ -28,4 +28,3 @@ All messages published to `cdr.raw.dlq` SHALL be wrapped in a standardized JSON
 #### Scenario: Failure reason taxonomy enforced
 - **WHEN** any record is routed to `cdr.raw.dlq`
 - **THEN** the `failure_reason` value matches one of: `missing_required_field`, `invalid_rat_value`, `invalid_epoch`, `malformed_json`, or `normalization_error`
-

openspec/changes/telecom-cdr-pipeline/specs/observability/spec.md

@@ -30,4 +30,3 @@ All pipeline services SHALL emit log lines formatted as structured JSON to stdou
 #### Scenario: Operational event logged
 - **WHEN** any service logs an event (info, warning, or error)
 - **THEN** the log output is a single-line JSON object containing at minimum `timestamp` (ISO 8601 UTC), `level` (INFO, WARN, ERROR), `service` (`ingestion-api` or `processing-worker`), `raw_id` (record identifier or null), and `event` (short descriptive string)
-

openspec/changes/telecom-cdr-pipeline/specs/processing-worker/spec.md

@@ -95,4 +95,3 @@ The Processing Worker SHALL handle `SIGTERM` and `SIGINT` signals to flush buffe
 #### Scenario: SIGTERM received during operation
 - **WHEN** a `SIGTERM` signal is received by the worker process
 - **THEN** the worker stops consuming new messages, flushes any remaining in-memory buffered records to ClickHouse, commits the associated Kafka offsets, and exits cleanly with exit code 0
-

openspec/changes/telecom-cdr-pipeline/specs/verification/spec.md

@@ -46,4 +46,3 @@ The test script SHALL inspect ClickHouse, the DLQ, and API metrics after test ex
   - Records in DLQ
   - API 400 errors
   - Duplicate `raw_id`s resolved
-

openspec/changes/telecom-cdr-pipeline/tasks.md

@@ -6,11 +6,11 @@
 
 ## 2. Framework-less Async Ingestion API
 
-- [ ] 2.1 Implement lightweight HTTP/1.1 server using standard library `asyncio.start_server` with route dispatching for `POST /api/v1/cdr/raw`, `GET /metrics`, and `GET /stats`.
+- [x] 2.1 Implement lightweight HTTP/1.1 server using standard library `asyncio.start_server` with route dispatching for `POST /api/v1/cdr/raw`, `GET /metrics`, and `GET /stats`.
-- [ ] 2.2 Implement bounded async queue (`asyncio.Queue`) backpressure: verify `429 Too Many Requests` when queue is saturated and `202 Accepted` when enqueued.
+- [x] 2.2 Implement bounded async queue (`asyncio.Queue`) backpressure: verify `429 Too Many Requests` when queue is saturated and `202 Accepted` when enqueued.
-- [ ] 2.3 Implement incoming payload validation: verify `400 Bad Request` with structured JSON error `{ "error": "...", "detail": "..." }` on malformed JSON or missing required fields.
+- [x] 2.3 Implement incoming payload validation: verify `400 Bad Request` with structured JSON error `{ "error": "...", "detail": "..." }` on malformed JSON or missing required fields.
-- [ ] 2.4 Integrate `confluent-kafka` producer publishing to `cdr.raw` using `raw_id` as message key, and return `503 Service Unavailable` if Kafka is unreachable.
+- [x] 2.4 Integrate `confluent-kafka` producer publishing to `cdr.raw` using `raw_id` as message key, and return `503 Service Unavailable` if Kafka is unreachable.
-- [ ] 2.5 Implement Prometheus metrics endpoint (`cdr_api_received_total`) and public `GET /stats` endpoint that fetches worker runtime stats via internal HTTP and returns unified JSON.
+- [x] 2.5 Implement Prometheus metrics endpoint (`cdr_api_received_total`) and public `GET /stats` endpoint that fetches worker runtime stats via internal HTTP and returns unified JSON.
 
 ## 3. Processing Worker Implementation
 
@@ -32,4 +32,3 @@
 - [ ] 5.1 Implement traffic generator script covering the 7 mandatory scenarios (valid records, missing fields, malformed JSON, bad RAT, invalid epoch, duplicates, burst load).
 - [ ] 5.2 Implement automated verification summary reporting verifying ClickHouse counts, DLQ counts, API 400 errors, and duplicate resolution.
 - [ ] 5.3 Write comprehensive `README.md` covering quickstart, architecture, micro-batching, offset management, deduplication, monitoring, and configuration reference.
-

scripts/init_db.sql

@@ -39,4 +39,3 @@ PARTITION BY toYYYYMM(session_start_time)
 ORDER BY (service_type, radio_access_type, cell_id, session_start_time, msisdn)
 TTL session_start_time + INTERVAL 90 DAY
 SETTINGS index_granularity = 8192;
-

tests/test_api_server.py

@@ -0,0 +1,186 @@
+import asyncio
+import json
+import unittest
+
+from app.api.producer import MockMessageProducer
+from app.api.server import IngestionAPIServer
+
+
+class TestIngestionAPIServer(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.mock_producer = MockMessageProducer()
+        self.server = IngestionAPIServer(
+            host="127.0.0.1",
+            port=18000,
+            queue_max_size=2,  # small queue to easily test backpressure
+            producer=self.mock_producer,
+            worker_stats_url="http://127.0.0.1:18001/internal/stats",
+        )
+        await self.server.start()
+
+    async def asyncTearDown(self):
+        await self.server.stop()
+
+    async def _http_request(
+        self, method: str, path: str, body: bytes = b"", headers: dict = None
+    ) -> tuple[int, dict, bytes]:
+        reader, writer = await asyncio.open_connection("127.0.0.1", 18000)
+        req_headers = headers or {}
+        req_lines = [f"{method} {path} HTTP/1.1", "Host: 127.0.0.1:18000"]
+        for k, v in req_headers.items():
+            req_lines.append(f"{k}: {v}")
+        if body and "content-length" not in {k.lower() for k in req_headers}:
+            req_lines.append(f"Content-Length: {len(body)}")
+        req_lines.append("")
+        req_lines.append("")
+
+        raw_req = "\r\n".join(req_lines).encode("utf-8") + body
+        writer.write(raw_req)
+        await writer.drain()
+
+        # Read response status line
+        resp_line = await reader.readline()
+        parts = resp_line.decode("utf-8").strip().split(" ", 2)
+        status = int(parts[1])
+
+        # Read response headers
+        resp_headers = {}
+        while True:
+            line = await reader.readline()
+            if not line or line in (b"\r\n", b"\n"):
+                break
+            line_str = line.decode("utf-8").strip()
+            if ":" in line_str:
+                k, v = line_str.split(":", 1)
+                resp_headers[k.strip().lower()] = v.strip()
+
+        # Read response body
+        c_len = int(resp_headers.get("content-length", 0))
+        if c_len > 0:
+            resp_body = await reader.readexactly(c_len)
+        else:
+            resp_body = await reader.read()
+
+        writer.close()
+        await writer.wait_closed()
+        return status, resp_headers, resp_body
+
+    async def test_health_endpoint(self):
+        status, _, body = await self._http_request("GET", "/health")
+        self.assertEqual(status, 200)
+        data = json.loads(body.decode("utf-8"))
+        self.assertEqual(data["status"], "ok")
+
+    async def test_valid_cdr_ingestion(self):
+        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,
+        }
+        body = json.dumps(payload).encode("utf-8")
+        status, _, resp_body = await self._http_request(
+            "POST",
+            "/api/v1/cdr/raw",
+            body=body,
+            headers={"Content-Type": "application/json"},
+        )
+        self.assertEqual(status, 202)
+        data = json.loads(resp_body.decode("utf-8"))
+        self.assertEqual(data["status"], "accepted")
+        self.assertEqual(data["raw_id"], "raw-us-east-9988221")
+
+        # Let async worker dispatch to mock producer
+        await asyncio.sleep(0.05)
+        self.assertEqual(len(self.mock_producer.messages), 1)
+        self.assertEqual(self.mock_producer.messages[0]["key"], "raw-us-east-9988221")
+
+    async def test_malformed_json_returns_400(self):
+        body = b'{"raw_id": "bad-json", '  # Invalid JSON syntax
+        status, _, resp_body = await self._http_request(
+            "POST",
+            "/api/v1/cdr/raw",
+            body=body,
+            headers={"Content-Type": "application/json"},
+        )
+        self.assertEqual(status, 400)
+        data = json.loads(resp_body.decode("utf-8"))
+        self.assertEqual(data["error"], "invalid_json")
+        self.assertIn("detail", data)
+
+    async def test_non_dict_json_returns_400(self):
+        body = json.dumps(["not", "an", "object"]).encode("utf-8")
+        status, _, resp_body = await self._http_request(
+            "POST",
+            "/api/v1/cdr/raw",
+            body=body,
+            headers={"Content-Type": "application/json"},
+        )
+        self.assertEqual(status, 400)
+        data = json.loads(resp_body.decode("utf-8"))
+        self.assertEqual(data["error"], "invalid_payload")
+
+    async def test_backpressure_returns_429_when_queue_full(self):
+        # Cancel dispatcher task so it won't consume from queue
+        self.server.dispatcher._task.cancel()
+
+        # Fill the queue (maxsize=2)
+        await self.server.queue.put(("k1", b"v1"))
+        await self.server.queue.put(("k2", b"v2"))
+
+        # Third request must be rejected with 429
+        body = json.dumps({"raw_id": "k3"}).encode("utf-8")
+        status, _, resp_body = await self._http_request(
+            "POST",
+            "/api/v1/cdr/raw",
+            body=body,
+            headers={"Content-Type": "application/json"},
+        )
+        self.assertEqual(status, 429)
+        data = json.loads(resp_body.decode("utf-8"))
+        self.assertEqual(data["error"], "queue_full")
+
+    async def test_broker_unavailable_returns_503(self):
+        self.mock_producer.set_healthy(False)
+        body = json.dumps({"raw_id": "test"}).encode("utf-8")
+        status, _, resp_body = await self._http_request(
+            "POST",
+            "/api/v1/cdr/raw",
+            body=body,
+            headers={"Content-Type": "application/json"},
+        )
+        self.assertEqual(status, 503)
+        data = json.loads(resp_body.decode("utf-8"))
+        self.assertEqual(data["error"], "broker_unavailable")
+
+    async def test_metrics_endpoint(self):
+        status, headers, body = await self._http_request("GET", "/metrics")
+        self.assertEqual(status, 200)
+        content = body.decode("utf-8")
+        self.assertIn("cdr_api_received_total", content)
+
+    async def test_stats_endpoint(self):
+        status, _, body = await self._http_request("GET", "/stats")
+        self.assertEqual(status, 200)
+        data = json.loads(body.decode("utf-8"))
+        self.assertIn("pipeline", data)
+        self.assertIn("breakdown", data)
+        self.assertIn("total_received", data["pipeline"])
+
+    async def test_not_found_returns_404(self):
+        status, _, body = await self._http_request("GET", "/unknown/route")
+        self.assertEqual(status, 404)
+        data = json.loads(body.decode("utf-8"))
+        self.assertEqual(data["error"], "not_found")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_logging.py

@@ -48,4 +48,3 @@ class TestStructuredJsonFormatter(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
-