package cn.iocoder.yudao.module.mes.service.collect;

import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.module.mes.dal.dataobject.dv.machinery.MesDvMachineryDO;
import cn.iocoder.yudao.module.mes.dal.dataobject.pro.task.MesProTaskDO;
import cn.iocoder.yudao.module.mes.dal.mysql.dv.machinery.MesDvMachineryMapper;
import cn.iocoder.yudao.module.mes.enums.collect.MesCollectMsgTypeEnum;
import cn.iocoder.yudao.module.mes.framework.collect.config.MesCollectProperties;
import tools.jackson.databind.JsonNode;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * MES 数采报文分发器
 *
 * <p>Paho 回调线程只负责入队，解析与落库在单线程中串行执行，保证同一设备报警等状态按顺序处理。</p>
 *
 * @author 超级管理员
 */
@Component
@Slf4j
public class MesCollectMessageDispatcher {

    /**
     * 网关格式里承载设备属性快照的主题末段
     */
    private static final String ATTR_TOPIC_SUFFIX = "attributes";

    @Resource
    private MesCollectProperties properties;
    @Resource
    private MesCollectDataParser parser;
    @Resource
    private MesCollectDataService dataService;
    @Resource
    private MesDvMachineryMapper machineryMapper;
    @Resource
    private MesCollectTaskResolver taskResolver;

    private final ExecutorService executor = new ThreadPoolExecutor(1, 1,
            0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(10_000),
            new ThreadFactory() {
                private final AtomicInteger index = new AtomicInteger();

                @Override
                public Thread newThread(Runnable runnable) {
                    Thread thread = new Thread(runnable, "mes-collect-dispatch-" + index.incrementAndGet());
                    thread.setDaemon(true);
                    return thread;
                }
            }, new ThreadPoolExecutor.CallerRunsPolicy());

    /**
     * 设备未绑定台账时的告警缓存，避免每条报文重复刷日志
     */
    private final ConcurrentHashMap<String, Long> unboundWarnCache = new ConcurrentHashMap<>();

    public void dispatch(String topic, String payload) {
        executor.execute(() -> process(topic, payload));
    }

    private void process(String topic, String payload) {
        // 网关格式报文结构完全不同（主题固定、设备以报文首层 key 区分），单独处理
        if (properties.getMqtt().isGatewayFormat()) {
            processGateway(topic, payload);
            return;
        }
        JsonNode root;
        try {
            root = parser.parsePayload(payload);
        } catch (Exception e) {
            log.warn("[process][主题({}) 报文不是合法 JSON：{}]", topic, StrUtil.maxLength(payload, 200));
            dataService.markRawMessageFailed(
                    dataService.saveRawMessage(null, null, topic, MesCollectMsgTypeEnum.UNKNOWN,
                            LocalDateTime.now(), payload),
                    "报文不是合法 JSON：" + e.getMessage());
            return;
        }

        JsonNode data = parser.parseData(root);
        MesCollectMsgTypeEnum msgType = parser.resolveMsgType(parser.getText(root, "topic"), topic, data,
                properties.getMqtt().getAlarmTopicSuffix());
        String deviceCode = parser.resolveDeviceId(root, topic);
        LocalDateTime eventTime = parser.parseEventTime(root);

        MesDvMachineryDO machinery = StrUtil.isBlank(deviceCode) ? null
                : machineryMapper.selectByCollectDeviceCode(deviceCode);
        if (machinery == null && StrUtil.isNotBlank(deviceCode)) {
            warnUnboundDevice(deviceCode);
        }
        Long machineryId = machinery == null ? null : machinery.getId();
        Long taskId = null;
        Long workOrderId = null;
        if (machinery != null && machinery.getWorkstationId() != null) {
            MesProTaskDO task = taskResolver.resolve(machinery.getWorkstationId());
            if (task != null) {
                taskId = task.getId();
                workOrderId = task.getWorkOrderId();
            }
        }

        Long rawId = dataService.saveRawMessage(deviceCode, machineryId, topic, msgType, eventTime, payload);
        try {
            switch (msgType) {
                case REALTIME -> dataService.saveRealtime(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
                case SPC -> dataService.saveCycle(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
                case TECH -> dataService.saveTech(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
                case OP_LOG -> dataService.saveOperateLog(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
                case ALARM -> dataService.saveAlarm(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
                default -> log.debug("[process][忽略未知类型报文，主题({})]", topic);
            }
        } catch (Exception e) {
            dataService.markRawMessageFailed(rawId, e.getMessage());
            log.error("[process][主题({}) 设备({}) 处理失败]", topic, deviceCode, e);
        }
    }

    /**
     * 网关格式报文处理。
     *
     * <p>attributes：报文首层 key 是设备名、value 是属性字典，按设备逐台落实时数据；
     * connect / disconnect：连接事件，只留痕不落实时数据，便于排查设备离线。</p>
     */
    private void processGateway(String topic, String payload) {
        JsonNode root;
        try {
            root = parser.parsePayload(payload);
        } catch (Exception e) {
            log.warn("[processGateway][主题({}) 报文不是合法 JSON：{}]", topic, StrUtil.maxLength(payload, 200));
            dataService.markRawMessageFailed(
                    dataService.saveRawMessage(null, null, topic, MesCollectMsgTypeEnum.UNKNOWN,
                            LocalDateTime.now(), payload),
                    "报文不是合法 JSON：" + e.getMessage());
            return;
        }

        // 连接事件：{"device":"大拉"}，没有属性字典
        if (!ATTR_TOPIC_SUFFIX.equalsIgnoreCase(parser.extractTopicSuffix(topic))) {
            String deviceCode = StrUtil.trim(parser.getText(root, "device"));
            MesDvMachineryDO machinery = StrUtil.isBlank(deviceCode) ? null
                    : machineryMapper.selectByCollectDeviceCode(deviceCode);
            dataService.saveRawMessage(deviceCode, machinery == null ? null : machinery.getId(), topic,
                    MesCollectMsgTypeEnum.getByType(parser.extractTopicSuffix(topic)),
                    LocalDateTime.now(), payload);
            return;
        }

        Map<String, JsonNode> devices = parser.splitGatewayDevices(root);
        if (devices.isEmpty()) {
            log.debug("[processGateway][主题({}) 报文里没有设备属性：{}]", topic, StrUtil.maxLength(payload, 200));
            dataService.saveRawMessage(null, null, topic, MesCollectMsgTypeEnum.UNKNOWN, LocalDateTime.now(), payload);
            return;
        }
        // 网关报文不带时间戳，统一按接收时间落库
        LocalDateTime eventTime = parser.parseEventTime(root);
        devices.forEach((deviceCode, data) -> {
            MesDvMachineryDO machinery = machineryMapper.selectByCollectDeviceCode(deviceCode);
            if (machinery == null) {
                warnUnboundDevice(deviceCode);
            }
            Long machineryId = machinery == null ? null : machinery.getId();
            Long taskId = null;
            Long workOrderId = null;
            if (machinery != null && machinery.getWorkstationId() != null) {
                MesProTaskDO task = taskResolver.resolve(machinery.getWorkstationId());
                if (task != null) {
                    taskId = task.getId();
                    workOrderId = task.getWorkOrderId();
                }
            }
            Long rawId = dataService.saveRawMessage(deviceCode, machineryId, topic,
                    MesCollectMsgTypeEnum.REALTIME, eventTime, payload);
            try {
                dataService.saveRealtime(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
            } catch (Exception e) {
                dataService.markRawMessageFailed(rawId, e.getMessage());
                log.error("[processGateway][主题({}) 设备({}) 处理失败]", topic, deviceCode, e);
            }
        });
    }

    private void warnUnboundDevice(String deviceCode) {
        long now = System.currentTimeMillis();
        Long last = unboundWarnCache.get(deviceCode);
        if (last != null && now - last < 60_000L) {
            return;
        }
        unboundWarnCache.put(deviceCode, now);
        log.warn("[process][数采设备({}) 未绑定任何设备台账，数据将按设备编号落库，machineryId 为空]", deviceCode);
    }

    @PreDestroy
    public void destroy() {
        executor.shutdownNow();
    }

}