"""Independent, run-scoped Kafka evidence for the assessment verifier."""
import json
import time
import uuid

from confluent_kafka import Consumer, Producer, TopicPartition


class KafkaEvidence:
    def __init__(self, brokers, raw_topic, dlq_topic, run_prefix):
        self.prefix = run_prefix
        self.raw_topic = raw_topic
        self.consumer = Consumer({"bootstrap.servers": brokers,
                                  "group.id": f"verification-{uuid.uuid4()}",
                                  "enable.auto.commit": False})
        try:
            metadata = self.consumer.list_topics(dlq_topic, timeout=10).topics[dlq_topic]
            if metadata.error:
                raise RuntimeError(str(metadata.error))
            positions = []
            for partition in metadata.partitions:
                tp = TopicPartition(dlq_topic, partition)
                _, high = self.consumer.get_watermark_offsets(tp, timeout=10)
                positions.append(TopicPartition(dlq_topic, partition, high))
            self.consumer.assign(positions)
            self.producer = Producer({"bootstrap.servers": brokers, "acks": "all",
                                      "enable.idempotence": True, "delivery.timeout.ms": 10000})
        except Exception:
            self.consumer.close()
            raise
        self.expected = {}
        self.records = []

    def inject(self, key, body, reason):
        self.expected[key] = (body, reason)
        errors = []
        self.producer.produce(self.raw_topic, key=key, value=body,
                              on_delivery=lambda err, msg: errors.append(err) if err else None)
        if self.producer.flush(12) or errors:
            raise RuntimeError(f"Direct Kafka injection failed: {errors}")

    def collect(self, expected_count, timeout):
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            msg = self.consumer.poll(0.1)
            if msg is None:
                if len(self.records) >= expected_count:
                    break
                continue
            if msg.error():
                raise RuntimeError(str(msg.error()))
            envelope = json.loads(msg.value())
            if self.prefix in json.dumps(envelope.get("original_payload")):
                self.records.append(envelope)
        return self.records

    def validate(self):
        seen = {}
        for envelope in self.records:
            if not all(name in envelope for name in ("original_payload", "failure_reason", "failed_field", "failed_at", "worker_instance")):
                return False
            original = envelope["original_payload"]
            if isinstance(original, str):
                try:
                    original = json.loads(original)
                except ValueError:
                    pass
            matched = [key for key in self.expected if key in json.dumps(original)]
            if len(matched) != 1:
                return False
            key = matched[0]
            body, reason = self.expected[key]
            try:
                expected_original = json.loads(body)
            except ValueError:
                expected_original = body.decode()
            if original != expected_original or envelope["failure_reason"] != reason:
                return False
            expected_field = {"missing_required_field": "caller_num", "invalid_rat_value": "rat",
                              "invalid_epoch": "epoch_start", "malformed_json": None}[reason]
            if envelope["failed_field"] != expected_field or not envelope["worker_instance"]:
                return False
            seen[key] = seen.get(key, 0) + 1
        return set(seen) == set(self.expected) and all(count == 1 for count in seen.values())

    def close(self):
        self.consumer.close()