package cn.axzo.msg.center.message.service.impl.oldmsg;

import cn.axzo.msg.center.dal.mapper.MessageRecordMapper;
import cn.axzo.msg.center.domain.entity.MessageRecord;
import cn.axzo.msg.center.message.service.impl.person.PersonService;
import cn.hutool.core.date.StopWatch;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
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.joda.time.DateTime;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

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

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

    private final MessageRecordMapper messageRecordMapper;
    private final PersonService personService;
    private final OldMsgStatCache oldMsgStatCache;

    @Override
    @XxlJob("oldMsgStatCacheReloadJob")
    public ReturnT<String> execute(String param) throws Exception {
        try {
            log.info("start - running job using param: {}", param);
            executeImpl(param);
            log.info("end - running job using param: {}", param);
            return ReturnT.SUCCESS;
        } catch (Exception e) {
            log.warn("run job error, param={}", param, e);
            return ReturnT.FAIL;
        }
    }

    private void executeImpl(String param) throws Exception {
        OldMsgCacheReloadJobParam jobParam = StringUtils.isBlank(param)
                ? OldMsgCacheReloadJobParam.defaultParam()
                : JSON.parseObject(param, OldMsgCacheReloadJobParam.class);
        Date sinceDate = DateTime.now().minusDays(jobParam.determineSyncDay()).toDate();
        StopWatch stopWatch = new StopWatch("reloadBackground driven by job");
        Set<Long> reloadedPersonIds = new HashSet<>();
        Iterable<Page<MessageRecord>> pages = messageRecordMapper
                .iterateBath(query(MessageRecord.class)
                        .ge(MessageRecord::getCreateAt, sinceDate)
                        // 先刷最近的
                        .orderByDesc(MessageRecord::getId));
        for (Page<MessageRecord> page : pages) {
            stopWatch.start("reload stat, records size=" + page.getRecords().size());
            reloadStat(reloadedPersonIds, page);
            stopWatch.stop();
        }
        log.info("end sync old msg stat. reloaded personIds size={}", reloadedPersonIds.size());
        log.info(stopWatch.prettyPrint(TimeUnit.MILLISECONDS));
    }

    private void reloadStat(Set<Long> reloadedPersonIds, Page<MessageRecord> page) throws Exception {
        log.info("start sync old msg stat. page size={}, page current={}, total pages={}",
                page.getSize(), page.getCurrent(), page.getPages());
        List<MessageRecord> records = page.getRecords();
        Set<Long> maybeReloadPersonIds = new HashSet<>();
        // 传了personId的
        records.stream()
                .map(MessageRecord::getPersonId)
                .filter(personId -> personId != 0L)
                .forEach(maybeReloadPersonIds::add);
        // 没有传personId的
        Set<Long> identityIds = records.stream()
                .filter(record -> record.getPersonId() == 0L)
                .map(MessageRecord::getToId)
                .collect(toSet());
        if (!identityIds.isEmpty()) {
            List<Long> fetchedPersonIds = personService.getPersonIdsByIdentities(identityIds).getPersonIds();
            maybeReloadPersonIds.addAll(fetchedPersonIds);
        }
        maybeReloadPersonIds.removeAll(reloadedPersonIds);
        if (!maybeReloadPersonIds.isEmpty()) {
            oldMsgStatCache.reloadBackground(maybeReloadPersonIds);
            reloadedPersonIds.addAll(maybeReloadPersonIds);
        }
    }

}