#!/usr/bin/env python3
"""
Inventory Consumer：库存消费者

特点：
- 业务幂等：用 (order_id, sku) 作为 DB 唯一约束，重复消费不会扣两次
- 失败 ≥ 3 次 → DLQ
"""

import os
import time
import json
import pymysql
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 = "inventory-group"

MY_CFG = dict(
    host=os.getenv("MY_HOST", "127.0.0.1"),
    port=int(os.getenv("MY_PORT", "3306")),
    user=os.getenv("MY_USER", "root"),
    password=os.getenv("MY_PWD", "root"),
    database="shop",
    autocommit=True,
)


def ensure_dedup_table():
    conn = pymysql.connect(**MY_CFG)
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS inventory_dedup (
              order_id VARCHAR(64),
              sku      VARCHAR(64),
              qty      INT,
              applied_at DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3),
              PRIMARY KEY (order_id, sku)
            ) ENGINE=InnoDB
        """)
    conn.close()


def apply_inventory(event):
    """利用 PRIMARY KEY 实现幂等。"""
    conn = pymysql.connect(**MY_CFG)
    try:
        with conn.cursor() as cur:
            for item in event.get("items", []):
                try:
                    cur.execute(
                        "INSERT INTO inventory_dedup (order_id, sku, qty) VALUES (%s, %s, %s)",
                        (event["order_id"], item["sku"], item["qty"]),
                    )
                    # 这里可以接真实库存表的扣减
                    print(f"[inv]   - 扣 sku={item['sku']} qty={item['qty']}")
                except pymysql.IntegrityError:
                    print(f"[inv]   = 已扣过，幂等跳过 sku={item['sku']}")
    finally:
        conn.close()


def main():
    ensure_dedup_table()

    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": 600_000,
        "max.poll.records": 50,
        "partition.assignment.strategy": "cooperative-sticky",
    })
    dlq_p = Producer({"bootstrap.servers": BOOTSTRAP})
    consumer.subscribe([TOPIC])
    print(f"[inv] start, group={GROUP}")

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

            key = (msg.partition(), msg.offset())
            try:
                event = deser(msg.value(), SerializationContext(TOPIC, MessageField.VALUE))
                if event["status"] != "CREATED":
                    consumer.commit(msg, asynchronous=False)
                    continue

                print(f"[inv] order={event['order_id']} 扣库存 ...")
                apply_inventory(event)
                consumer.commit(msg, asynchronous=False)
                retry_count.pop(key, None)
            except Exception as e:
                retry_count[key] = retry_count.get(key, 0) + 1
                print(f"[inv][ERR] retry {retry_count[key]}/3: {e}")
                if retry_count[key] >= 3:
                    print(f"[inv] → DLQ {key}")
                    dlq_p.produce(DLQ, key=msg.key(), value=msg.value(),
                                  headers=[("error", str(e).encode()),
                                           ("origin", b"inventory")])
                    dlq_p.flush(5)
                    consumer.commit(msg, asynchronous=False)
                    retry_count.pop(key, None)
                else:
                    time.sleep(1)
    except KeyboardInterrupt:
        pass
    finally:
        consumer.close()


if __name__ == "__main__":
    main()