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

import cn.axzo.framework.rocketmq.Event;
import cn.axzo.framework.rocketmq.EventConsumer;
import cn.axzo.framework.rocketmq.EventHandler;
import cn.axzo.msg.center.api.mq.SendMessageRecordMessage;
import cn.axzo.msg.center.api.request.CmsMsgQueryReq;
import cn.axzo.msg.center.api.request.CmsReadMsgReq;
import cn.axzo.msg.center.api.request.GeneralMessageReq;
import cn.axzo.msg.center.api.request.MessageReadAllReq;
import cn.axzo.msg.center.api.response.MessageNewRes;
import cn.axzo.msg.center.inside.notices.config.PendingMessageBizConfig;
import cn.axzo.msg.center.message.service.MessageSendTwiceRecordService;
import cn.axzo.msg.center.message.service.impl.GeneralMessageOldServiceImpl;
import cn.axzo.msg.center.message.service.impl.oldmsg.OldMsgStatCacheValue.IdentityResponse;
import cn.axzo.msg.center.message.service.impl.person.PersonIdInfo;
import cn.axzo.msg.center.message.service.impl.person.PersonService;
import cn.axzo.msg.center.notices.common.enums.ReturnCodeEnum;
import cn.axzo.msg.center.notices.common.exception.BizException;
import cn.axzo.msg.center.service.dto.IdentityDTO;
import cn.axzo.msg.center.service.enums.MessageCategoryEnum;
import cn.axzo.msg.center.service.enums.MqMessageType;
import cn.axzo.msg.center.service.general.request.GeneralMessageOldDataStatisticRequest;
import cn.axzo.msg.center.service.general.response.GeneralMessageOldDataStatisticResponse;
import cn.azxo.framework.common.model.Page;
import cn.hutool.core.date.StopWatch;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * @author yanglin
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class OldMsgStatCache implements EventHandler, InitializingBean {

    private final ApplicationContext applicationContext;
    private final MessageSendTwiceRecordService messageSendTwiceRecordService;
    private final OldMsgIdentifyFilter oldMsgIdentifyFilter;
    private final StringRedisTemplate stringRedisTemplate;
    private final PendingMessageBizConfig cfg;
    private final EventConsumer eventConsumer;
    private final PersonService personService;
    private final ForkJoinPool foregroundExecutor = new ForkJoinPool(15);
    private final ForkJoinPool backgroundExecutor = new ForkJoinPool(10);

    public GeneralMessageOldDataStatisticResponse getCacheResponseOrReload(GeneralMessageOldDataStatisticRequest request) {
        if (!cfg.determineOldMsgStatCacheOn()) {
            return statisticOldDataImpl(request);
        }
        GeneralMessageOldDataStatisticResponse resp = getCacheResponse(request);
        if (resp != null) {
            resp.setReloadForeground(false);
            return resp;
        }
        GeneralMessageOldDataStatisticResponse response = reloadForeground(request);
        response.setReloadForeground(true);
        return response;
    }

    private GeneralMessageOldDataStatisticResponse getCacheResponse(GeneralMessageOldDataStatisticRequest request) {
        OldMsgStatCacheValue cacheValue = getCacheValue(request.getPersonId());
        if (cacheValue == null) {
            log.info("oldMsgStat: cache miss, need to reload in foreground. request={}", request);
            return null;
        }
        GeneralMessageOldDataStatisticResponse response = cacheValue
                .findResponse(request.getIdentities())
                .orElse(null);
        if (response != null)
            return response;
        // 身份发生了变化, 而且异步刷新不及时
        log.info("oldMsgStat: identities cache miss, need to reload in foreground. request={}", request);
        return null;
    }

    private GeneralMessageOldDataStatisticResponse reloadForeground(GeneralMessageOldDataStatisticRequest request) {
        log.info("oldMsgStat: reload cache for request. request={}", request);
        StopWatch stopWatch = new StopWatch(String.format(
                "%s-reloadForeground", getClass().getSimpleName()));
        stopWatch.start("initPersonCacheValueForeground");
        OldMsgStatCacheValue cacheValue = initPersonCacheValueForeground(request.getPersonId());
        stopWatch.stop();
        GeneralMessageOldDataStatisticResponse response = cacheValue
                .findResponse(request.getIdentities())
                .orElse(null);
        // 边界
        if (response == null) {
            log.info("oldMsgStat: identities cache missing? reload stat for request identities. request={}", request);
            stopWatch.start("initPersonCacheValueForeground - miss again?");
            IdentityResponse newIdentityResponse = getIdentityResponse(request.getPersonId(), request.getIdentities());
            stopWatch.stop();
            cacheValue.addIdentityResponse(newIdentityResponse);
        }
        setCacheValue(request.getPersonId(), cacheValue);
        log.info(stopWatch.prettyPrint(TimeUnit.MILLISECONDS));
        return cacheValue
                .findResponse(request.getIdentities())
                // should never happen
                .orElse(null);
    }

    public Map<Long, OldMsgStatCacheValue> reloadBackground(Set<Long> personIds) throws Exception {
        log.info("oldMsgStat: reload cache for person in background. personIds={}", JSON.toJSONString(personIds));
        if (CollectionUtils.isEmpty(personIds)) {
            log.info("oldMsgStat: personIds is empty");
            return Collections.emptyMap();
        }
        HashMap<Long, OldMsgStatCacheValue> personId2CacheValue = new HashMap<>();
        // 温柔点, 别一次性全丢到线程池
        int bachSize = cfg.getOldMsgStateDataBackendUpdateBatchSize();
        List<List<Long>> batches = Lists.partition(new ArrayList<>(personIds), bachSize);
        for (List<Long> batch : batches) {
            ArrayList<CompletableFuture<PersonAndCacheValue>> futures = new ArrayList<>();
            for (Long personId : batch) {
                futures.add(CompletableFuture.supplyAsync(() -> {
                    OldMsgStatCacheValue cacheValue = initPersonCacheValueBackground(personId);
                    setCacheValue(personId, cacheValue);
                    return new PersonAndCacheValue(personId, cacheValue);
                }, backgroundExecutor));
            }
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            for (CompletableFuture<PersonAndCacheValue> future : futures) {
                PersonAndCacheValue tmp = future.get();
                personId2CacheValue.put(tmp.getPersonId(), tmp.getCacheValue());
            }
        }
        return personId2CacheValue;
    }

    private OldMsgStatCacheValue initPersonCacheValueForeground(Long personId) {
        log.info("initPersonCacheValueForeground. personId={}", personId);
        Function<Boolean, IdentityResponse> buildResponseFun = asManager -> {
            OldMsgIdentifyInfo identifyInfo = oldMsgIdentifyFilter.getIdentifyInfo(personId, asManager);
            return getIdentityResponse(personId, identifyInfo.getIdentities());
        };
        OldMsgStatCacheValue cacheValue = new OldMsgStatCacheValue();
        CompletableFuture<IdentityResponse> asManagerFuture = CompletableFuture
                .supplyAsync(() -> buildResponseFun.apply(true), foregroundExecutor);
        CompletableFuture<IdentityResponse> asWorkerFuture = CompletableFuture
                .supplyAsync(() -> buildResponseFun.apply(false), foregroundExecutor);
        CompletableFuture.allOf(asManagerFuture, asWorkerFuture).join();
        try {
            cacheValue.addIdentityResponse(asManagerFuture.get());
            cacheValue.addIdentityResponse(asWorkerFuture.get());
        } catch (Exception e) {
            log.warn("oldMsgStat: fail to stat old msg. personId={}", personId, e);
            throw new BizException(ReturnCodeEnum.SYSTEM_ERROR.getCode(), e);
        }
        return cacheValue;
    }

    private OldMsgStatCacheValue initPersonCacheValueBackground(Long personId) {
        log.info("initPersonCacheValueBackground. personId={}", personId);
        Function<Boolean, IdentityResponse> buildResponseFun = asManager -> {
            OldMsgIdentifyInfo identifyInfo = oldMsgIdentifyFilter.getIdentifyInfo(personId, asManager);
            return getIdentityResponse(personId, identifyInfo.getIdentities());
        };
        OldMsgStatCacheValue cacheValue = new OldMsgStatCacheValue();
        cacheValue.addIdentityResponse(buildResponseFun.apply(true));
        cacheValue.addIdentityResponse(buildResponseFun.apply(false));
        return cacheValue;
    }

    private IdentityResponse getIdentityResponse(Long personId, List<IdentityDTO> identities) {
        GeneralMessageOldDataStatisticRequest request = new GeneralMessageOldDataStatisticRequest();
        request.setPersonId(personId);
        request.setIdentities(identities);
        GeneralMessageOldDataStatisticResponse response = statisticOldDataImpl(request);
        IdentityResponse identityResponse = new IdentityResponse();
        identityResponse.setResponse(response);
        identityResponse.setIdentities(new HashSet<>(identities));
        return identityResponse;
    }

    // !! cache helper

    private void setCacheValue(Long personId, OldMsgStatCacheValue cacheValue) {
        cacheValue.clear();
        log.info("set cache value for person. personId={}", personId);
        long secondsDelta = TimeUnit.HOURS.toSeconds(cfg.getOldMsgStatCacheDataExpireHours());
        // add some random delta seconds, range: 0-600 seconds
        secondsDelta += new SecureRandom().nextInt(10 * 60);

        String redisKey = buildCacheKey(personId);
        String redisValue = JSON.toJSONString(cacheValue);
        stringRedisTemplate.opsForValue().set(redisKey, redisValue, secondsDelta, TimeUnit.SECONDS);
    }

    private OldMsgStatCacheValue getCacheValue(Long personId) {
        String redisKey = buildCacheKey(personId);
        String json = stringRedisTemplate.opsForValue().get(redisKey);
        if (StringUtils.isBlank(json)) {
            return null;
        }
        try {
            return JSON.parseObject(json, OldMsgStatCacheValue.class);
        } catch (Exception e) {
            // 人为改了缓存, 而且没有改对
            log.warn("fail to parse cache value, will reload. personId={}, cacheValue={}", personId, json, e);
            return null;
        }
    }

    private static String buildCacheKey(Long personId) {
        return String.format("msg-center:old-msg-stat:v1:%d", personId);
    }

    // !! 以下为原有的统计逻辑

    private GeneralMessageOldDataStatisticResponse statisticOldDataImpl(GeneralMessageOldDataStatisticRequest request) {
        if (request.determineIdentities().isEmpty()) {
            return GeneralMessageOldDataStatisticResponse.builder()
                    .unreadCount(0)
                    .latestMsgSendTimestamp(null)
                    .latestMsgContent(null)
                    .build();
        }
        // 查询双发的消息记录
        List<Long> sendTwiceMsgIds = messageSendTwiceRecordService.listByPerson(request.getPersonId());
        // 打破循环依赖, 避免大量改造 (反正快下线了)
        GeneralMessageOldServiceImpl generalMessageOldService = applicationContext
                .getBean(GeneralMessageOldServiceImpl.class);
        // 分页查询最新一条数据
        Page<MessageNewRes> result = generalMessageOldService.pageMsgInfo(build(request, sendTwiceMsgIds));
        // 统计旧的未读普通消息数量
        int count = generalMessageOldService.countUnread(
                request.getPersonId(), request.determineIdentities(), sendTwiceMsgIds);
        // 编排组合成界面展示的数据结构
        MessageNewRes msg = CollectionUtils.isNotEmpty(result.getList()) ? result.getList().get(0) : null;
        return GeneralMessageOldDataStatisticResponse.builder()
                .unreadCount(count)
                .latestMsgSendTimestamp(Optional.ofNullable(msg).map(v -> v.getCreateAt().getTime()).orElse(null))
                .latestMsgContent(Optional.ofNullable(msg).map(MessageNewRes::getContent).orElse(null))
                .build();
    }

    private CmsMsgQueryReq build(GeneralMessageOldDataStatisticRequest request, List<Long> excludeMsgIds) {
        CmsMsgQueryReq req = new CmsMsgQueryReq();
        req.setMsgType(MessageCategoryEnum.GENERAL_MESSAGE.getCode());
        // 这里查询消息中心全部状态的数据
        req.setMsgStatus(0);
        req.setPage(1L);
        req.setPageSize(1L);
        req.setPersonId(request.getPersonId());
        req.setIdentities(request.determineIdentities());
        req.setExcludeMsgIds(excludeMsgIds);
        req.setLogRequest(false);
        return req;
    }

    // !! 处理mq消息, 异步更新缓存

    @Override
    public void afterPropertiesSet() {
        eventConsumer.registerHandler(MqMessageType.OLD_MSG_SEND.getEventCode(), this);
    }

    @Override
    public void onEvent(Event event, EventConsumer.Context context) {
        log.info("oldMsgStat: start - handle mq event, event={}", JSON.toJSONString(event));
        try {
            Thread.sleep(500);
            long start = System.currentTimeMillis();
            handleMqMessage(event);
            long end = System.currentTimeMillis();
            log.info("oldMsgStat: end - handle mq event, used={}ms, event={}", end - start, JSON.toJSONString(event));
        } catch (Exception e) {
            log.warn("oldMsgStat: error - handle mq event, event={}", JSON.toJSONString(event));
        }
    }

    private void handleMqMessage(Event event) throws Exception {
        SendMessageRecordMessage mqMessage = event.normalizedData(SendMessageRecordMessage.class);
        GeneralMessageReq sendRequest = mqMessage.getSendRequest();
        MessageReadAllReq readRequest = mqMessage.getReadRequest();
        CmsReadMsgReq cmsReadRequest = mqMessage.getCmsReadRequest();
        if (sendRequest != null) {
            log.info("handle mq event - driven by sendRequest={}", sendRequest);
            handleSendRequestEvent(sendRequest);
        } else if (readRequest != null) {
            log.info("handle mq event - driven by readRequest={}", readRequest);
            handleReadRequest(readRequest);
        } else if (cmsReadRequest != null) {
            log.info("handle mq event - driven by cmsReadRequest={}", cmsReadRequest);
            handleCmsReadRequest(cmsReadRequest);
        }
    }

    private void handleSendRequestEvent(GeneralMessageReq sendRequest) throws Exception {
        Map<Long, Long> knownIdentityId2PersonId = sendRequest.getToldIdPersonIdMap();
        if (knownIdentityId2PersonId == null)
            knownIdentityId2PersonId = new HashMap<>();
        HashSet<Long> personIds = new HashSet<>(knownIdentityId2PersonId.values());
        Set<Long> receiverIdentityIds = sendRequest.getToId();
        if (receiverIdentityIds == null)
            receiverIdentityIds = Collections.emptySet();
        HashSet<Long> toQueryPersonIdIdentityIds = new HashSet<>();
        for (Long identityId : receiverIdentityIds) {
            if (!knownIdentityId2PersonId.containsKey(identityId))
                toQueryPersonIdIdentityIds.add(identityId);
        }
        if (!toQueryPersonIdIdentityIds.isEmpty()) {
            PersonIdInfo personIdInfo = personService.getPersonIdsByIdentities(toQueryPersonIdIdentityIds);
            personIds.addAll(personIdInfo.getPersonIds());
        }
        StopWatch stopWatch = new StopWatch("reloadBackground");
        stopWatch.start("handleSendRequestEvent");
        reloadBackground(personIds);
        stopWatch.stop();
        log.info(stopWatch.prettyPrint(TimeUnit.MILLISECONDS));
    }

    private void handleReadRequest(MessageReadAllReq readRequest) throws Exception {
        StopWatch stopWatch = new StopWatch("reloadBackground");
        stopWatch.start("handleReadRequest");
        reloadBackground(Sets.newHashSet(readRequest.getPersonId()));
        stopWatch.stop();
        log.info(stopWatch.prettyPrint(TimeUnit.MILLISECONDS));
    }

    private void handleCmsReadRequest(CmsReadMsgReq cmsReadRequest) throws Exception {
        StopWatch stopWatch = new StopWatch("reloadBackground");
        stopWatch.start("handleCmsReadRequest");
        reloadBackground(Sets.newHashSet(cmsReadRequest.getPersonId()));
        stopWatch.stop();
        log.info(stopWatch.prettyPrint(TimeUnit.MILLISECONDS));
    }

}