#!/usr/bin/env python3
"""
Notify Consumer：通知消费者
- 模拟向用户/商家发短信/IM 通知
- 失败有限次重试 → DLQ
"""

import os
import random
import time
from confluent_kafka import Consumer, Producer
from confluent_kafka.serialization import SerializationContext, MessageField
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer

BOOTSTRAP = os.getenv("BOOTSTRAP", "localhost:9092")
SR_URL = os.getenv("SR_URL", "http://localhost:8081")
TOPIC = "orders.events"
DLQ = "orders.events.dlq"
GROUP = "notify-group"


def send_notification(event):
    """模拟通知，5% 概率失败。"""
    if random.random() < 0.05:
        raise RuntimeError("notification gateway timeout")
    print(f"[notify] 📧 user={event['user_id']} order={event['order_id']} 已通知")


def main():
    sr = SchemaRegistryClient({"url": SR_URL})
    deser = AvroDeserializer(sr, from_dict=lambda d, ctx: d)
    consumer = Consumer({
        "bootstrap.servers": BOOTSTRAP,
        "group.id": GROUP,
        "auto.offset.reset": "earliest",
        "enable.auto.commit": False,
        "max.poll.interval.ms": 300_000,
        "partition.assignment.strategy": "cooperative-sticky",
    })
    dlq_p = Producer({"bootstrap.servers": BOOTSTRAP})
    consumer.subscribe([TOPIC])
    print(f"[notify] start, group={GROUP}")

    retries = {}
    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None or msg.error():
                continue

            key = (msg.partition(), msg.offset())
            try:
                event = deser(msg.value(), SerializationContext(TOPIC, MessageField.VALUE))
                send_notification(event)
                consumer.commit(msg, asynchronous=False)
                retries.pop(key, None)
            except Exception as e:
                retries[key] = retries.get(key, 0) + 1
                if retries[key] < 3:
                    print(f"[notify][ERR] retry {retries[key]}/3: {e}")
                    time.sleep(0.5)
                else:
                    print(f"[notify] → DLQ after 3 retries")
                    dlq_p.produce(DLQ, key=msg.key(), value=msg.value(),
                                  headers=[("error", str(e).encode()),
                                           ("origin", b"notify")])
                    dlq_p.flush(5)
                    consumer.commit(msg, asynchronous=False)
                    retries.pop(key, None)
    except KeyboardInterrupt:
        pass
    finally:
        consumer.close()


if __name__ == "__main__":
    main()