package cn.axzo.workflow.server.xxljob;

import cn.axzo.karma.client.feign.FlowSupportApi;
import cn.axzo.karma.client.model.request.PersonProfileQueryReq;
import cn.axzo.karma.client.model.response.PersonProfileResp;
import cn.axzo.workflow.common.model.dto.es.DataSyncSummaryDTO;
import cn.axzo.workflow.common.model.dto.es.HistoricProcessInstanceSearchForEsDTO;
import cn.axzo.workflow.common.model.request.bpmn.task.BpmnTaskDelegateAssigner;
import cn.axzo.workflow.core.conf.SupportRefreshProperties;
import cn.axzo.workflow.core.engine.cmd.CustomInsertPropertyCmd;
import cn.axzo.workflow.core.service.BpmnProcessInstanceForEsService;
import cn.axzo.workflow.es.model.ProcessTaskDocument;
import cn.axzo.workflow.es.service.aggregation.AggregateProcessInstanceService;
import cn.axzo.workflow.server.controller.listener.task.SyncToEsTaskEvent_104_Listener;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.NumberUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.flowable.common.engine.impl.interceptor.CommandExecutor;
import org.flowable.engine.history.HistoricProcessInstance;
import org.flowable.spring.SpringProcessEngineConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static cn.axzo.workflow.common.constant.BpmnConstants.LATEST_SYNC_TO_ELASTICSEARCH_TIME;
import static cn.axzo.workflow.server.controller.delegate.AbstractBpmnTaskAssigneeSelector.parseApiResult;

/**
 * 审批结束的流程实例数据同步至 ES
 * <p>
 * 在途的，或新发起的流程实例，均通过 {@link SyncToEsTaskEvent_104_Listener} 进行实时同步,
 * 但由于用户的操作时间不可控,所以统一由 {@link OtherProcessInstanceSyncEsJobHandler} 单独进行一次同步
 *
 * @author wangli
 * @since 2024-09-27 11:03
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class CompletedProcessInstanceSyncEsJobHandler {
    private final BpmnProcessInstanceForEsService bpmnProcessInstanceForEsService;
    private final AggregateProcessInstanceService aggregateProcessInstanceService;
    private final SpringProcessEngineConfiguration springProcessEngineConfiguration;
    private final ApplicationContext context;
    private final SupportRefreshProperties refreshProperties;
    private final FlowSupportApi flowSupportApi;

    @XxlJob("completedProcessInstanceSyncToEs")
    public ReturnT<String> execute(String s) {
        // 分片参数
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();
        log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
        XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);

        log.debug("start exec finished process instance data sync... ");
        XxlJobHelper.log("start exec finished process instance data sync... ");

        HistoricProcessInstanceSearchForEsDTO search = new HistoricProcessInstanceSearchForEsDTO();
        search.setFinished(true);
        Date endTime = new Date();
        search.setEndTime(endTime);

        try {
            if (StringUtils.hasText(s)) {
                search = JSON.parseObject(s, HistoricProcessInstanceSearchForEsDTO.class);
                log.info("根据入参转换后的查询入参：{}", JSON.toJSONString(search));
                XxlJobHelper.log("根据入参转换后的查询入参：{}", JSON.toJSONString(search));
            } else {
                log.info("入参为空, 将以默认条件执行");
                XxlJobHelper.log("入参为空, 将以默认条件执行");
            }
        } catch (Exception e) {
            log.warn("无法解析传参, 入参必须是 JSON 格式, 入参可参考 HistoricProcessInstanceSearchForEsDTO 模型");
            XxlJobHelper.log("无法解析传参, 入参必须是 JSON 格式, 入参可参考 HistoricProcessInstanceSearchForEsDTO 模型");
            return ReturnT.FAIL;
        }

        // 开始同步完成的实例到 ES
        DataSyncSummaryDTO summary = doSync(search, shardIndex, shardTotal);

        // 执行完同步后的一些额外操作
        afterSync(endTime);

        log.info("Insert Summary: ProcessInstance Count: {}, ProcessTask Count: {}", summary.getProcessInstanceCount(), summary.getProcessTaskCount());
        XxlJobHelper.log("Insert Summary: ProcessInstance Count: {}, ProcessTask Count: {}", summary.getProcessInstanceCount(), summary.getProcessTaskCount());
        return ReturnT.SUCCESS;
    }

    /**
     * 1. 记录本次操作的时间点
     */
    private void afterSync(Date endTime) {
        //1. 记录本次操作的时间点, 多次执行也会自动更新操作时间点
        CommandExecutor commandExecutor = springProcessEngineConfiguration.getCommandExecutor();
        commandExecutor.execute(new CustomInsertPropertyCmd(LATEST_SYNC_TO_ELASTICSEARCH_TIME, DateUtil.formatDateTime(endTime)));
    }

    /**
     * 同步指定查询结果的数据至 ES
     * <p>
     * 内部自动分页循环同步
     *
     * @param search
     * @return 应同步至 ES 的统计数据
     */
    private DataSyncSummaryDTO doSync(HistoricProcessInstanceSearchForEsDTO search, int shardIndex, int shardTotal) {
        Long totalProcessInstanceCount = bpmnProcessInstanceForEsService.queryHistoricProcessInstanceTotalCount(search);
        log.info("查询到待同步实例维度数据量: {} 条", totalProcessInstanceCount);
        XxlJobHelper.log("查询到待同步实例维度数据量: {} 条", totalProcessInstanceCount);
        AtomicLong totalProcessTaskCount = new AtomicLong(0);
        IntStream.iterate(0, i -> i + search.getOverPageSize()).limit((totalProcessInstanceCount + search.getOverPageSize() - 1) / search.getOverPageSize())
                .forEach(skipRows -> {
                    log.info("处理进度: {} %, current startRow: {}, endRow: {}", String.format("%.2f", (double) skipRows / totalProcessInstanceCount * 100),
                            skipRows, skipRows + search.getOverPageSize());
                    XxlJobHelper.log("处理进度: {} %, current startRow: {}, endRow: {}", String.format("%.2f", (double) skipRows / totalProcessInstanceCount * 100),
                            skipRows, skipRows + search.getOverPageSize());
                    int pageNo = skipRows / search.getOverPageSize() + 1;
                    List<HistoricProcessInstance> instances = bpmnProcessInstanceForEsService.queryHistoricProcessInstance(search, new Page<HistoricProcessInstance>(pageNo, search.getOverPageSize()));

                    instances.parallelStream().filter(hpi -> {
                        if (shardIndex >= 0 && shardTotal >= 0) {
                            if (hpi.getId().contains("-")) {
                                // 最开始的实例编号是 UUID，无法取模，所以统一交给第一个节点处理
                                return shardIndex == 0;
                            }
                            return Long.parseLong(hpi.getId().substring(12)) % shardTotal == shardIndex;
                        }
                        // 如果配置的非分片广播的方式，那么 shardingVO 将是 null
                        return true;
                    }).forEach(hpi -> {
                        List<ProcessTaskDocument> processTaskDocuments = aggregateProcessInstanceService.syncProcessInstance(hpi, this::queryPersonByIds);
                        totalProcessTaskCount.getAndSet(totalProcessTaskCount.get() + processTaskDocuments.size());
                    });
                });
        return new DataSyncSummaryDTO(totalProcessInstanceCount, totalProcessTaskCount.get());
    }

    private Map<Long, BpmnTaskDelegateAssigner> queryPersonByIds(List<BpmnTaskDelegateAssigner> users) {
        Map<Long, BpmnTaskDelegateAssigner> result = new HashMap<>();
        if (CollectionUtils.isEmpty(users)) {
            return result;
        }
        Map<Long, BpmnTaskDelegateAssigner> map = users.stream()
                .filter(i -> StringUtils.hasText(i.getPersonId()) && !Objects.equals("null", i.getPersonId()) && NumberUtil.isNumber(i.getPersonId()))
                .collect(Collectors.toMap(i -> NumberUtil.parseLong(i.getPersonId()), Function.identity(), (s, t) -> s));

        PersonProfileQueryReq query = new PersonProfileQueryReq();
        ArrayList<Long> personIds = Lists.newArrayList(map.keySet());
        query.setPersonIds(personIds);
        Map<Long, PersonProfileResp> personProfileMap = parseApiResult(() -> flowSupportApi.listPersons(PersonProfileQueryReq.builder().personIds(personIds).build()),
                "根据 PersonId 查询自然人档案", "cn.axzo.karma.client.feign.FlowSupportApi.listPersons", refreshProperties, context, personIds)
                .stream().collect(Collectors.toMap(PersonProfileResp::getId, Function.identity(), (s, t) -> s));

        personProfileMap.forEach((k, v) -> {
            result.put(k, BpmnTaskDelegateAssigner.builder()
                    .personId(String.valueOf(v.getId()))
                    .assignerName(v.getRealName())
                    .avatar(v.getAvatarUrl())
                    .build());
        });
        return result;
    }


}
