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

import cn.axzo.basics.common.util.AssertUtil;
import cn.axzo.msg.center.api.response.ReachResp;
import cn.axzo.msg.center.domain.entity.BizEventMapping;
import cn.axzo.msg.center.domain.enums.BizActionCategory;
import cn.axzo.msg.center.message.domain.param.GeneralMessagePushParam;
import cn.axzo.msg.center.message.send.processor.ISendProcessor;
import cn.axzo.msg.center.message.send.processor.ReachContext;
import cn.axzo.msg.center.message.service.BizEventMappingService;
import cn.axzo.msg.center.message.service.CommonMessageRecordService;
import java.util.Optional;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import jodd.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

/**
 * @author syl
 * @date 2024/1/8
 */
@Slf4j
@Service
public class CommonMessageRecordServiceImpl implements CommonMessageRecordService {

    @Value("${msg-im-server.partition-size:500}")
    private Integer partitionSize;

    private final ThreadFactory asyncReachThreadFactory = ThreadFactoryBuilder.create()
            .setDaemon(true).setNameFormat("REACH_IM_MESSAGE_%d").get();
    private final ExecutorService reachExecutorService = new ThreadPoolExecutor(5,
            10, 5, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1024), asyncReachThreadFactory);

    @Autowired
    private BizEventMappingService bizEventMappingService;
    @Autowired
    private ApplicationContext applicationContext;

    @Override
    public ReachResp reach(GeneralMessagePushParam param) {
        String bizEventCode = param.getBizEventCode();
        Optional<BizEventMapping> bizEventOpt = bizEventMappingService.getByBizCode(bizEventCode);
        AssertUtil.isTrue(bizEventOpt.isPresent(), String.format("业务事件映射%s不存在", bizEventCode));
        BizEventMapping bizEvent = bizEventOpt.get();
        AssertUtil.notEmpty(bizEvent.getReachConfig(), String.format("业务事件映射%s无业务动作配置", bizEventCode));

        log.info("start to send reach message. bizCode:[{}]", bizEventCode);
        ReachResp resp = new ReachResp();
        bizEvent.getReachConfig().forEach(r -> {
            BizActionCategory category = BizActionCategory.getCategory(r.getCategory());
            ISendProcessor processor = applicationContext.getBean(category.getProcessor(), ISendProcessor.class);

            ReachContext reachContext = ReachContext.builder()
                    .bizEvent(bizEvent).reachConfig(r)
                    .messageParam(param)
                    .build();
            processor.save(reachContext);
            switch (category) {
                case NOTIFICATION:
                   // resp.setMsgReturnList(reachContext.getMsgReturnList());
                    break;
                case PENDING:
                    resp.setPendingList(reachContext.getPendingList());
                    break;
                default:
            }
            CompletableFuture.runAsync(() -> processor.reach(reachContext), reachExecutorService);
        });
        return resp;
    }
}
