"""
第 15 章 · 安全与多租户
sasl_producer.py - 用 SASL_SSL + SCRAM-SHA-256 连接 Kafka 的 Python Producer

依赖：confluent-kafka >= 2.5
    pip install confluent-kafka

前提：
  1. Broker 已开启 SASL_SSL + SCRAM-SHA-256
  2. 已用 scram_user_setup.sh 创建用户 alice，密码 alice-secret
  3. 已用 acl_examples.sh 给 User:alice 加上 learn.security.demo Topic 的 Write 权限

用法：
  python sasl_producer.py
"""
from __future__ import annotations

import json
import os
import time
from confluent_kafka import Producer

TOPIC = "learn.security.demo"


def build_producer() -> Producer:
    conf = {
        # ------- 基础 -------
        "bootstrap.servers": os.getenv("KAFKA_BOOTSTRAP", "localhost:9093"),
        "client.id": "sasl-producer-demo",

        # ------- 安全协议 -------
        # SASL_SSL = TLS 加密 + SASL 鉴权
        "security.protocol": "SASL_SSL",

        # ------- SASL 机制 -------
        "sasl.mechanism": "SCRAM-SHA-256",
        "sasl.username": os.getenv("KAFKA_USERNAME", "alice"),
        "sasl.password": os.getenv("KAFKA_PASSWORD", "alice-secret"),

        # ------- TLS（验证 broker 证书；mTLS 时还需 keystore） -------
        # CA 证书：用于校验 broker 证书是否由可信 CA 签发
        "ssl.ca.location": os.getenv("KAFKA_CA_CERT", "/etc/kafka/ssl/ca.crt"),

        # 强烈建议开启 hostname 校验，防中间人；与 broker 端
        # ssl.endpoint.identification.algorithm=https 配套
        "ssl.endpoint.identification.algorithm": "https",

        # ------- 可靠性 -------
        "acks": "all",
        "enable.idempotence": True,   # ★ 注意：需要给账号 Cluster:IdempotentWrite ACL
        "compression.type": "zstd",
        "linger.ms": 20,
        "batch.size": 64 * 1024,
        "retries": 10,
        "delivery.timeout.ms": 120_000,
    }
    return Producer(conf)


def delivery_report(err, msg):
    if err is not None:
        # 鉴权失败时，常见错误码：
        #   _AUTHENTICATION（broker 拒绝凭证）
        #   _ALL_BROKERS_DOWN（TLS 握手失败时也可能报这个）
        print(f"❌ Delivery failed: {err}")
    else:
        print(f"✅ Delivered to {msg.topic()}[{msg.partition()}]@{msg.offset()}: "
              f"key={msg.key()!r} val={msg.value()!r}")


def main() -> None:
    producer = build_producer()
    print(f"🔐 Connecting as user={os.getenv('KAFKA_USERNAME', 'alice')} "
          f"to {os.getenv('KAFKA_BOOTSTRAP', 'localhost:9093')} via SASL_SSL+SCRAM-SHA-256")

    for i in range(10):
        payload = {"order_id": i, "ts": time.time(), "msg": f"hello-secure-{i}"}
        producer.produce(
            topic=TOPIC,
            key=str(i).encode(),
            value=json.dumps(payload).encode(),
            on_delivery=delivery_report,
        )
        # 触发回调（不是 flush）
        producer.poll(0)
        time.sleep(0.2)

    # 退出前 flush，确保所有 in-flight 消息全部 ack 或 error
    remaining = producer.flush(10)
    if remaining:
        print(f"⚠️  Still {remaining} messages in queue after flush timeout")
    else:
        print("🎉 All messages delivered")


if __name__ == "__main__":
    main()


# ============================================================================
# 故障排查指南
# ----------------------------------------------------------------------------
# 1) AUTHENTICATION_FAILED / SASL authentication failed
#    → 用户名 / 密码错；或者 broker 端没启用对应的 SASL 机制
#
# 2) UNKNOWN_SERVER_ERROR + broker log: SSL handshake failed
#    → ca.location 不对 / broker 证书过期 / 时间不同步
#
# 3) TOPIC_AUTHORIZATION_FAILED
#    → 用户没有 Topic:Write 的 ACL
#       kafka-acls.sh ... --add --allow-principal User:alice \
#         --operation Write --operation Describe --topic learn.security.demo
#
# 4) CLUSTER_AUTHORIZATION_FAILED（仅 enable.idempotence=True 时出现）
#    → 用户没有 Cluster:IdempotentWrite 的 ACL
#       kafka-acls.sh ... --add --allow-principal User:alice \
#         --operation IdempotentWrite --cluster
#
# 5) GROUP_AUTHORIZATION_FAILED（消费者端，本脚本不涉及）
#    → 用户没有 Group:Read
# ============================================================================