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.request.bpmn.task.BpmnTaskDelegateAssigner;
import cn.axzo.workflow.core.conf.SupportRefreshProperties;
import cn.axzo.workflow.core.engine.cmd.CustomGetPropertyCmd;
import cn.axzo.workflow.core.service.BpmnProcessInstanceForEsService;
import cn.axzo.workflow.es.model.ProcessTaskDocument;
import cn.axzo.workflow.es.service.aggregation.AggregateProcessInstanceService;
import cn.hutool.core.util.NumberUtil;
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.AtomicInteger;
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
 *
 * @author wangli
 * @since 2024-09-27 11:03
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class OtherProcessInstanceSyncEsJobHandler {
    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("otherProcessInstanceSyncToEs")
    public ReturnT<String> execute(String param) {
        // 分片参数
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();
        log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
        XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);


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

        // 同步前的一些额外动作
        Date endTime = beforeSync();
        if (Objects.isNull(endTime)) {
            log.info("请先执行“流程实例数据同步 ES 第一步”, Bean Name：completedProcessInstanceSyncToEs。 待其执行完成后再执行该任务！");
            XxlJobHelper.log("请先执行“流程实例数据同步 ES 第一步”, Bean Name：completedProcessInstanceSyncToEs。 待其执行完成后再执行该任务！");
            return ReturnT.FAIL;
        }

        // 开始同步流程数据到 ES
        DataSyncSummaryDTO summary = doSync(endTime, param, shardIndex, shardTotal);

        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;
    }

    private Date beforeSync() {
        CommandExecutor commandExecutor = springProcessEngineConfiguration.getCommandExecutor();
        return commandExecutor.execute(new CustomGetPropertyCmd(LATEST_SYNC_TO_ELASTICSEARCH_TIME));
    }

    /**
     * 开始处理的入口
     *
     * @param endTime
     * @param pageSizeStr
     */
    private DataSyncSummaryDTO doSync(Date endTime, String pageSizeStr, int shardIndex, int shardTotal) {
        Long totalCount = bpmnProcessInstanceForEsService.queryHistoricProcessInstanceByUnfinishedAndAlterEndTimeTotalCount(endTime);
        log.info("查询到待同步实例维度数据量: {} 条", totalCount);
        XxlJobHelper.log("查询到待同步实例维度数据量: {} 条", totalCount);
        AtomicInteger pageSize = new AtomicInteger(50);
        if (StringUtils.hasText(pageSizeStr)) {
            try {
                pageSize.set(Integer.parseInt(pageSizeStr));
            } catch (Exception e) {
                // ignore
            }
        }
        AtomicLong totalProcessTaskCount = new AtomicLong(0);
        IntStream.iterate(0, i -> i + pageSize.get()).limit((totalCount + pageSize.get() - 1) / pageSize.get())
                .forEach(skipRows -> {
                    log.info("处理进度: {} %, current startRow: {}, endRow: {}", String.format("%.2f", (double) skipRows / totalCount * 100),
                            skipRows, skipRows + pageSize.get());
                    XxlJobHelper.log("处理进度: {} %, current startRow: {}, endRow: {}", String.format("%.2f", (double) skipRows / totalCount * 100),
                            skipRows, skipRows + pageSize.get());
                    int pageNo = skipRows / pageSize.get() + 1;
                    List<HistoricProcessInstance> instances = bpmnProcessInstanceForEsService.queryHistoricProcessInstanceByUnfinishedAndAlterEndTime(endTime, new Page<HistoricProcessInstance>(pageNo, pageSize.get()));

                    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(totalCount, 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;
    }

}
