package com.kafka;

import com.common.entity.DeviceDevice;
import com.haobang.config.AppProperties;
import com.haobang.util.DeviceInfoUtil;
import com.haobang.util.Sm4Util;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Locale;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * 进程级单例 Kafka Producer。
 *
 * <p>保留历史类名以避免影响现有调用方，但实例由 Spring 托管。KafkaProducer 本身线程安全，
 * 长生命周期复用才能使用连接、RecordAccumulator、批量和压缩；逐条 new/close 会破坏这些机制。</p>
 */
@Slf4j
@Component
public class kafkaProducer {

    private static final ZoneId EVENT_ID_ZONE = ZoneId.of("Asia/Shanghai");
    private static final DateTimeFormatter EVENT_ID_DATE = DateTimeFormatter.ofPattern("yyyyMMdd");

    private final AppProperties appProperties;
    private final Producer<String, String> producer;
    private final PartitionKeyStrategy partitionKeyStrategy;

    @Autowired
    public kafkaProducer(AppProperties appProperties,
                         @Value("${spring.kafka.producer.properties.retries:10}") int retries,
                         @Value("${spring.kafka.producer.properties.delivery.timeout.ms:120000}") int deliveryTimeoutMs,
                         @Value("${spring.kafka.producer.properties.request.timeout.ms:30000}") int requestTimeoutMs,
                         @Value("${spring.kafka.producer.properties.linger.ms:5}") int lingerMs,
                         @Value("${spring.kafka.producer.properties.batch.size:65536}") int batchSize,
                         @Value("${spring.kafka.producer.compression-type:lz4}") String compressionType,
                         @Value("${spring.kafka.producer.properties.max.block.ms:10000}") int maxBlockMs,
                         @Value("${spring.kafka.producer.properties.buffer.memory:134217728}") long bufferMemory,
                         @Value("${spring.kafka.producer.properties.max.request.size:8388608}") int maxRequestSize,
                         @Value("${syslog.kafka.partition-key-strategy:event_id}") String partitionKeyStrategy) {
        this.appProperties = appProperties;
        this.partitionKeyStrategy = PartitionKeyStrategy.parse(partitionKeyStrategy);
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, appProperties.getKafkaProducerBootstrap());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // [SYSLOG-ALARM][KPROD][已修复] acks=all 与幂等发送共同避免 Leader 切换和客户端重试造成假成功或重复。
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeoutMs);
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMs);
        props.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType);
        // Dispatcher 已限制全局在途数；Producer 缓冲耗尽时快速返回失败，让事件留在 WAL 重试。
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        applyMaxRequestSize(props, maxRequestSize);
        // [SYSLOG-ALARM][KPROD][已修复] linger、batch 与压缩恢复跨消息批量能力，吞吐优化不改变 Broker ACK 边界。
        // [SYSLOG-ALARM][KPROD][已修复] Producer 只在应用启动时构造一次，单条热路径不再创建网络线程和连接。
        this.producer = new KafkaProducer<>(props);
        log.info("Kafka Producer 已启用分区策略: strategy={}", this.partitionKeyStrategy.configValue);
    }

    static void applyMaxRequestSize(Properties props, int maxRequestSize) {
        if (maxRequestSize <= 0) {
            throw new IllegalArgumentException("spring.kafka.producer.properties.max.request.size 必须大于 0");
        }
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize);
    }

    /** 测试入口：复用真实的分区键选择和 ProducerRecord 构造逻辑，不建立 Kafka 网络连接。 */
    kafkaProducer(AppProperties appProperties, Producer<String, String> producer, String partitionKeyStrategy) {
        this.appProperties = appProperties;
        this.producer = producer;
        this.partitionKeyStrategy = PartitionKeyStrategy.parse(partitionKeyStrategy);
    }

    /**
     * [流程节点3] 为原始 Syslog 生成稳定 event_id，加密并异步投递 Kafka。
     * 返回 Future 只有在 Broker 按 acks=all 确认后成功完成，调用方必须据此更新接收审计。
     */
    public CompletableFuture<DeliveryResult> messagePush(DeviceDevice deviceDevice,
                                                          String syslog,
                                                          String receiveTime) {
        return messagePush(deviceDevice, syslog, receiveTime, newEventId());
    }

    public CompletableFuture<DeliveryResult> messagePush(DeviceDevice deviceDevice,
                                                          String syslog,
                                                          String receiveTime,
                                                          String eventId) {
        return messagePush(deviceDevice, syslog, receiveTime, eventId, null);
    }

    public CompletableFuture<DeliveryResult> messagePush(DeviceDevice deviceDevice,
                                                          String syslog,
                                                          String receiveTime,
                                                          String eventId,
                                                          Long pointTimeNs) {
        return messagePush(deviceDevice, syslog, receiveTime, receiveTime, eventId, pointTimeNs);
    }

    /**
     * 显式区分接收时间和事件时间。TCP/UDP 调用旧签名时两者相同，
     * 后续 API 拉取可以在不改 Kafka 头结构的前提下传入设备事件时间。
     */
    public CompletableFuture<DeliveryResult> messagePush(DeviceDevice deviceDevice,
                                                          String syslog,
                                                          String receiveTime,
                                                          String eventTime,
                                                          String eventId,
                                                          Long pointTimeNs) {
        // [流程节点3.1] event_id 在接入边界生成并写入加密载荷，Retry/Replay 不再重新生成。
        String fullMessage = DeviceInfoUtil.getFullLogString(
                deviceDevice, syslog, receiveTime, eventTime, eventId, pointTimeNs, appProperties);
        // [流程节点3.2] Kafka 中保存携带 event_id 的完整密文，原始正文不写普通应用日志。
        String encrypted = Sm4Util.encryptCbc(fullMessage, appProperties.getSm4Key());
        ProducerRecord<String, String> record = buildRecord(deviceDevice.getId(), eventId, encrypted);

        CompletableFuture<DeliveryResult> result = new CompletableFuture<>();
        // [SYSLOG-ALARM][KPROD][已修复] 回调成功才表示 Broker 接管；send() 返回本身不再记录 PRODUCED。
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                result.completeExceptionally(exception);
                return;
            }
            result.complete(new DeliveryResult(eventId, metadata));
        });
        return result;
    }

    ProducerRecord<String, String> buildRecord(Integer deviceId, String eventId, String encrypted) {
        String partitionKey = partitionKeyStrategy.partitionKey(deviceId, eventId);
        return new ProducerRecord<>(appProperties.getKafkaProducerTopic(), partitionKey, encrypted);
    }

    public static String newEventId() {
        return newEventId(Instant.now());
    }

    /** event_id 日期固定按接入业务时区计算，不依赖容器或 JVM 默认时区。 */
    public static String newEventId(Instant receivedAt) {
        return EVENT_ID_DATE.format(receivedAt.atZone(EVENT_ID_ZONE)) + "-" + UUID.randomUUID();
    }

    /** 仅用于运维探针的同步健康验证，不用于逐条热路径。 */
    public DeliveryResult sendAndAwait(DeviceDevice deviceDevice, String syslog, String receiveTime,
                                       long timeoutMs) throws Exception {
        return messagePush(deviceDevice, syslog, receiveTime).get(timeoutMs, TimeUnit.MILLISECONDS);
    }

    @PreDestroy
    public void close() {
        // [SYSLOG-ALARM][KPROD][已修复] 只在优雅停机阶段 flush/close，确保缓冲中的已接收消息完成发送。
        try {
            producer.flush();
        } finally {
            producer.close(java.time.Duration.ofSeconds(10));
        }
    }

    enum PartitionKeyStrategy {
        EVENT_ID("event_id") {
            @Override
            String partitionKey(Integer deviceId, String eventId) {
                if (eventId == null || eventId.trim().isEmpty()) {
                    throw new IllegalArgumentException("event_id 分区策略要求 eventId 非空");
                }
                return eventId;
            }
        },
        DEVICE_ID("device_id") {
            @Override
            String partitionKey(Integer deviceId, String eventId) {
                if (deviceId == null) {
                    throw new IllegalArgumentException("device_id 分区策略要求 deviceId 非空");
                }
                return deviceId.toString();
            }
        };

        private final String configValue;

        PartitionKeyStrategy(String configValue) {
            this.configValue = configValue;
        }

        abstract String partitionKey(Integer deviceId, String eventId);

        static PartitionKeyStrategy parse(String configuredValue) {
            String normalized = configuredValue == null
                    ? "" : configuredValue.trim().toLowerCase(Locale.ROOT);
            for (PartitionKeyStrategy strategy : values()) {
                if (strategy.configValue.equals(normalized)) return strategy;
            }
            throw new IllegalArgumentException(
                    "不支持的 Kafka 分区键策略: " + configuredValue + "，仅支持 event_id 或 device_id");
        }
    }

    @Getter
    public static class DeliveryResult {
        private final String eventId;
        private final String topic;
        private final int partition;
        private final long offset;

        DeliveryResult(String eventId, RecordMetadata metadata) {
            this.eventId = eventId;
            this.topic = metadata.topic();
            this.partition = metadata.partition();
            this.offset = metadata.offset();
        }
    }
}