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

import cn.axzo.basics.common.exception.ServiceException;
import cn.axzo.msg.center.notices.common.enums.ReturnCodeEnum;
import cn.axzo.msg.center.notices.manager.api.dto.request.MnsRequestDto;
import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * @author yanglin
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class MnsLimiter {

    private final MnsProperties props;
    private final StringRedisTemplate stringRedisTemplate;

    /**
     * 基于固定窗口, 限制重复内容的发送次数
     */
    void maybeLimitSend(MnsRequestDto request) {
        if (!shouldLimit(request)) {
            return;
        }
        log.info("try - send mns, request={}", request);
        // key中包含了固定窗口因子
        String key = buildTimeAwareKey(request);
        String countStr = stringRedisTemplate.opsForValue().get(key);
        if (StringUtils.isBlank(countStr)) {
            return;
        }
        long alreadySendCount = Long.parseLong(countStr);

        MnsType mnsType = getMnsType(request);

        long limitCount;
        String error;
        if (mnsType == MnsType.BIZ) {
            limitCount = props.getBizLimitPerDay();
            error = String.format("业务短信: 每天发送给同一个手机号的重复内容不能超过 %s 条。 请勿重试发送!!", limitCount);
        } else {
            limitCount = props.getVerifyCodeLimitPerWindow();
            error = String.format("验证码短信: 每 %s 分钟发送给同一个手机号的重复内容不能超过 %s 条。 请勿重试发送!!",
                    props.getVerifyCodeLimitWindowMinutes(), limitCount);
        }
        if (alreadySendCount + 1 > limitCount) {
            log.warn("mns rate limited, {}, request={}", error, request);
            throw new ServiceException(ReturnCodeEnum.FAIL.getCode(), error);
        }
    }

    void setSendSuccess(MnsRequestDto request) {
        if (!shouldLimit(request)) {
            return;
        }
        log.info("success - send mns, request={}", request);
        // key中包含了固定窗口因子
        String key = buildTimeAwareKey(request);
        stringRedisTemplate.opsForValue().increment(key);
        // 这里设置过期时间只是为了删除对应的key, 和限制逻辑本身没有太大的关系
        long expireMinutes = getTimeAwareExpireTimeMinutes(request);
        // 延迟删除
        stringRedisTemplate.expire(key, expireMinutes + 1, TimeUnit.MINUTES);
    }

    @SuppressWarnings("BooleanMethodIsAlwaysInverted")
    private boolean shouldLimit(MnsRequestDto request) {
        if (getMnsType(request) == MnsType.VERIFY_CODE) {
            return false;
        }
        return props.shouldLimit(request.getTemplateNo());
    }

    void setSendFail(MnsRequestDto request, Exception e) {
        log.warn("fail - send mns, request={}", request, e);
    }

    /**
     * 获取key过期时间
     */
    private long getTimeAwareExpireTimeMinutes(MnsRequestDto request) {
        return getMnsType(request) == MnsType.BIZ
                ? TimeUnit.DAYS.toMinutes(1)
                : props.getVerifyCodeLimitWindowMinutes();
    }

    /**
     * 构建包含时间元素的缓存key, key中包含了固定窗口因子. 固定窗口: 限制发送重复的内容
     */
    private String buildTimeAwareKey(MnsRequestDto request) {
        StringBuilder buf = new StringBuilder();
        buf.append("msg-center:mns-limit:v1");
        Consumer<String> appender = value -> {
            if (StringUtils.isNotBlank(value))
                buf.append(":").append(value);
        };
        // 通过templateNo和params就可以判断出发送的内容是不是一样的
        appender.accept(getMnsType(request).getCacheKey());
        appender.accept(request.getTemplateNo());
        appender.accept(request.getPhoneNo());
        appender.accept(buildTimeAwareSegment(request));
        appender.accept(hashParams(request.getParams()));
        return buf.toString();
    }

    private String buildTimeAwareSegment(MnsRequestDto request) {
        Date now = new Date();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
        int windowMinutes = props.getVerifyCodeLimitWindowMinutes();
        return getMnsType(request) == MnsType.BIZ
                ? sdf.format(now)
                : String.format("%s-%d-%d",
                sdf.format(now), windowMinutes, getVerifySegmentIndex(now, windowMinutes));
    }

    public static int getVerifySegmentIndex(Date now, int windowMinutes) {
        DateTime dateTime = new DateTime(now, DateTimeZone.forID("Asia/Shanghai"));
        // Calculate the total number of intervals in a day
        int intervalsPerDay = 24 * 60 / windowMinutes;
        // Get the current time in minutes
        int currentMinutes = dateTime.getHourOfDay() * 60 + dateTime.getMinuteOfHour();
        // Find the index of the interval that the current time falls into
        int intervalIndex = currentMinutes / windowMinutes;
        // Ensure the index is within the range of the array
        if (intervalIndex >= intervalsPerDay) {
            intervalIndex = intervalsPerDay - 1;
        }
        return intervalIndex;
    }

    @SuppressWarnings("UnstableApiUsage")
    private static String hashParams(Map<String, Object> params) {
        if (params == null || params.isEmpty()) {
            return "";
        }
        Hasher hasher = Hashing.murmur3_128().newHasher();
        Consumer<Object> appender = value -> {
            if (value != null) {
                hasher.putString(String.valueOf(value), StandardCharsets.UTF_8);
            }
        };
        for (Map.Entry<String, Object> e : new TreeMap<>(params).entrySet()) {
            if (e.getKey() == null || e.getValue() == null) {
                continue;
            }
            appender.accept(e.getKey());
            appender.accept(e.getValue());
        }
        return hasher.hash().toString();
    }

    private static MnsType getMnsType(MnsRequestDto request) {
        MnsType msnType = request.getInternalObj(MnsType.class);
        if (msnType == null) {
            msnType = MnsType.BIZ;
        }
        return msnType;
    }

}