package cn.axzo.workflow.server.mq.inside.consumer;

import cn.axzo.framework.domain.data.IdHelper;
import cn.axzo.framework.rocketmq.Event;
import cn.axzo.framework.rocketmq.utils.TraceUtils;
import cn.axzo.workflow.core.conf.SupportRefreshProperties;
import cn.axzo.workflow.es.model.ProcessTaskDocument;
import cn.axzo.workflow.es.service.aggregation.AggregateProcessInstanceService;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Strings;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.flowable.engine.HistoryService;
import org.flowable.engine.history.HistoricProcessInstance;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.*;

import static cn.axzo.workflow.common.enums.ElasticSearchEventEnum.ELASTIC_SEARCH_SYNC;

/**
 * 批量同步 ES
 *
 * @author wangli
 * @since 2024-11-20 18:06
 */
@Component
@Slf4j
public class ElasticSearchBatchSyncListener {
    @Resource
    private SupportRefreshProperties refreshProperties;
    @Resource
    private RocketMQProperties rocketMQProperties;
    @Value("GID_${spring.application.name}_workflow_engine_${spring.profiles.active}_consumer")
    private String consumerGroup;
    @Value("topic_workflow_engine_${spring.profiles.active}")
    private String topic;
    @Resource
    private HistoryService historyService;
    @Resource
    private AggregateProcessInstanceService aggregateProcessInstanceService;

    @SneakyThrows
    @PostConstruct
    public void init() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe(topic, ELASTIC_SEARCH_SYNC.getTag());
        consumer.setConsumeMessageBatchMaxSize(refreshProperties.getEsSyncBatchSize());
//        consumer.setConsumeThreadMax(1);
//        consumer.setConsumeThreadMin(1);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            log.info("batch get msg size: {}", msgs.size());
            if(CollectionUtils.isEmpty(msgs)) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            try {
                HashSet<String> idSet = new HashSet<>();
                for (MessageExt msg : msgs) {
                    String msgBody = new String(msg.getBody(), StandardCharsets.UTF_8);
                    Event event = convertEvent(msg, msgBody);
                    if (support(event)) {
                        String processInstanceId = event.normalizedData(String.class);
                        idSet.add(processInstanceId);
                    }
                }
                log.info("deduplication ids size: {}", idSet.size());
                idSet.forEach(this::sync);
            } catch (Exception e) {
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }

    private Boolean support(Event event) {
        if (Objects.isNull(event)) {
            return false;
        }
        return Objects.equals(event.getEventModule(), ELASTIC_SEARCH_SYNC.getModule())
                && Objects.equals(event.getEventName(), ELASTIC_SEARCH_SYNC.getTag());
    }

    private Event convertEvent(MessageExt msg, String msgBody) {
        try {
            return JSONObject.parseObject(msgBody, Event.class);
        } catch (Exception e) {
            log.error("====MQ CONSUMER {} ===={}, msgId = {}, parse event error, event = {}",
                    getTraceId(msg), msg.getMsgId(), msgBody, e.getMessage(), e);
            throw e;
        }
    }

    public String getTraceId(MessageExt msg) {
        Map<String, String> headers = msg.getProperties();
        MDC.put(TraceUtils.CTX_LOG_ID, headers.get(TraceUtils.TRACE_ID));
        MDC.put(TraceUtils.TRACE_ID, headers.get(TraceUtils.TRACE_ID));
        MDC.put(TraceUtils.TRACE_ID_IN_MDC, headers.get(TraceUtils.TRACE_ID));
        if (log.isDebugEnabled()) {
            log.debug("received message, topic={}, headers={}", topic, headers);
        }
        String traceId = null;
        if (!CollectionUtils.isEmpty(headers) && headers.containsKey(TraceUtils.TRACE_ID)) {
            traceId = headers.get(TraceUtils.TRACE_ID);
        }
        if (Strings.isNullOrEmpty(traceId)) {
            traceId = IdHelper.get32UUID();
        }
        TraceUtils.putTraceId(traceId);
        return traceId;
    }

    private void sync(String processInstanceId) {
        HistoricProcessInstance historicProcessInstance = historyService.createHistoricProcessInstanceQuery()
                .processInstanceId(processInstanceId)
                .singleResult();
        // 删除指定实例的父子文档
        log.info("delete document processInstanceId: {}", processInstanceId);
        aggregateProcessInstanceService.deleteDocumentParentAndChild(processInstanceId);
        log.info("reInsert document processInstanceId: {}", processInstanceId);
        List<ProcessTaskDocument> processTaskDocuments = aggregateProcessInstanceService.syncProcessInstance(historicProcessInstance, null);
        log.info("Insert Summary: ProcessInstance Count: {}, ProcessTask Count: {}", 1, processTaskDocuments.size());
    }
}
