package com.ruoyi.other.udan.util;

import com.google.protobuf.CodedInputStream;
import com.google.protobuf.WireFormat;
import com.ruoyi.other.udan.config.UdanEmsKafkaProperties;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.Duration;
import java.util.Base64;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.io.IOException;

/**
 * 优旦 EMS 平台 Kafka 消费工具。
 *
 * <p>按照《UDAN-EMS云平台对平台协议》订阅设备监控 Topic：消息 Key 为 EMS 系统 ID，
 * 消息 Value 为 protobuf3 二进制数据，不是 JSON。业务方应注册消息处理器，并使用
 * 客户最终确认版本的 iems-open proto 生成类反序列化消息。</p>
 */
@Component
@ConditionalOnProperty(prefix = "udan.ems.kafka", name = "enabled", havingValue = "true")
public class UdanEmsKafkaUtil {
    private static final Logger log = LoggerFactory.getLogger(UdanEmsKafkaUtil.class);
    private final UdanEmsKafkaProperties properties;

    private final ExecutorService executor = Executors.newSingleThreadExecutor(r -> {
        Thread t = new Thread(r, "udan-ems-kafka-consumer");
        t.setDaemon(true);
        return t;
    });
    private volatile KafkaConsumer<String, byte[]> consumer;
    private volatile boolean running;
    private volatile MessageHandler messageHandler = (key, payload) -> { };
    private volatile ErrorHandler errorHandler = Throwable::printStackTrace;

    /**
     * 创建优旦 EMS Kafka 消费工具。
     *
     * @param properties Kafka 连接、认证、Topic 和消费组配置
     */
    public UdanEmsKafkaUtil(UdanEmsKafkaProperties properties) {
        this.properties = properties;
    }

    /**
     * 校验配置并在独立守护线程中启动持续消费。
     * 当 udan.ems.kafka.enabled=true 时由 Spring 在 Bean 初始化后自动调用。
     * PostConstruct 创建并完成依赖注入后，自动执行这个方法一次
     */
    //@PostConstruct
    public void start() {
        validate();
        running = true;
        executor.submit(this::consumeLoop);
    }

    /**
     * 注册设备监控消息处理器。
     *
     * @param handler 接收 EMS 系统 ID 和 protobuf 原始字节的处理器；传入 null 时忽略消息
     */
    public void setMessageHandler(MessageHandler handler) {
        this.messageHandler = handler == null ? (key, payload) -> { } : handler;
    }

    /**
     * 注册消费或业务处理异常的处理器。
     *
     * @param handler 异常处理器；传入 null 时恢复为打印异常堆栈
     */
    public void setErrorHandler(ErrorHandler handler) {
        this.errorHandler = handler == null ? Throwable::printStackTrace : handler;
    }

    /**
     * 按配置的超时时间轮询一次 Kafka，并将取得的消息交给消息处理器。
     *
     * @return 本次轮询取得的消息数量
     * @throws IllegalStateException 消费者尚未启动时抛出
     */
    public int pollOnce() {
        if (consumer == null) {
            throw new IllegalStateException("Kafka consumer has not started");
        }
        ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(Math.max(1, properties.getPollSeconds())));
        dispatch(records);
        return records.count();
    }

    /**
     * 唤醒并关闭 Kafka 消费者，同时停止后台消费线程。
     * 由 Spring 在 Bean 销毁前自动调用。
     */
    @PreDestroy
    public void stop() {
        running = false;
        if (consumer != null) consumer.wakeup();
        executor.shutdownNow();
    }

    /**
     * 创建消费者、订阅配置的设备监控 Topic，并持续执行轮询。
     * WakeupException 在正常关闭时忽略，运行期间出现的异常交给异常处理器。
     */
    private void consumeLoop() {
        try (KafkaConsumer<String, byte[]> c = new KafkaConsumer<>(consumerConfig())) {
            consumer = c;
            c.subscribe(Collections.singletonList(properties.getTopic()));
            while (running) pollOnce();
        } catch (WakeupException ignored) {
            if (running) errorHandler.onError(ignored);
        } catch (Throwable e) {
            if (running) errorHandler.onError(e);
        } finally {
            consumer = null;
        }
    }

    /**
     * 逐条分发 Kafka 消息；单条消息处理失败不会阻断同批次后续消息。
     *
     * @param records Kafka 本次轮询返回的消息集合
     */
    private void dispatch(ConsumerRecords<String, byte[]> records) {
        for (ConsumerRecord<String, byte[]> record : records) {
            byte[] payload = record.value() == null ? new byte[0] : record.value();
            try {
                // 先输出可核验信息，再交给业务回调；线上 Value 是 protobuf 原始字节，不能当 JSON 打印。
                log.info("1、UDAN EMS可读字段: emsSystemId={}, data={}", record.key(), toReadableProtobuf(payload));

                // 此处即为业务接入点：解析 protobuf 后调用保存服务，当前仅由回调方决定如何处理。
                messageHandler.onMessage(record.key(), payload);
            } catch (Throwable e) {
                errorHandler.onError(e);
            }
        }
    }

    /**
     * 将消息前若干字节转为十六进制，便于核对消息是否为稳定的二进制 protobuf 内容。
     *
     * @param bytes 原始消息字节
     * @param maxBytes 最多输出的字节数
     * @return 大写十六进制字符串，截断时追加省略号
     */
    private static String toHex(byte[] bytes, int maxBytes) {
        StringBuilder result = new StringBuilder();
        int length = Math.min(bytes.length, Math.max(0, maxBytes));
        for (int i = 0; i < length; i++) {
            if (i > 0) result.append(' ');
            result.append(String.format("%02X", bytes[i] & 0xFF));
        }
        if (bytes.length > length) result.append(" ...");
        return result.toString();
    }

    /**
     * 将 protobuf 二进制按 wire type 转为可读文本。由于未绑定生成的 proto 类，字段名显示为 fieldN；
     * 字符串、整数、浮点数和嵌套消息内容会被直接展开，便于联调确认实际收到的数据。
     */
    private static String toReadableProtobuf(byte[] bytes) {
        if (bytes == null || bytes.length == 0) return "{}";
        try {
            return readReadableFields(CodedInputStream.newInstance(bytes), 0);
        } catch (IOException e) {
            return "<protobuf解析失败: " + e.getMessage() + ">";
        }
    }

    /** 递归读取 protobuf 字段，限制递归深度和字段数量避免日志过大。 */
    private static String readReadableFields(CodedInputStream input, int depth) throws IOException {
        if (depth > 3) return "<nested>";
        StringBuilder text = new StringBuilder("{");
        int count = 0;
        while (!input.isAtEnd() && count++ < 80) {
            int tag = input.readTag();
            if (tag == 0) break;
            int field = WireFormat.getTagFieldNumber(tag);
            int type = WireFormat.getTagWireType(tag);
            if (text.length() > 1) text.append(", ");
            text.append("field").append(field).append('=');
            switch (type) {
                case WireFormat.WIRETYPE_VARINT:
                    text.append(input.readInt64());
                    break;
                case WireFormat.WIRETYPE_FIXED64:
                    text.append(input.readDouble());
                    break;
                case WireFormat.WIRETYPE_LENGTH_DELIMITED:
                    byte[] data = input.readBytes().toByteArray();
                    if (isReadableString(data)) {
                        text.append('"').append(escape(new String(data, java.nio.charset.StandardCharsets.UTF_8))).append('"');
                    } else {
                        text.append(readNestedOrHex(data, depth));
                    }
                    break;
                case WireFormat.WIRETYPE_FIXED32:
                    text.append(input.readFloat());
                    break;
                default:
                    text.append("<wireType=").append(type).append('>');
                    return text.append('}').toString();
            }
        }
        if (count > 80) text.append(", ...");
        return text.append('}').toString();
    }

    /** 尝试将长度字段识别为嵌套 protobuf，否则显示前 32 字节十六进制。 */
    private static String readNestedOrHex(byte[] data, int depth) {
        try {
            String nested = readReadableFields(CodedInputStream.newInstance(data), depth + 1);
            if (!nested.equals("{}")) return nested;
        } catch (IOException ignored) {
            // 不是嵌套 protobuf 时按原始二进制展示。
        }
        return "0x" + toHex(data, 32);
    }

    /** 判断长度字段是否是可直接阅读的 UTF-8 文本。 */
    private static boolean isReadableString(byte[] data) {
        if (data.length == 0) return true;
        for (byte value : data) {
            int c = value & 0xFF;
            if (c < 0x20 || c > 0x7E) return false;
        }
        return true;
    }

    /** 转义日志中的双引号和反斜杠。 */
    private static String escape(String value) {
        return value.replace("\\", "\\\\").replace("\"", "\\\"");
    }

    /**
     * 按协议示例构造 Kafka Consumer 配置。
     * 固定使用 SASL_SSL、SCRAM-SHA-512、字符串 Key 和字节数组 Value。
     *
     * @return 可直接用于创建 KafkaConsumer 的配置项
     */
    private Map<String, Object> consumerConfig() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
        config.put(ConsumerConfig.GROUP_ID_CONFIG, properties.getGroupId());
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, properties.getAutoOffsetReset());
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
        config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        config.put("sasl.mechanism", "SCRAM-SHA-512");
        config.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\""
                + properties.getUsername() + "\" password=\"" + properties.getPassword() + "\";");
        return config;
    }

    /**
     * 解码协议文档第 4.2 节提供的 Base64 测试报文。
     * 线上 Kafka Value 已经是 protobuf byte[]，不得调用此方法二次解码。
     *
     * @param base64 文档中的 Base64 报文，可以包含换行
     * @return protobuf 原始字节；输入为空时返回空数组
     */
    public static byte[] decodeDocumentSample(String base64) {
        if (!StringUtils.hasText(base64)) {
            return new byte[0];
        }
        return Base64.getMimeDecoder().decode(base64);
    }

    /**
     * 校验连接和订阅所需的必填参数，防止以不完整配置启动消费者。
     *
     * @throws IllegalStateException 缺少 Kafka 地址、用户名、密码或 Topic 时抛出
     */
    private void validate() {
        if (!StringUtils.hasText(properties.getBootstrapServers()) || !StringUtils.hasText(properties.getUsername())
                || !StringUtils.hasText(properties.getPassword()) || !StringUtils.hasText(properties.getTopic())) {
            throw new IllegalStateException("udan.ems.kafka requires bootstrap-servers, username, password and topic");
        }
    }

    /** 业务消息处理回调。 */
    @FunctionalInterface
    public interface MessageHandler {
        /**
         * 处理一条优旦 EMS 设备监控消息。
         *
         * @param emsSystemId Kafka Key，即 EMS 系统 ID
         * @param protobufBytes Kafka Value，即 protobuf3 原始字节
         */
        void onMessage(String emsSystemId, byte[] protobufBytes);
    }

    /** 消费和消息处理异常回调。 */
    @FunctionalInterface
    public interface ErrorHandler {
        /**
         * 处理 Kafka 消费或业务消息处理过程中发生的异常。
         *
         * @param error 捕获到的异常
         */
        void onError(Throwable error);
    }
}