package com.xiaopeng.automaticPilot.util;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

@Slf4j
@Component
public class VehicleBuriedPointProducer {

    // 预发环境 Kafka 集群

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    // 目标 Topic
    private static final String TOPIC = "topic-robotaxi-vehicle-buriedPoint";

    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    private KafkaProducer<String, String> producer;

    @PostConstruct
    public void init() {
        Properties props = buildProperties();
        this.producer = new KafkaProducer<>(props);
    }

    @javax.annotation.PreDestroy
    public void destroy() {
        if (producer != null) {
            producer.close();
        }
    }

    /**
     * 发送埋点消息
     *
     * @param event
     * @throws Exception
     */
    public void sendVehicleEvent(VehicleEventDTO event){
        try {
            String messageBody = OBJECT_MAPPER.writeValueAsString(event);
            ProducerRecord<String, String> record =
                    new ProducerRecord<>(TOPIC, event.getVin(), messageBody);
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    log.error("埋点消息发送失败，vin={}, eventKey={}", event.getVin(), event.getEventKey(), exception);
                } else {
                    log.info("埋点消息发送成功，topic={}, partition={}, offset={}", metadata.topic(), metadata.partition(), metadata.offset());
                }
            });
        } catch (Exception e) {
            log.error("埋点消息序列化失败，vin={}", event.getVin(), e);
        }
    }

    /**
     * 构造埋点消息体
     *
     * @param vin
     * @param orderId
     * @param xpuId
     * @param eventKey
     * @param eventTime
     * @param extraAttributes
     * @return
     */
    public VehicleEventDTO buildVehicleEvent(String vin, String orderId, String xpuId,
                                          String eventKey, Long eventTime,
                                          Map<String, Object> extraAttributes) {
        VehicleEventDTO event = new VehicleEventDTO();
        event.setVin(vin);
        event.setXpuId(xpuId != null ? xpuId : "");
        event.setOrderId(orderId != null ? orderId : "");
        event.setEventKey(eventKey);
        event.setEventTime(eventTime != null ? eventTime : System.currentTimeMillis());

        Map<String, Object> attributes = new HashMap<>();
        attributes.put("vin", vin);
        attributes.put("order_id", orderId != null ? orderId : "");
        attributes.put("timestamp", event.getEventTime());

        if (extraAttributes != null) {
            attributes.putAll(extraAttributes);
        }
        event.setAttributes(attributes);
        return event;
    }

    private Properties buildProperties() {
        Properties props = new Properties();

        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 生产建议配置
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 关闭幂等性：当前 Kafka broker 版本过低（< 0.11，不支持 message format v2），
        // 开启幂等会抛 UnsupportedVersionException 导致埋点消息发送失败。
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);

        return props;
    }

//    public static void main(String[] args) throws Exception {
//
//        long timeMillis = System.currentTimeMillis();
//
//        // 构造埋点消息体（示例）
//        VehicleEvent event = new VehicleEvent();
//        event.setVin("L1NSPGHB7SB018421");
//        event.setXpuId("");
//        event.setOrderId("");
//        event.setEventKey("RT_web_vehicle_status_changed");
//        event.setEventTime(timeMillis);
//
//        Map<String, Object> attributes = new HashMap<>();
//        attributes.put("vin", "L1NSPGHB7SB018421");
//        attributes.put("order_id", "");
//        attributes.put("timestamp", timeMillis);
//        attributes.put("sub_status", "");
//        attributes.put("prev_sub_status", "");
//        event.setAttributes(attributes);
//
//        // 转 JSON
//        String messageBody = OBJECT_MAPPER.writeValueAsString(event);
//
//        // 建议用 vin 作为 key（保证同车有序）
//        ProducerRecord<String, String> record =
//                new ProducerRecord<>(TOPIC, event.getVin(), messageBody);
//
//        try {
//            Future<RecordMetadata> future = producer.send(record);
//            RecordMetadata metadata = future.get();
//
//            System.out.println("发送成功：");
//            System.out.println("Topic: " + metadata.topic());
//            System.out.println("Partition: " + metadata.partition());
//            System.out.println("Offset: " + metadata.offset());
//
//        } catch (Exception e) {
//            e.printStackTrace();
//        } finally {
//            producer.close();
//        }
//    }
}