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

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 canonical initialization schema (app/worker/schema.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 outstanding-delivery backpressure: verify 429 Too Many Requests at capacity and 202 Accepted only after broker acknowledgement.
• 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.

6. Delivery Safety Regression Gates

• 6.1 Add and pass unit regressions for confirmed API delivery, shutdown, strict validation and HTTP limits.
• 6.2 Add and pass worker regressions for partition ordering, DLQ failure, commit failure, replay, batching, range checks and telemetry.
• 6.3 Run isolated Compose integration with independent ports/storage, fresh defaults, schema collision/replay and failure recovery checks.
• 6.4 Run actual DLQ traffic verification and the acknowledged 500-record burst; record evidence in README.
• 6.5 Validate OpenSpec and reconcile all task completion claims with evidence.

Current evidence

The current files pass 64 pytest tests, resolved Compose configuration validation, git diff --check, and strict OpenSpec validation. The isolated 2026-09-20 Compose assessment passed all seven traffic scenarios with run-correlated ClickHouse and Kafka DLQ evidence, metric deltas, schema collision protection, broker recovery, worker restart/replay, and the acknowledged 500-record burst. The disposable project and its volumes were removed after verification.