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

import cn.axzo.basics.common.exception.ServiceException;
import cn.axzo.msg.center.dal.mapper.MessageRecordMapper;
import cn.axzo.msg.center.domain.entity.MessageRecord;
import cn.axzo.msg.center.domain.persistence.BaseEntity;
import cn.axzo.msg.center.inside.notices.utils.FunctionalTransactionTemplate;
import cn.axzo.msg.center.message.service.todo.mybatis.ReplaceInterceptor;
import cn.axzo.msg.center.message.service.todo.mybatis.Replacement;
import cn.axzo.msg.center.utils.DateFormatUtil;
import cn.axzo.msg.center.utils.JSONObjectUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import com.taobao.api.internal.util.NamedThreadFactory;
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.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.springframework.stereotype.Component;

import javax.annotation.Nullable;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static cn.axzo.msg.center.inside.notices.utils.Queries.query;
import static java.util.stream.Collectors.toList;

/**
 * @author yanglin
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class MigrateOldMsgHotDataJob extends IJobHandler {

    private final MessageRecordMapper messageRecordMapper;
    private final FunctionalTransactionTemplate transactionTemplate;
    private final ExecutorService executor = Executors.newSingleThreadExecutor(
            new NamedThreadFactory("migrate-old-msg-hot-data-job"));
    private volatile boolean isRunning = false;

    @Override
    @XxlJob("migrateOldMsgHotDataJob")
    public ReturnT<String> execute(String paramStr) throws Exception {
        log.info("start - run job with param={}", paramStr);
        try {
            Param param = JSONObjectUtil.parseObject(paramStr, Param.class);
            tryExecute(param, false);
            log.info("end - run job with param={}", param);
            return ReturnT.SUCCESS;
        } catch (Exception e) {
            log.warn("job failed. param={}", paramStr, e);
            return ReturnT.FAIL;
        }
    }

    public void tryExecute(@Nullable Param param, boolean async) {
        synchronized (this) {
            if (isRunning) throw new ServiceException("job is running, please wait");
            isRunning = true;
        }
        Runnable task = () -> {
            try {
                executeImpl(param == null ? new Param() : param);
            } finally {
                isRunning = false;
            }
        };
        if (async) {
            executor.execute(task);
        } else {
            task.run();
        }
    }

    private void executeImpl(Param param) {
        Date until = DateTime.now().minusDays(param.daysAgo).toDate();
        Iterable<Page<MessageRecord>> pages = messageRecordMapper
                .iterateBathStickyPage(query(MessageRecord.class)
                        .le(MessageRecord::getCreateAt, until)
                        .orderByAsc(MessageRecord::getId));
        log.info("migrating records before/equal date: {} and it's {} days ago",
                DateFormatUtil.toReadableString(until), param.daysAgo);
        int currentPage = 0, totalPage = 0, totalRecordCount = 0;
        for (Page<MessageRecord> page : pages) {
            ++currentPage;
            // 由于会删除数据, 总页数会变化, 所以只取第一次的总页数才是对的
            if (totalPage == 0) totalPage = (int) page.getPages();
            log.info("migrating page={}, total page={}", currentPage, totalPage);
            for (List<MessageRecord> batch : Lists.partition(page.getRecords(), param.saveBatch)) {
                totalRecordCount += batch.size();
                transactionTemplate.exec(() -> migrateRecords(batch));
                log.info("current total migrated record count={}", totalRecordCount);
            }
        }
        log.info("final total migrated record count={}", totalRecordCount);
    }

    private void migrateRecords(List<MessageRecord> records) {
        ReplaceInterceptor.enable(Replacement.TO_MESSAGE_RECORD_COLD);
        try {
            messageRecordMapper.batchInsertWithId(records);
        } catch (Exception e) {
            List<Long> ids = records.stream()
                    .map(BaseEntity::getId).collect(toList());
            log.warn("migrate records failed, recordIds={}, records={}",
                    JSON.toJSONString(ids), JSON.toJSONString(records), e);
            throw e;
        } finally {
            ReplaceInterceptor.disable();
        }
        List<Long> ids = records.stream()
                .map(MessageRecord::getId)
                .collect(toList());
        messageRecordMapper.deleteBatchIds(ids);
    }

    @Data
    public static class Param {
        private int daysAgo = 180;
        private int saveBatch = 200;
    }

}