package cn.gintone.config;

import cn.gintone.controller.vo.DeviceReqConfListReqVO;
import cn.gintone.controller.vo.DeviceReqConfSaveReqVO;
import cn.gintone.dal.DeviceReqConfMapper;
import cn.gintone.entity.DeviceReqConfDO;
import cn.gintone.service.DeviceReqConfService;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.concurrent.CompletableFuture;

@Component
public class PumpStationKafkaConsumer {

    @Autowired
    private DeviceReqConfMapper deviceReqConfMapper;

    private static final Logger logger = LoggerFactory.getLogger(PumpStationKafkaConsumer.class);

    /**
     * 主消费者 - 处理大部分topic
     */
    /*@KafkaListener(
            topics = {
                    "MKLDYL_JCSJ", "BZYYZ_JCSJ", "BZPSQ_JCSJ", "BZXZY_JCSJ",
                    "BZZLB_JCSJ", "BZYZJ_JCSJ", "BZJSZM_JCSJ", "BZYLZM_JCSJ", "BZCSZM_JCSJ",
                    "BZSCZM_JCSJ", "BZTXZM_JCSJ", "BZTFFJ_JCSJ", "BZSFFJ_JCSJ", "BZPFFJ_JCSJ",
                    "BZFJ_JCSJ", "BZFSJ_JCSJ", "BZYWCJ_JCSJ_2", "BZQBJ_JCSJ", "BZYDYHQT_JCSJ",
                    "BZZF_JCSJ", "BZQSB_JCSJ", "BZGSJ_JCSJ", "BZPDJ_JCSJ", "BZLDYWJ_JCSJ",
                    "BZYWCJ_JCSJ", "BZYWJ_JCSJ", "BZBSF_JCSJ", "BZFF_JCSJ", "BZDF_JCSJ",
                    "TYJG_JCSJ", "BZSB_JCSJ", "TYYLD_JCSJ", "TYYSGW_JCSJ", "YSGWLLJ_JCSJ",
                    "YSGWSWJ_JCSJ"
            },
            containerFactory = "kafkaListenerContainerFactory2",
            groupId = "pump_station_main_group"  // 改为不同的组
    )
    public void consumePumpStationData(String message,
                                       @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                       Acknowledgment ack) {
        try {
            logger.info("主消费者收到泵站数据, topic: {}, 消息长度: {}", topic, message);

            // 立即提交偏移量，避免阻塞
            ack.acknowledge();

            // 异步处理业务逻辑
            processMessageAsync(message, topic);

        } catch (Exception e) {
            logger.error("处理泵站数据时发生错误, topic: {}, 错误: {}", topic, e.getMessage(), e);
        }
    }*/

    /**
     * BZHSZM_JCSJ专用消费者
     */
    /*@KafkaListener(
            topics = {"BZHSZM_JCSJ"},
            containerFactory = "kafkaListenerContainerFactory2",
            groupId = "pump_station_bzhszm_group"  // 专用组
    )
    public void consumePumpStationData_BZHSZM_JCSJ(String message,
                                                   @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                                   Acknowledgment ack) {
        logger.info("BZHSZM_JCSJ专用消费者收到数据, topic: {}, 消息长度: {}", topic, message);
        ack.acknowledge();
        processMessageAsync(message, topic);
    }*/

    /**
     * BZPSZM_JCSJ专用消费者
     */
    /*@KafkaListener(
            topics = {"BZPSZM_JCSJ"},
            containerFactory = "kafkaListenerContainerFactory2",
            groupId = "pump_station_bzpszm_group"  // 专用组
    )
    public void consumePumpStationData_BZPSZM_JCSJ(String message,
                                                   @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                                   Acknowledgment ack) {
        logger.info("BZPSZM_JCSJ专用消费者收到数据, topic: {}, 消息长度: {}", topic, message);
        ack.acknowledge();
        processMessageAsync(message, topic);
    }*/

    /**
     * 异步处理消息，避免阻塞消费者
     */
    private void processMessageAsync(String message, String topic) {
        CompletableFuture.runAsync(() -> {
            try {
                JSONObject msgJsonObject = JSON.parseObject(message);
                String snCode = msgJsonObject.get("serialNo").toString();

                DeviceReqConfListReqVO temDRCLRV = new DeviceReqConfListReqVO();
                temDRCLRV.setSnCode(snCode);
                List<DeviceReqConfDO> deviceReqConfList = deviceReqConfMapper.selectList(temDRCLRV);

                if (null != deviceReqConfList && deviceReqConfList.size() > 0) {
                    for (DeviceReqConfDO drcd : deviceReqConfList) {
                        DeviceReqConfDO deviceReqConfSaveReqVO = new DeviceReqConfDO();
                        deviceReqConfSaveReqVO.setId(drcd.getId());
                        deviceReqConfSaveReqVO.setState(1);
                        deviceReqConfSaveReqVO.setSycnDate(new Date());
                        deviceReqConfMapper.update(new UpdateWrapper<DeviceReqConfDO>().lambda()
                                .eq(DeviceReqConfDO::getId, deviceReqConfSaveReqVO.getId())
                                .set(DeviceReqConfDO::getState, 1)
                                .set(DeviceReqConfDO::getSycnDate, deviceReqConfSaveReqVO.getSycnDate())
                        );
                    }
                    logger.info("异步处理完成, topic: {}, SN: {}", topic, snCode);
                }
            } catch (Exception e) {
                logger.error("异步处理消息失败, topic: {}, 错误: {}", topic, e.getMessage(), e);
            }
        });
    }
}