docs(specs): initialize OpenSpec and define telecom CDR pipeline specifications

openspec/changes/telecom-cdr-pipeline/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-09-19

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

@@ -0,0 +1,3 @@
+# telecom-cdr-pipeline
+
+Telecom Raw CDR Ingestion & Analytics Pipeline

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

@@ -0,0 +1,72 @@
+## Context
+
+See `proposal.md` for background and problem motivation.
+
+The architecture is constrained to pure Python without web frameworks (FastAPI, Flask, Starlette, or Django). The pipeline must ingest high-throughput telecom CDRs, decouple ingestion from processing via Kafka, normalize heterogeneous fields, micro-batch into ClickHouse with at-least-once delivery guarantees, and provide complete operational observability via Prometheus metrics and structured JSON logging.
+
+## Goals / Non-Goals
+
+**Goals:**
+- Implement a non-blocking HTTP ingestion server using Python standard library `asyncio` primitives.
+- Enforce backpressure using bounded async queues (`429 Too Many Requests`).
+- Publish to Kafka topic `cdr.raw` using `raw_id` partition keys with `confluent-kafka`.
+- Build a resilient worker that normalizes phone numbers (E.164 via `phonenumbers`), maps RATs, generates deterministic UUIDv5 `cdr_id`s, and safely routes bad data to `cdr.raw.dlq`.
+- Micro-batch inserts into ClickHouse (`ReplacingMergeTree`) with configurable count (1,000) and time (2s) flush thresholds.
+- Guarantee at-least-once delivery by committing Kafka offsets only after ClickHouse write ACK.
+- Expose dual-service Prometheus `/metrics` and aggregate public `/stats` via an internal worker HTTP endpoint.
+- Emit structured JSON logs across all services.
+- Containerize all services in `docker-compose.yml` with health checks and `condition: service_healthy`.
+
+**Non-Goals:**
+- Web-based visualization UI (analytical queries and verification scripts provide reporting).
+- Schema Registry broker integration (deferred from optional bonus challenges).
+- Distributed multi-worker consumer coordination (single worker instance fulfills baseline requirements).
+
+## Decisions
+
+### 1. Framework-less `asyncio` HTTP Server
+- **Choice**: Implement the HTTP listener using `asyncio.start_server` with a lightweight, zero-dependency HTTP/1.1 request parser.
+- **Rationale**: Strictly satisfies the constraint of "pure Python without frameworks" while delivering high concurrent connection handling and native integration with `asyncio.Queue` for backpressure.
+- **Alternatives Considered**: 
+  - `http.server.ThreadingHTTPServer`: Standard library, but thread-per-request model has higher memory overhead and makes bounded async backpressure coordination harder.
+  - `FastAPI`/`Starlette`: Disallowed by project constraints.
+
+### 2. Service Separation and Stats Aggregation
+- **Choice**: Run `Ingestion API` and `Processing Worker` as separate Docker containers. The public `/stats` endpoint is hosted by the Ingestion API, which queries a lightweight internal HTTP endpoint on the Worker (`GET http://worker:8001/internal/stats`) and aggregates with local ingestion metrics.
+- **Rationale**: Isolates HTTP ingestion from processing CPU/network bottlenecks. Enables independent scaling and lifecycle management.
+- **Alternatives Considered**:
+  - Single combined monolithic process: Violates explicit requirement for separate services.
+  - Querying ClickHouse directly for `/stats`: Incurs database query overhead and does not report real-time in-flight buffer size or consumer lag.
+
+### 3. Kafka Client and Partitioning Strategy
+- **Choice**: Use `confluent-kafka` with `raw_id` as the message key when publishing to `cdr.raw`.
+- **Rationale**: `confluent-kafka` is built on `librdkafka` and offers superior throughput over pure Python clients. Using `raw_id` ensures deterministic partition affinity for duplicate CDR submissions while spreading traffic evenly across the 3 partitions via Murmur2 hashing.
+- **Alternatives Considered**:
+  - `caller_num` as key: Risk of partition hotspots if certain subscribers generate disproportionate traffic.
+  - `null` (round-robin): Loses record-level partition affinity.
+
+### 4. ClickHouse Client and Idempotent Deduplication
+- **Choice**: Use `clickhouse-connect` HTTP client targeting table `telecom.cdr_records` with engine `ReplacingMergeTree(processed_at)`.
+- **Rationale**: ClickHouse recommends HTTP batch inserts. `ReplacingMergeTree` combined with deterministic UUIDv5 `cdr_id` ensures that if a batch is replayed after a worker crash prior to offset commit, duplicate records with identical order keys are safely merged without data duplication.
+- **Alternatives Considered**:
+  - `clickhouse-driver` (native TCP protocol): Slightly higher client-side setup complexity; `clickhouse-connect` is the official modern ClickHouse client.
+  - Standard `MergeTree`: Does not deduplicate replayed batches, violating idempotency.
+
+### 5. Micro-Batching and Fault Tolerance Strategy
+- **Choice**: Worker accumulates records into an in-memory batch. Flushes when `len(batch) >= BATCH_SIZE` (default 1,000) or elapsed time >= `BATCH_FLUSH_INTERVAL_SECONDS` (default 2s).
+- **Failure Handling**:
+  - If ClickHouse write times out, worker retries with exponential backoff up to `CH_RETRY_MAX_ATTEMPTS` (default 3, initial delay 500ms, factor 2.0).
+  - If all retries fail, the entire batch is wrapped and routed to `cdr.raw.dlq`, and offsets are committed to prevent pipeline stall.
+  - On `SIGTERM`, consumer stops, remaining buffer is flushed to ClickHouse, offsets are committed, and worker exits cleanly.
+
+### 6. Structured JSON Logging
+- **Choice**: Standard library Python logging with a custom JSON formatter producing fields: `timestamp`, `level`, `service`, `raw_id`, and `event`.
+- **Rationale**: Clean, zero-dependency implementation producing machine-readable logs compatible with modern observability stacks.
+
+## Risks / Trade-offs
+
+- **[Risk] High ingestion bursts saturate in-memory queue** → **[Mitigation]** The bounded `asyncio.Queue` rejects requests with `429 Too Many Requests` as soon as capacity is reached, protecting process memory.
+- **[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

@@ -0,0 +1,33 @@
+## Why
+
+Telecom network elements generate high-volume, heterogeneous Call Detail Records (CDRs) that must be ingested, validated, normalized, and stored in near real-time without data loss. This change establishes a resilient, high-throughput ingestion and analytics pipeline using pure Python (framework-less `asyncio`), Kafka, and ClickHouse to guarantee at-least-once delivery, backpressure protection, deterministic deduplication, and full observability.
+
+## What Changes
+
+- **Ingestion API**: A framework-less `asyncio` HTTP service accepting raw CDRs at `POST /api/v1/cdr/raw`, applying bounded queue backpressure (429), publishing to Kafka (`cdr.raw`), and serving `/metrics` and aggregated `/stats`.
+- **Message Broker & DLQ**: Kafka topic definitions for `cdr.raw` (3 partitions, keyed by `raw_id`) and `cdr.raw.dlq` (1 partition) with a standardized error envelope schema.
+- **Processing Worker**: A resilient consumer worker performing schema validation, E.164 normalization, RAT mapping, deterministic UUIDv5 ID generation, micro-batching (1,000 records or 2 seconds), and offset commits only after ClickHouse ACK.
+- **ClickHouse Storage**: `ReplacingMergeTree` table schema in `telecom.cdr_records` with automated idempotent schema initialization, `toIPv4()` casting, and 90-day TTL.
+- **Observability & Logging**: Structured JSON logging (`timestamp`, `level`, `service`, `raw_id`, `event`), Prometheus metrics (counters, histograms, gauges), and dual-service stats aggregation.
+- **Verification & Deployment**: Docker Compose orchestrating all services with health checks and `condition: service_healthy`, alongside an automated traffic generator script verifying all 7 test scenarios.
+
+## Capabilities
+
+### New Capabilities
+- `ingestion-api`: Framework-less `asyncio` HTTP gate, bounded queue backpressure, Kafka producer, `/metrics`, and aggregated `/stats`.
+- `message-broker`: Topic definitions, partition strategy with `raw_id`, and DLQ envelope schema with failure taxonomy.
+- `processing-worker`: Consumer loop, data cleansing, E.164 normalization, RAT mapping, micro-batching, at-least-once offset management, and graceful shutdown.
+- `clickhouse-storage`: Database DDL, `ReplacingMergeTree` table design, `toIPv4()` casting, and idempotent initialization.
+- `observability`: Structured JSON logging, Prometheus metrics definitions, and internal/public stats reporting.
+- `verification`: Automated traffic generator covering 7 scenarios and pipeline verification reporting.
+
+### Modified Capabilities
+<!-- None: Initial implementation -->
+
+## Impact
+
+- **New Services**: Ingestion API service and Processing Worker service.
+- **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

@@ -0,0 +1,27 @@
+## Purpose
+
+Defines the ClickHouse database schema, table engines, deduplication semantics, and automated idempotent initialization for storing normalized CDRs.
+
+## ADDED Requirements
+
+### Requirement: Database and Table Schema Definition
+The storage layer SHALL provide a database named `telecom` and a table named `cdr_records` optimized for high-volume append and analytical queries.
+
+#### Scenario: Database and table initialized idempotently
+- **WHEN** the initialization script or container entrypoint executes
+- **THEN** ClickHouse creates database `telecom` (`CREATE DATABASE IF NOT EXISTS`) and table `telecom.cdr_records` (`CREATE TABLE IF NOT EXISTS`) using `ReplacingMergeTree(processed_at)`, partitioned by `toYYYYMM(session_start_time)`, ordered by `(service_type, radio_access_type, cell_id, session_start_time, msisdn)`, with `TTL session_start_time + INTERVAL 90 DAY` and `SETTINGS index_granularity = 8192`
+
+### Requirement: Deduplication on At-Least-Once Replay
+The storage layer SHALL support idempotent re-insertion of records without creating duplicate entries in analytical views.
+
+#### Scenario: Duplicate record inserted after crash replay
+- **WHEN** a batch containing previously written records is re-inserted due to consumer restart before offset commit
+- **THEN** ClickHouse accepts the rows into `ReplacingMergeTree(processed_at)` and deduplicates rows sharing the same order key upon merge or when queried with `FINAL`
+
+### Requirement: Source IP Column Type Compatibility
+The storage layer and worker client SHALL store `source_ip` using the ClickHouse `IPv4` data type.
+
+#### 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

@@ -0,0 +1,56 @@
+## Purpose
+
+Provides a fast, non-blocking HTTP ingestion gate for raw vendor CDRs with backpressure control, publishing to Kafka, and unified operational statistics.
+
+## ADDED Requirements
+
+### Requirement: Raw CDR Ingestion Endpoint
+The Ingestion API SHALL provide an HTTP endpoint at `POST /api/v1/cdr/raw` to ingest vendor-specific CDR JSON payloads.
+
+#### Scenario: Valid CDR payload accepted
+- **WHEN** a client sends a `POST /api/v1/cdr/raw` request with valid JSON containing all required fields (`raw_id`, `device_id`, `caller_num`, `svc`, `rat`, `epoch_start`, `epoch_end`, `up_bytes`, `down_bytes`, `lac`, `ci`) and the internal buffer has capacity
+- **THEN** the API returns HTTP status `202 Accepted` with a structured JSON confirmation and queues the payload for Kafka publishing
+
+#### Scenario: Malformed JSON payload rejected
+- **WHEN** a client sends a `POST /api/v1/cdr/raw` request with an unparseable or corrupted JSON body
+- **THEN** the API returns HTTP status `400 Bad Request` with structured JSON `{"error": "malformed_json", "detail": "<error_message>"}`
+
+#### Scenario: Missing required fields rejected
+- **WHEN** a client sends a `POST /api/v1/cdr/raw` request missing one or more required fields (such as `caller_num` or `raw_id`)
+- **THEN** the API returns HTTP status `400 Bad Request` with structured JSON `{"error": "missing_required_field", "detail": "Field <field_name> is required"}`
+
+### Requirement: Bounded Async Queue Backpressure
+The Ingestion API SHALL enforce a bounded in-memory async queue to prevent unbounded memory growth when downstream publishing is delayed.
+
+#### Scenario: Internal queue capacity saturated
+- **WHEN** incoming request rate exceeds publishing capacity and the bounded async queue reaches its maximum configured limit
+- **THEN** the API immediately returns HTTP status `429 Too Many Requests` with structured JSON `{"error": "queue_full", "detail": "Ingestion buffer saturated"}` without blocking the event loop
+
+### Requirement: Broker Availability Protection
+The Ingestion API SHALL detect Kafka unavailability and return a service unavailable response.
+
+#### Scenario: Kafka broker unreachable
+- **WHEN** the Kafka producer fails to connect or queue messages within the configured timeout
+- **THEN** the API returns HTTP status `503 Service Unavailable` with structured JSON `{"error": "broker_unavailable", "detail": "Kafka broker unreachable"}`
+
+### Requirement: Partition Keyed Kafka Publishing
+The Ingestion API SHALL publish accepted raw records to the Kafka topic `cdr.raw` using `raw_id` as the message key to preserve record affinity.
+
+#### Scenario: Record published with raw_id partition key
+- **WHEN** a valid raw CDR record is queued for dispatch
+- **THEN** the producer publishes the message bytes to `cdr.raw` with Kafka message key set to the record's `raw_id`
+
+### Requirement: Prometheus Metrics Endpoint
+The Ingestion API SHALL expose a Prometheus metrics endpoint at `GET /metrics`.
+
+#### Scenario: Metrics scraped by Prometheus
+- **WHEN** a Prometheus scraper issues a `GET /metrics` request to the Ingestion API
+- **THEN** the API returns HTTP status `200 OK` with text-based Prometheus metrics including `cdr_api_received_total`
+
+### Requirement: Unified Pipeline Stats Endpoint
+The Ingestion API SHALL expose a human-readable JSON status endpoint at `GET /stats` combining local API ingestion counters with runtime metrics retrieved from the Processing Worker.
+
+#### 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

@@ -0,0 +1,31 @@
+## Purpose
+
+Defines message broker topics, partitioning, retention, and dead-letter queue envelope specifications for decoupling and reliable asynchronous buffer management.
+
+## ADDED Requirements
+
+### Requirement: Raw Ingestion Topic Configuration
+The message broker SHALL provide a topic named `cdr.raw` configured for high-throughput buffering of raw vendor CDR payloads.
+
+#### Scenario: Raw topic configuration applied
+- **WHEN** the message broker initializes or topic metadata is queried
+- **THEN** topic `cdr.raw` exists with 3 partitions, replication factor of 1, retention time of 86,400,000 ms (24 hours), and LZ4 compression enabled
+
+### Requirement: Dead Letter Queue Topic Configuration
+The message broker SHALL provide a topic named `cdr.raw.dlq` to retain unprocessable or malformed records for inspection and replay.
+
+#### Scenario: DLQ topic configuration applied
+- **WHEN** the message broker initializes or topic metadata is queried
+- **THEN** topic `cdr.raw.dlq` exists with 1 partition, replication factor of 1, and retention time of 604,800,000 ms (7 days)
+
+### Requirement: Standardized DLQ Message Envelope
+All messages published to `cdr.raw.dlq` SHALL be wrapped in a standardized JSON envelope containing error diagnostic metadata.
+
+#### Scenario: DLQ envelope structure enforced
+- **WHEN** a message is routed to `cdr.raw.dlq` due to processing or validation failure
+- **THEN** the message payload contains a JSON object with top-level fields: `original_payload` (object or string), `failure_reason` (string), `failed_field` (nullable string), `failed_at` (ISO 8601 UTC timestamp), and `worker_instance` (string identifier)
+
+#### 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

@@ -0,0 +1,33 @@
+## Purpose
+
+Defines Prometheus metrics collection, operational stats reporting, and structured JSON logging across pipeline components.
+
+## ADDED Requirements
+
+### Requirement: Prometheus Metrics Exposition
+Each pipeline service SHALL expose standard Prometheus metrics at `GET /metrics`.
+
+#### Scenario: Ingestion API metrics exposed
+- **WHEN** Prometheus scrapes `GET /metrics` on the Ingestion API
+- **THEN** the response includes counter `cdr_api_received_total`
+
+#### Scenario: Processing Worker metrics exposed
+- **WHEN** Prometheus scrapes `GET /metrics` on the Processing Worker
+- **THEN** the response includes counters (`cdr_processed_total{service_type, radio_access_type}`, `cdr_dlq_total{reason}`), histograms (`cdr_kafka_to_clickhouse_latency_seconds`, `cdr_batch_flush_duration_seconds`, `cdr_batch_size_at_flush`), and gauges (`cdr_in_flight_buffer_size`, `cdr_kafka_consumer_lag{partition}`)
+
+### Requirement: Human-Readable Pipeline Stats JSON
+The Ingestion API SHALL serve a unified JSON summary at `GET /stats` representing end-to-end pipeline health.
+
+#### Scenario: Stats endpoint returns unified summary
+- **WHEN** a client issues `GET /stats` to the Ingestion API
+- **THEN** the API returns HTTP 200 with JSON matching the structure:
+  - `pipeline`: `total_received`, `total_processed`, `total_dlq`, `in_flight_buffer`, `last_flush_at`
+  - `breakdown`: `by_service_type`, `by_radio_access_type`, `dlq_by_reason`
+
+### Requirement: Structured JSON Logging
+All pipeline services SHALL emit log lines formatted as structured JSON to stdout.
+
+#### 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

@@ -0,0 +1,98 @@
+## Purpose
+
+Defines stream consumption, safe parsing, schema normalization, micro-batching into ClickHouse, and at-least-once offset management for raw telecom CDRs.
+
+## ADDED Requirements
+
+### Requirement: Safe Parsing and Fault Tolerance
+The Processing Worker SHALL consume messages from `cdr.raw` and process them without crashing, routing malformed or unprocessable payloads to `cdr.raw.dlq`.
+
+#### Scenario: Malformed or unparseable JSON payload
+- **WHEN** a message consumed from `cdr.raw` cannot be parsed as valid JSON
+- **THEN** the worker wraps the raw message in a DLQ envelope with `failure_reason: "malformed_json"` and publishes it to `cdr.raw.dlq` without terminating the process
+
+#### Scenario: Missing required field
+- **WHEN** a consumed record is missing any required field (`raw_id`, `device_id`, `caller_num`, `svc`, `rat`, `epoch_start`, `epoch_end`, `up_bytes`, `down_bytes`, `lac`, `ci`) or contains null for a required field
+- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "missing_required_field"` and `failed_field` set to the missing field name
+
+#### Scenario: Inverted epoch timestamps
+- **WHEN** a consumed record has `epoch_start > epoch_end`
+- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "invalid_epoch"` and `failed_field: "epoch_start"`
+
+#### Scenario: Negative byte counts
+- **WHEN** a consumed record has `up_bytes < 0` or `down_bytes < 0`
+- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "normalization_error"` and `failed_field` set to the negative byte field
+
+### Requirement: E.164 Phone Number Normalization
+The Processing Worker SHALL normalize `caller_num` into a valid E.164 format `msisdn` using the `phonenumbers` library.
+
+#### Scenario: Leading international prefix normalization
+- **WHEN** a valid record contains `caller_num` starting with `00` or `011`
+- **THEN** the worker replaces the prefix with `+`, parses the number using `phonenumbers`, and sets `msisdn` to the standardized E.164 string
+
+#### Scenario: Existing E.164 number preserved
+- **WHEN** a record contains `caller_num` already formatted with a leading `+`
+- **THEN** the worker parses and validates the number using `phonenumbers` and outputs the verified E.164 string
+
+#### Scenario: Unparseable phone number routed to DLQ
+- **WHEN** `caller_num` cannot be normalized or validated by `phonenumbers`
+- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "normalization_error"` and `failed_field: "caller_num"`
+
+### Requirement: Radio Access Type Mapping
+The Processing Worker SHALL map the integer `rat` field to standard string representations.
+
+#### Scenario: Supported RAT values mapped
+- **WHEN** a record has `rat` equal to `1`, `6`, or `9`
+- **THEN** the worker maps `radio_access_type` to `"3G"`, `"LTE_4G"`, or `"5G_NR"` respectively
+
+#### Scenario: Unsupported RAT value routed to DLQ
+- **WHEN** a record has `rat` not equal to 1, 6, or 9
+- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "invalid_rat_value"` and `failed_field: "rat"`
+
+### Requirement: Schema Transformation and Deterministic Identity
+The Processing Worker SHALL transform valid raw CDRs into the normalized schema with deterministic identity derivation.
+
+#### Scenario: Deterministic UUIDv5 cdr_id generation
+- **WHEN** a raw record with `raw_id` is normalized
+- **THEN** the worker computes `cdr_id` as `uuid.uuid5(uuid.NAMESPACE_DNS, raw_id)` ensuring identical IDs for reprocessed duplicate records
+
+#### Scenario: Computed metrics and cell identification
+- **WHEN** raw fields are mapped to the normalized schema
+- **THEN** `cell_id` is set to `lac + "_" + ci`, `total_bytes` is computed as `up_bytes + down_bytes`, `duration_seconds` is computed as `epoch_end - epoch_start`, `session_start_time` and `session_end_time` are converted to UTC DateTimes, and `source_ip` defaults to `"0.0.0.0"` if absent
+
+### Requirement: Micro-Batching Buffer Strategy
+The Processing Worker SHALL buffer normalized records in memory and flush batches to ClickHouse based on size or elapsed time thresholds.
+
+#### Scenario: Batch flushed on count threshold
+- **WHEN** the in-memory normalized record buffer reaches `BATCH_SIZE` (default 1,000 records)
+- **THEN** the worker immediately flushes the batch to ClickHouse
+
+#### Scenario: Batch flushed on time interval threshold
+- **WHEN** `BATCH_FLUSH_INTERVAL_SECONDS` (default 2 seconds) elapses since the last flush and the buffer contains at least one record
+- **THEN** the worker flushes the accumulated buffer to ClickHouse
+
+### Requirement: At-Least-Once Delivery and Offset Management
+The Processing Worker SHALL commit Kafka consumer offsets only after ClickHouse successfully acknowledges the batch write.
+
+#### Scenario: Offsets committed after successful write
+- **WHEN** a micro-batch is written and acknowledged by ClickHouse
+- **THEN** the worker commits the highest offsets corresponding to the records in that batch
+
+#### Scenario: ClickHouse write timeout retry with exponential backoff
+- **WHEN** a batch write to ClickHouse fails or times out
+- **THEN** the worker retries the write up to `CH_RETRY_MAX_ATTEMPTS` (default 3) with exponential backoff (`CH_RETRY_INITIAL_DELAY_MS` multiplied by `CH_RETRY_BACKOFF_FACTOR`); if all attempts fail, the worker routes all records in the batch to `cdr.raw.dlq` and commits offsets to avoid pipeline deadlock
+
+### Requirement: Internal Runtime Statistics Endpoint
+The Processing Worker SHALL expose a lightweight internal HTTP endpoint (e.g., `GET /internal/stats`) to provide its runtime statistics to the Ingestion API.
+
+#### Scenario: Internal stats queried
+- **WHEN** an HTTP request is made to the worker's internal stats endpoint
+- **THEN** the worker returns HTTP status `200 OK` with JSON containing `total_processed`, `total_dlq`, `in_flight_buffer`, `last_flush_at`, and breakdown counts by service type, RAT, and DLQ reason
+
+### Requirement: Graceful Shutdown Without Data Loss
+The Processing Worker SHALL handle `SIGTERM` and `SIGINT` signals to flush buffered records before terminating.
+
+#### 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

@@ -0,0 +1,49 @@
+## Purpose
+
+Defines the automated test traffic generator, scenario coverage, and end-to-end pipeline verification reporting requirements.
+
+## ADDED Requirements
+
+### Requirement: Automated Traffic Generator Scenario Coverage
+The test suite SHALL provide a script that generates synthetic CDR traffic to exercise all core pipeline behaviors.
+
+#### Scenario: Valid records ingestion
+- **WHEN** the generator sends 1,000 valid records with varied RAT and service types
+- **THEN** all 1,000 records are successfully stored in ClickHouse and `cdr_processed_total` increments accordingly
+
+#### Scenario: Missing required field validation
+- **WHEN** the generator sends 20 records missing `caller_num`
+- **THEN** all 20 records are rejected by the API with HTTP 400 or routed to DLQ with `missing_required_field`
+
+#### Scenario: Malformed JSON validation
+- **WHEN** the generator sends 10 invalid JSON payloads
+- **THEN** the API returns HTTP 400 with structured JSON error details
+
+#### Scenario: Unknown RAT routing
+- **WHEN** the generator sends 10 records with unsupported `rat` values (such as `rat: 99`)
+- **THEN** all 10 records land in `cdr.raw.dlq` with `invalid_rat_value`
+
+#### Scenario: Invalid epoch timestamps
+- **WHEN** the generator sends 10 records with `epoch_start > epoch_end`
+- **THEN** all 10 records land in `cdr.raw.dlq` with `invalid_epoch`
+
+#### Scenario: Duplicate record deduplication
+- **WHEN** the generator sends 20 duplicate records with identical `raw_id`
+- **THEN** ClickHouse deduplicates the rows so that only 1 record per unique `raw_id` is retained
+
+#### Scenario: High-volume burst resilience
+- **WHEN** the generator sends a burst of 500 records in less than 5 seconds
+- **THEN** the API accepts the requests without returning 429 and all 500 records are processed without loss
+
+### Requirement: Verification Summary Reporting
+The test script SHALL inspect ClickHouse, the DLQ, and API metrics after test execution and output a verification summary.
+
+#### Scenario: Summary report printed
+- **WHEN** all traffic generation scenarios complete and assertions are verified
+- **THEN** the script prints a formatted summary comparing:
+  - Records sent
+  - Records in ClickHouse
+  - Records in DLQ
+  - API 400 errors
+  - Duplicate `raw_id`s resolved
+

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

@@ -0,0 +1,35 @@
+## 1. Project Setup & Common Utilities
+
+- [ ] 1.1 Create project structure, `requirements.txt` (with `confluent-kafka`, `clickhouse-connect`, `phonenumbers`), and `.env.example` defining all environment variables and verify file presence.
+- [ ] 1.2 Implement standard library structured JSON logging module emitting `timestamp`, `level`, `service`, `raw_id`, and `event`, and verify output formatting with unit test.
+- [ ] 1.3 Create idempotent ClickHouse initialization script (`init.sql`) creating database `telecom` and `ReplacingMergeTree` table `cdr_records` with TTL and indexing settings, and verify schema correctness.
+
+## 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`.
+- [ ] 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.
+- [ ] 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.
+
+## 3. Processing Worker Implementation
+
+- [ ] 3.1 Implement Kafka consumer loop with `confluent-kafka` consuming from `cdr.raw` and DLQ producer publishing to `cdr.raw.dlq`.
+- [ ] 3.2 Implement safe parser routing malformed records to DLQ with structured envelope and reason taxonomy (`missing_required_field`, `invalid_rat_value`, `invalid_epoch`, `malformed_json`, `normalization_error`).
+- [ ] 3.3 Implement schema normalizer: E.164 phone conversion via `phonenumbers`, RAT mapping (`1`→`3G`, `6`→`LTE_4G`, `9`→`5G_NR`), `cell_id` derivation (`lac_ci`), and deterministic UUIDv5 `cdr_id` generation.
+- [ ] 3.4 Implement micro-batching buffer flushing on 1,000 records or 2 seconds elapsed, batch-inserting into ClickHouse via `clickhouse-connect`, and committing offsets only after ClickHouse ACK.
+- [ ] 3.5 Implement exponential backoff retry for ClickHouse write timeouts (`CH_RETRY_MAX_ATTEMPTS`, `CH_RETRY_INITIAL_DELAY_MS`, `CH_RETRY_BACKOFF_FACTOR`) with fallback to DLQ.
+- [ ] 3.6 Implement worker internal HTTP stats endpoint (`GET /internal/stats`) and worker Prometheus metrics (`cdr_processed_total`, `cdr_dlq_total`, histograms, gauges).
+- [ ] 3.7 Implement `SIGTERM` and `SIGINT` signal handlers to stop consuming, flush remaining in-memory buffer to ClickHouse, commit offsets, and exit cleanly without data loss.
+
+## 4. Containerization & Orchestration
+
+- [ ] 4.1 Create lightweight Dockerfiles for Ingestion API and Processing Worker without framework dependencies.
+- [ ] 4.2 Create `docker-compose.yml` defining Kafka/Redpanda, ClickHouse, Ingestion API, and Processing Worker with healthcheck blocks and `depends_on: { condition: service_healthy }`.
+
+## 5. Verification & Test Traffic Generator
+
+- [ ] 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.
+