package cn.axzo.msg.center.message.xxl;

import cn.axzo.msg.center.common.enums.TableIsDeleteEnum;
import cn.axzo.msg.center.dal.TodoBusinessDao;
import cn.axzo.msg.center.dal.TodoDao;
import cn.axzo.msg.center.dal.mapper.PendingMessageRecordMapper;
import cn.axzo.msg.center.domain.entity.PendingMessageRecord;
import cn.axzo.msg.center.domain.entity.PendingRecordExt;
import cn.axzo.msg.center.domain.entity.Todo;
import cn.axzo.msg.center.domain.entity.TodoBusiness;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
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.handler.IJobHandler;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static cn.axzo.msg.center.inside.notices.utils.Queries.query;
import static java.util.function.Function.identity;
import static java.util.stream.Collectors.groupingBy;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;

/**
 * -- 统计老的
 * SELECT COUNT(*) FROM pending_message_record WHERE is_delete = 0;
 * SELECT COUNT(*) FROM (SELECT DISTINCT template_code, biz_code FROM pending_message_record WHERE is_delete = 0) a;
 * -- 统计新的
 * SELECT COUNT(*) FROM todo_business;
 * SELECT COUNT(*) FROM todo;
 *
 * @author yanglin
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class MigrateFromPendingMessageJob extends IJobHandler {

    private final PendingMessageRecordMapper pendingMessageRecordMapper;
    private final TodoBusinessDao todoBusinessDao;
    private final TodoDao todoDao;

    @Override
    @XxlJob("migrateFromPendingMessageJob")
    public ReturnT<String> execute(String jsonStr) {
        return doMigrate();
    }

    @NotNull
    private ReturnT<String> doMigrate() {
        try {
            log.info("开始执行...");
            MigrateCount migrateCount = new MigrateCount();

            List<Long> businessIds;
            businessIds = todoDao.getBaseMapper().getMigratedBusinessId();
            log.info("remove businessIds:{}", JSON.toJSONString(businessIds));

            if (!businessIds.isEmpty()) {
                for (List<Long> batch : Lists.partition(businessIds, 2000)) {
                    int removedBusiness = todoBusinessDao.getBaseMapper().deleteByIds(batch);
                    log.info("remove businessIds:{}, count={}", JSON.toJSONString(batch), removedBusiness);
                }
            }

            int removedTodo = todoDao.getBaseMapper().delete(
                    query(Todo.class)
                            .last("AND record_ext -> '$.isMigratedFromPendingMessage' = true"));
            log.info("remove todos count={}", removedTodo);

            new MigrateBizCodePresentAction(migrateCount).compute();
            new MigrateBizCodeAbsentAction(migrateCount).compute();
            log.info("迁移结束, migrateCount={}", migrateCount);
            return ReturnT.SUCCESS;
        } catch (Exception e) {
            log.error("执行错误", e);
            return ReturnT.FAIL;
        }
    }

    @RequiredArgsConstructor
    private class MigrateBizCodePresentAction extends RecursiveAction {

        final MigrateCount migrateCount;

        @Override
        protected void compute() {
            Supplier<List<TemplateAndBiz>> tbFun = templateBizPageFun(false);
            int i = 0;
            for (List<TemplateAndBiz> tbs = tbFun.get(); !tbs.isEmpty(); tbs = tbFun.get()) {
                int count = ++i;
                log.info("start - MigrateBizCodePresentAction {}", count);
                new MigrateBizCodePresentPerBizCodesAction(migrateCount, tbs).compute();
                log.info("end - MigrateBizCodePresentAction {}", count);
            }
        }

    }

    // 按templateCode和bizCode迁移
    @RequiredArgsConstructor
    private class MigrateBizCodePresentPerBizCodesAction extends RecursiveAction {

        final MigrateCount migrateCount;
        final List<TemplateAndBiz> tbs;

        @Override
        protected void compute() {
            String tbsStr = tbs.stream().map(TemplateAndBiz::toString).collect(Collectors.joining(","));
            log.info("migrate biz code present: {}", tbsStr);
            // 按template code和biz code 获取下所有待办
            LambdaQueryWrapper<PendingMessageRecord> query = query(PendingMessageRecord.class)
                    .eq(PendingMessageRecord::getIsDelete, TableIsDeleteEnum.NORMAL.value);
            query.and(nested -> {
                for (TemplateAndBiz tb : tbs) {
                    nested.or()
                            .eq(PendingMessageRecord::getTemplateCode, tb.templateCode)
                            .eq(PendingMessageRecord::getBizCode, tb.bizCode);
                }
            });

            List<PendingMessageRecord> allRecords = pendingMessageRecordMapper.selectList(query);
            Map<String, List<PendingMessageRecord>> templateBizCode2Records = allRecords.stream()
                    .collect(groupingBy(MigrateFromPendingMessageJob::formatTemplateAndBizCode));

            ArrayList<TodoBusiness> businesses = new ArrayList<>();
            for (List<PendingMessageRecord> recordsPerTemplateCodeAndBiz : templateBizCode2Records.values()) {
                PendingMessageRecord sample = recordsPerTemplateCodeAndBiz.get(0);
                businesses.add(buildTodoBusiness(sample));
            }

            Map<String, TodoBusiness> templateAndBizCode2Business = businesses.stream()
                    .collect(toMap(r -> formatTemplateAndBizCode(r.getTemplateCode(), r.getBizCode()), identity()));
            try {
                todoBusinessDao.saveBatch(businesses);
            } catch (Exception e) {
                log.error("MigrateBizCodePresentPerBizCodesAction, businesses={}", JSON.toJSONString(businesses), e);
                throw e;
            }
            migrateCount.businessCount.addAndGet(businesses.size());

            ArrayList<Todo> todos = new ArrayList<>();
            for (PendingMessageRecord record : allRecords) {
                TodoBusiness business = templateAndBizCode2Business.get(formatTemplateAndBizCode(record));
                todos.add(buildTodo(business, record));
            }
            migrateCount.todoCount.addAndGet(todos.size());
            for (List<Todo> todosBatch : Lists.partition(todos, 200)) {
                try {
                    todoDao.saveBatch(todosBatch);
                } catch (Exception e) {
                    log.error("MigrateBizCodePresentPerBizCodesAction, todosBatch={}", JSON.toJSONString(todosBatch), e);
                    throw e;
                }
            }
        }

    }

    @RequiredArgsConstructor
    private class MigrateBizCodeAbsentAction extends RecursiveAction {

        final MigrateCount migrateCount;

        @Override
        protected void compute() {
            Supplier<List<TemplateAndBiz>> tbFun = templateBizPageFun(true);
            List<TemplateAndBiz> records = tbFun.get();
            if (records.isEmpty()) {
                log.info("没有找到bizCode为空的模版");
                return;
            }
            int i = 0;
            // bizCode的记录的模版是有限的, 只有几个
            for (TemplateAndBiz bizCodeRecord : records) {
                int count = ++i;
                log.info("start - MigrateBizCodeAbsentAction {}", count);
                new MigrateBizCodeAbsentPerTemplateAction(migrateCount, bizCodeRecord).compute();
                log.info("end - MigrateBizCodeAbsentAction {}", count);
            }
        }

    }

    @RequiredArgsConstructor
    private class MigrateBizCodeAbsentPerTemplateAction extends RecursiveAction {

        final MigrateCount migrateCount;
        final TemplateAndBiz tb;

        @Override
        protected void compute() {
            // 防止一上线就有人生成了bizCode为空的待办, 避免重复生成
            TodoBusiness business = findBusiness();
            if (business == null) {
                PendingMessageRecord sample = pendingMessageRecordMapper.selectOne(query(PendingMessageRecord.class)
                        .eq(PendingMessageRecord::getTemplateCode, tb.templateCode)
                        .eq(PendingMessageRecord::getBizCode, tb.bizCode)
                        .last("LIMIT 1"));
                business = buildTodoBusiness(sample);
                // 尽量不用事务
                try {
                    todoBusinessDao.save(business);
                } catch (DuplicateKeyException e) {
                    log.error("MigrateBizCodeAbsentPerTemplateAction, todosBatch={}", JSON.toJSONString(business), e);
                    business = findBusiness();
                } catch (Exception e) {
                    log.error("MigrateBizCodeAbsentPerTemplateAction, todosBatch={}", JSON.toJSONString(business), e);
                    throw e;
                }
                migrateCount.businessCount.incrementAndGet();
            }

            Page<PendingMessageRecord> samplePage = absentPageData(tb.templateCode, 1);

            int j = 0;
            for (int i = 1; i <= samplePage.getPages(); i++) {
                int count = ++j;
                log.info("start - MigrateBizCodeAbsentPerTemplateAction {}", count);
                new MigrateBizCodeAbsentPerPageDataAction(migrateCount, business, i, samplePage.getPages()).compute();
                log.info("end - MigrateBizCodeAbsentPerTemplateAction {}", count);
            }
        }

        private TodoBusiness findBusiness() {
            return todoBusinessDao.lambdaQuery()
                    .eq(TodoBusiness::getTemplateCode, tb.templateCode)
                    .eq(TodoBusiness::getBizCode, tb.bizCode)
                    .one();
        }

    }

    @RequiredArgsConstructor
    private class MigrateBizCodeAbsentPerPageDataAction extends RecursiveAction {

        private final MigrateCount migrateCount;
        private final TodoBusiness business;
        private final int pageNumber;
        private final long totalPageNumber;

        @Override
        protected void compute() {
            log.info("migrate biz code absent, templateCode={}, pageNumber={}, totalPageNumber={}",
                    business.getTemplateCode(), pageNumber, totalPageNumber);
            List<PendingMessageRecord> records = absentPageData(business.getTemplateCode(), pageNumber).getRecords();
            List<Todo> todos = records.stream()
                    .map(r -> buildTodo(business, r))
                    .collect(toList());
            try {
                todoDao.saveBatch(todos);
            } catch (Exception e) {
                log.error("MigrateBizCodeAbsentPerPageDataAction, todosBatch={}", JSON.toJSONString(todos), e);
                throw e;
            }
            migrateCount.todoCount.addAndGet(todos.size());
        }

    }

    private Page<PendingMessageRecord> absentPageData(String templateCode, int pageNumber) {
        Page<PendingMessageRecord> page = new Page<>();
        page.setCurrent(pageNumber);
        page.setSize(4000);
        return pendingMessageRecordMapper.selectPage(page, query(PendingMessageRecord.class)
                .eq(PendingMessageRecord::getTemplateCode, templateCode)
                .eq(PendingMessageRecord::getBizCode, "")
                .orderByAsc(PendingMessageRecord::getId));
    }

    // !! helper

    private Supplier<List<TemplateAndBiz>> templateBizPageFun(boolean forEmptyBizCode) {
        Page<PendingMessageRecord> bizCodePage = new Page<>();
        bizCodePage.setCurrent(0);
        bizCodePage.setSize(2000);
        return () -> {
            bizCodePage.setCurrent(bizCodePage.getCurrent() + 1);
            QueryWrapper<PendingMessageRecord> query = new QueryWrapper<>();
            query.select("distinct template_code, biz_code")
                    .orderByAsc("template_code, biz_code");
            query.lambda()
                    .eq(PendingMessageRecord::getIsDelete, TableIsDeleteEnum.NORMAL.value)
                    .ne(!forEmptyBizCode, PendingMessageRecord::getBizCode, "")
                    .eq(forEmptyBizCode, PendingMessageRecord::getBizCode, "");
            return pendingMessageRecordMapper.selectPage(bizCodePage, query).getRecords().stream()
                    .map(i -> new TemplateAndBiz(i.getTemplateCode(), i.getBizCode()))
                    .collect(toList());
        };
    }

    private TodoBusiness buildTodoBusiness(PendingMessageRecord sample) {
        TodoBusiness business = new TodoBusiness();
        business.setTemplateCode(sample.getTemplateCode());
        business.setBizCode(sample.getBizCode());
        business.setTitle(sample.getTitle());
        business.setContent(sample.getContent());
        business.setPromoterId(sample.getPromoterId());
        business.setPromoterPersonId(sample.getPromoterPersonId());
        business.setPromoterName(sample.getPromoterName());
        business.setPromoterType(sample.getPromoterType());
        business.setBizFinalState(sample.getBizFinalState());
        business.setBizFlag(sample.getBizFlag());
        business.setBizCategory(sample.getBizCategory());
        business.setBizExtParam(StringUtils.isBlank(sample.getBizExtParam())
                ? null : parseJson(sample.getBizExtParam()));
        business.setRouterParams(StringUtils.isBlank(sample.getRouterParams())
                ? null : parseJson(sample.getRouterParams()));
        business.setDeadline(sample.getDeadline());
        business.setCreateAt(sample.getCreateAt());
        business.setUpdateAt(sample.getUpdateAt());
        business.setSampleTodoCode(sample.getIdentityCode());
        return business;
    }

    private static JSONObject parseJson(String json) {
        try {
            return JSON.parseObject(json);
        } catch (Exception e) {
            try {
                // 老数据
                JSONObject merged = new JSONObject();
                JSONArray array = JSON.parseArray(json);
                for (Object obj : array) {
                    if (obj instanceof JSONObject) {
                        merged.putAll((JSONObject) obj);
                    }
                }
                return merged;
            } catch (Exception ex) {
                log.error("解析错误, json={}", json, ex);
                throw ex;
            }
        }
    }

    private Todo buildTodo(TodoBusiness business, PendingMessageRecord record) {
        Todo todo = new Todo();
        todo.setTodoBusinessId(business.getId());
        todo.setIdentityCode(record.getIdentityCode());
        todo.setSrcTemplateCode(record.getTemplateCode());
        todo.setTemplateCode(record.getTemplateCode());
        todo.setBizCode(record.getBizCode());
        todo.setSubBizCode(record.getSubBizCode());
        todo.setType(record.getTodoType());
        todo.setState(record.getState());
        todo.setOuId(record.getOuId());
        todo.setTitle(record.getTitle());
        todo.setContent(record.getContent());
        todo.setBizExtParam(StringUtils.isBlank(record.getBizExtParam())
                ? null : parseJson(record.getBizExtParam()));
        todo.setRouterParams(StringUtils.isBlank(record.getRouterParams())
                ? null : parseJson(record.getRouterParams()));
        todo.setExecutorPersonId(record.getExecutorPersonId());
        todo.setExecutorId(record.getExecutorId());
        todo.setRequestNo(record.getRequestNo());
        todo.setExecutorName(record.getExecutorName());
        todo.setExecutorType(record.getExecutorType());
        todo.setOrgType(record.getOrgType());
        todo.setOrgId(record.getOrgId());
        todo.setOrgName(record.getOrgName());
        todo.setBizDesc(record.getBizDesc());
        todo.setHideUntil(record.getHideUntil());
        todo.setIsOuIdMigrated(record.getIsOuIdMigrated());
        todo.setCreateAt(record.getCreateAt());
        todo.setUpdateAt(record.getUpdateAt());
        PendingRecordExt ext = record.getRecordExt();
        if (ext == null) {
            ext = new PendingRecordExt();
        }
        ext.setIsMigratedFromPendingMessage(true);
        todo.setRecordExt(ext);
        return todo;
    }

    private static String formatTemplateAndBizCode(PendingMessageRecord p) {
        return formatTemplateAndBizCode(p.getTemplateCode(), p.getBizCode());
    }

    private static String formatTemplateAndBizCode(String templateCode, String bizCode) {
        return String.format("%s:%s", templateCode, bizCode);
    }

    @RequiredArgsConstructor
    private static class TemplateAndBiz {
        final String templateCode;
        final String bizCode;

        @Override
        public int hashCode() {
            return toString().hashCode();
        }

        @Override
        public String toString() {
            return String.format("%s:%s", templateCode, bizCode);
        }
    }

    private static class MigrateCount {

        final AtomicInteger todoCount = new AtomicInteger();
        final AtomicInteger businessCount = new AtomicInteger();

        @Override
        public String toString() {
            return String.format("待办业务: %s, 待办数量: %s", todoCount.get(), businessCount.get());
        }
    }

}