package com.aizuda.snailjob.server.common.alarm;

import cn.hutool.core.collection.CollUtil;
import com.aizuda.snailjob.common.core.alarm.Alarm;
import com.aizuda.snailjob.common.core.alarm.AlarmContext;
import com.aizuda.snailjob.common.core.alarm.SnailJobAlarmFactory;
import com.aizuda.snailjob.common.core.enums.StatusEnum;
import com.aizuda.snailjob.common.core.util.JsonUtil;
import com.aizuda.snailjob.common.core.util.StreamUtils;
import com.aizuda.snailjob.common.log.SnailJobLog;
import com.aizuda.snailjob.server.common.Lifecycle;
import com.aizuda.snailjob.server.common.cache.CacheNotifyRateLimiter;
import com.aizuda.snailjob.server.common.convert.AlarmInfoConverter;
import com.aizuda.snailjob.server.common.dto.AlarmInfo;
import com.aizuda.snailjob.server.common.dto.NotifyConfigInfo;
import com.aizuda.snailjob.server.common.dto.NotifyConfigInfo.RecipientInfo;
import com.aizuda.snailjob.server.common.enums.SyetemTaskTypeEnum;
import com.aizuda.snailjob.template.datasource.access.AccessTemplate;
import com.aizuda.snailjob.template.datasource.persistence.mapper.NotifyRecipientMapper;
import com.aizuda.snailjob.template.datasource.persistence.po.NotifyConfig;
import com.aizuda.snailjob.template.datasource.persistence.po.NotifyRecipient;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationEvent;
import org.springframework.scheduling.TaskScheduler;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author xiaowoniu
 * @date 2023-12-03 09:47:11
 * @since 2.5.0
 */
@Slf4j
public abstract class AbstractAlarm<E extends ApplicationEvent, A extends AlarmInfo> implements
        Runnable,
        Lifecycle {

    @Autowired
    @Qualifier("alarmExecutorService")
    protected TaskScheduler taskScheduler;
    @Autowired
    protected AccessTemplate accessTemplate;
    @Autowired
    protected NotifyRecipientMapper recipientMapper;

    @Override
    public void run() {

        try {
            // 从队列获取数据
            List<A> alarmInfos = poll();
            if (CollUtil.isEmpty(alarmInfos)) {
                return;
            }

            // 通知场景
            Set<Integer> notifyScene = new HashSet<>();

            // 转换AlarmDTO 为了下面循环发送使用
            Map<Long, List<A>> waitSendAlarmInfos = convertAlarmDTO(alarmInfos, notifyScene);
            Set<Long> notifyIds = waitSendAlarmInfos.keySet();

            // 批量获取通知配置
            Map<Long, NotifyConfigInfo> notifyConfigMap = obtainNotifyConfig(notifyScene, notifyIds);

            // 循环发送消息
            waitSendAlarmInfos.forEach((key, list) -> {
                Optional.ofNullable(notifyConfigMap.get(key)).ifPresent(notifyConfig -> {
                    for (A alarmDTO : list) {
                        sendAlarm(notifyConfig, alarmDTO);
                    }
                });
            });
        } catch (InterruptedException e) {
            SnailJobLog.LOCAL.info("retry task fail dead letter alarm stop");
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            SnailJobLog.LOCAL.error("RetryTaskFailDeadLetterAlarmListener queue poll Exception", e);
        }
    }

    protected Map<Long, NotifyConfigInfo> obtainNotifyConfig(Set<Integer> notifyScene,
                                                             Set<Long> notifyIds) {
        if (CollUtil.isEmpty(notifyIds) || CollUtil.isEmpty(notifyScene)) {
            return Maps.newHashMap();
        }

        // 批量获取所需的通知配置
        List<NotifyConfig> notifyConfigs = accessTemplate.getNotifyConfigAccess().list(
                new LambdaQueryWrapper<NotifyConfig>()
                        .eq(NotifyConfig::getNotifyStatus, StatusEnum.YES.getStatus())
                        .in(NotifyConfig::getNotifyScene, notifyScene)
                        .in(NotifyConfig::getSystemTaskType, StreamUtils.toList(getSystemTaskType(), SyetemTaskTypeEnum::getType))
                        .in(NotifyConfig::getId, notifyIds)
        );
        if (CollUtil.isEmpty(notifyConfigs)) {
            return Maps.newHashMap();
        }

        Set<Long> recipientIds = notifyConfigs.stream()
                .flatMap(config -> JsonUtil.parseList(config.getRecipientIds(), Long.class).stream())
                .collect(Collectors.toSet());

        List<NotifyRecipient> notifyRecipients = recipientMapper.selectByIds(recipientIds);
        Map<Long, NotifyRecipient> recipientMap = StreamUtils.toIdentityMap(notifyRecipients, NotifyRecipient::getId);

        if (CollUtil.isEmpty(recipientIds)) {
            return Maps.newHashMap();
        }

        List<NotifyConfigInfo> notifyConfigInfos = AlarmInfoConverter.INSTANCE.retryToNotifyConfigInfos(notifyConfigs);
        for (NotifyConfigInfo notifyConfigInfo : notifyConfigInfos) {
            List<RecipientInfo> recipients = StreamUtils.toList(notifyConfigInfo.getRecipientIds(), recipientId -> {
                NotifyRecipient notifyRecipient = recipientMap.get(recipientId);
                if (Objects.isNull(notifyRecipient)) {
                    return null;
                }

                RecipientInfo recipientInfo = new RecipientInfo();
                recipientInfo.setNotifyAttribute(notifyRecipient.getNotifyAttribute());
                recipientInfo.setNotifyType(notifyRecipient.getNotifyType());
                return recipientInfo;
            });
            notifyConfigInfo.setRecipientInfos(recipients);
        }

        return StreamUtils.toIdentityMap(notifyConfigInfos, NotifyConfigInfo::getId);
    }

    protected abstract List<SyetemTaskTypeEnum> getSystemTaskType();

    protected abstract Map<Long, List<A>> convertAlarmDTO(List<A> alarmData, Set<Integer> notifyScene);

    protected abstract List<A> poll() throws InterruptedException;

    protected abstract AlarmContext buildAlarmContext(A alarmDTO, NotifyConfigInfo notifyConfig);

    @Override
    public void start() {
        startLog();
        taskScheduler.scheduleAtFixedRate(this, Duration.parse("PT1S"));
    }

    protected abstract void startLog();

    @Override
    public void close() {
    }

    protected void sendAlarm(NotifyConfigInfo notifyConfig, A alarmDTO) {
        if (Objects.equals(notifyConfig.getRateLimiterStatus(), StatusEnum.YES.getStatus())) {
            // 限流
            RateLimiter rateLimiter = getRateLimiter(String.valueOf(notifyConfig.getId()), notifyConfig.getRateLimiterThreshold());
            // 每秒发送rateLimiterThreshold个告警
            if (Objects.nonNull(rateLimiter) && !rateLimiter.tryAcquire(1, TimeUnit.SECONDS)) {
                return;
            }
        }

        // 重试通知阈值
        if (Objects.nonNull(alarmDTO.getCount())
                && Objects.nonNull(notifyConfig.getNotifyThreshold())
                && alarmDTO.getCount() < notifyConfig.getNotifyThreshold()) {
            return;
        }

        for (final RecipientInfo recipientInfo : notifyConfig.getRecipientInfos()) {
            if (Objects.isNull(recipientInfo)) {
                continue;
            }
            AlarmContext context = buildAlarmContext(alarmDTO, notifyConfig);
            context.setNotifyAttribute(recipientInfo.getNotifyAttribute());
            Alarm<AlarmContext> alarm = SnailJobAlarmFactory.getAlarmType(recipientInfo.getNotifyType());
            alarm.asyncSendMessage(context);
        }

    }

    protected RateLimiter getRateLimiter(String key, double rateLimiterThreshold) {
        RateLimiter rateLimiter = CacheNotifyRateLimiter.getRateLimiterByKey(key);
        if (Objects.isNull(rateLimiter) || rateLimiter.getRate() != rateLimiterThreshold) {
            CacheNotifyRateLimiter.put(key, RateLimiter.create(rateLimiterThreshold));
        }

        return rateLimiter;
    }

    protected abstract int getNotifyScene();

    protected abstract void doOnApplicationEvent(E event);
}
