package com.yeejoin.amos.boot.module.jg.biz.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.yeejoin.amos.boot.biz.common.utils.RedisUtils;
import com.yeejoin.amos.boot.module.common.api.dao.ESEquipmentCategory;
import com.yeejoin.amos.boot.module.jg.api.entity.SafetyProblemTracing;
import com.yeejoin.amos.boot.module.jg.api.enums.SafetyProblemSourceTypeEnum;
import com.yeejoin.amos.boot.module.jg.api.enums.SafetyProblemStatusEnum;
import com.yeejoin.amos.boot.module.jg.api.enums.SafetyProblemTypeEnum;
import com.yeejoin.amos.boot.module.jg.api.event.SafetyProblemEvent;
import com.yeejoin.amos.boot.module.jg.api.event.handler.SafetyProblemEventHandler;
import com.yeejoin.amos.boot.module.jg.api.mapper.SafetyProblemTracingMapper;
import com.yeejoin.amos.boot.module.jg.biz.handler.SafetyProblemEventHandlerFactory;
import com.yeejoin.amos.boot.module.jg.biz.service.impl.IdxBizJgOtherInfoServiceImpl;
import com.yeejoin.amos.boot.module.jg.biz.service.impl.SafetyProblemTracingServiceImpl;
import com.yeejoin.amos.boot.module.jg.biz.service.impl.StartPlatformTokenService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.typroject.tyboot.component.emq.EmqKeeper;
import org.typroject.tyboot.component.emq.EmqxListener;
import org.typroject.tyboot.core.foundation.utils.ValidationUtil;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

@Component
@Slf4j
public class SafetyProblemTopicMessage extends EmqxListener {

    @Value("${safety.problem.message.handle.thread.number:3}")
    private int threadNumber;

    @Value("${spring.application.name}")
    private String applicationName;

    @Resource
    protected EmqKeeper emqKeeper;

    @Value("${safety.problem.tracing.topic:safetyProblemTracing/#}")
    private String problemEventTopic;

    @Autowired
    RedisUtils redisUtil;

    @Autowired
    StartPlatformTokenService platformTokenService;

    @Autowired
    SafetyProblemTracingMapper safetyProblemTracingMapper;

    @Autowired
    IdxBizJgOtherInfoServiceImpl idxBizJgOtherInfoService;

    @Autowired
    ESEquipmentCategory esEquipmentCategory;

    private final BlockingQueue<SafetyProblemEvent> blockingQueue = new LinkedBlockingQueue<>();

    private ExecutorService executorService;

    @PostConstruct
    void init() throws Exception {
        emqKeeper.subscript(buildTopic(problemEventTopic), 2, this);
        executorService = Executors.newFixedThreadPool(threadNumber);
        for (int i = 0; i < threadNumber; i++) {
            executorService.execute(() -> {
                try {
                    while (true) {
                        SafetyProblemEvent safetyProblemMessageEvent = blockingQueue.take();
                        JSONObject jsonObject = JSON.parseObject(safetyProblemMessageEvent.getMessage().toString());
                        log.info("接收到问题生产消息:{}", jsonObject);
                        SafetyProblemEventHandler eventHandler = SafetyProblemEventHandlerFactory.createProblemHandler(safetyProblemMessageEvent.getTopic());
                        eventHandler.handle(safetyProblemMessageEvent);
                        log.info("处理问题生产消息完成");
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("处理问题生产消息异常", e);
                    throw new RuntimeException(e);
                }
            });
        }
    }

    @Override
    public void processMessage(String topic, MqttMessage message) {
        log.info("接收问题生产消息开始");
        blockingQueue.add(new SafetyProblemEvent(topic, message));
        log.info("接收问题生产消息完成");
    }

    public static void generateProblem(JSONObject jsonObject, SafetyProblemTypeEnum problemTypeEnum, SafetyProblemTracingServiceImpl safetyProblemTracingService) {
        if (jsonObject == null || problemTypeEnum == null) {
            throw new IllegalArgumentException("jsonObject and problemTypeEnum must not be null.");
        }
        JSONArray records = jsonObject.getJSONArray(problemTypeEnum.getMsgKey());
        if (!ValidationUtil.isEmpty(records)) {
            List<SafetyProblemTracing> safetyProblemTracingList = records.stream().map(item -> {
                if (!(item instanceof JSONObject)) {
                    throw new IllegalArgumentException("item is not a JSONObject.");
                }
                JSONObject json = (JSONObject) item;
                SafetyProblemTracing safetyProblemTracing = new SafetyProblemTracing();
//            safetyProblemTracing.setProblemNum(json.getString("problemNum"));
                safetyProblemTracing.setProblemType(problemTypeEnum.getName());
                safetyProblemTracing.setProblemTypeCode(problemTypeEnum.getCode());
                safetyProblemTracing.setProblemDesc(problemTypeEnum.getDesc());
//            safetyProblemTracing.setProblemLevel(json.getString("problemLevel"));
//            safetyProblemTracing.setProblemLevelCode(json.getString("problemLevelCode"));
                safetyProblemTracing.setSourceType(SafetyProblemSourceTypeEnum.EQUIP.getName());
                safetyProblemTracing.setSourceTypeCode(SafetyProblemSourceTypeEnum.EQUIP.getCode());
                safetyProblemTracing.setSourceId(json.getOrDefault("RECORD", "").toString());
                safetyProblemTracing.setProblemTime(new Date());
                safetyProblemTracing.setEquipSuperviseCode(json.getOrDefault("SUPERVISORY_CODE", "").toString());
                safetyProblemTracing.setEquipList(json.getOrDefault("equList", "").toString());
                safetyProblemTracing.setEquipListCode(json.getOrDefault("EQU_LIST", "").toString());
                safetyProblemTracing.setPrincipalUnit(json.getOrDefault("USE_UNIT_NAME", "").toString());
                safetyProblemTracing.setPrincipalUnitCode(json.getOrDefault("USE_UNIT_CREDIT_CODE", "").toString());
                safetyProblemTracing.setPrincipalUnitType(json.getOrDefault("unitType", "").toString());
                safetyProblemTracing.setGoverningBody(json.getOrDefault("ORG_BRANCH_NAME", "").toString());
                safetyProblemTracing.setGoverningBodyCode(json.getOrDefault("COMPANY_ORG_BRANCH_CODE", "").toString());
                safetyProblemTracing.setGoverningBodyOrgCode(json.getOrDefault("ORG_BRANCH_CODE", "").toString());
//            safetyProblemTracing.setRegionName(json.getString("regionName"));
                safetyProblemTracing.setRegionCode(json.getString("regionCode"));
                safetyProblemTracing.setCreateDate(new Date());
//            safetyProblemTracing.setCreateUser(json.getString("createUser"));
//            safetyProblemTracing.setCreateUserId(json.getString("createUserId"));
                safetyProblemTracing.setProblemStatus(SafetyProblemStatusEnum.UNHANDLED.getName());
                safetyProblemTracing.setProblemStatusCode(SafetyProblemStatusEnum.UNHANDLED.getCode());
                return safetyProblemTracing;
            }).collect(Collectors.toList());
            safetyProblemTracingService.saveOrUpdateBatchByColumns(safetyProblemTracingList,
                    safetyProblemTracing -> new QueryWrapper<>()
                            .eq("problem_type_code", safetyProblemTracing.getProblemTypeCode())
                            .eq("source_id", safetyProblemTracing.getSourceId())
                            .eq("problem_status_code", SafetyProblemStatusEnum.UNHANDLED.getCode()));}
    }

    public String buildTopic(String topic) {
        String topicPrefix = "$share/" + applicationName;
        return String.format("%s/%s", topicPrefix, topic);
    }

    @PreDestroy
    public void shutdown() {
        if (executorService != null) {
            executorService.shutdown();
        }
    }
}
