package com.ruoyi.common.third.cache.redis;

import com.ruoyi.common.third.cache.HzCache;
import com.ruoyi.common.third.cache.util.StringUtil;
import com.ruoyi.common.third.constants.StringPool;
import org.springframework.cache.support.SimpleValueWrapper;
import org.springframework.core.convert.ConversionService;
import org.springframework.data.redis.cache.RedisCache;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.lang.Nullable;

import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;


public class HzRedisCache extends RedisCache implements HzCache {

    private final RedisConnectionFactory redisConnectionFactory;
    private final RedisCacheConfiguration cacheConfig;
    private final ConversionService conversionService;
    private final String name;
    private final HzRedisCacheWriter cacheWriter;

    protected HzRedisCache(RedisCache cache, HzRedisCacheWriter cacheWriter, RedisConnectionFactory redisConnectionFactory) {
        super(cache.getName(), cacheWriter, cache.getCacheConfiguration());
        this.redisConnectionFactory = redisConnectionFactory;
        this.cacheConfig = cache.getCacheConfiguration();
        this.conversionService = cacheConfig.getConversionService();
        this.cacheWriter = cacheWriter;
        this.name = cache.getName();
    }

    @Override
    @Nullable
    public ValueWrapper get(Object key) {
        return toValueWrapper(lookup(getRedisCacheKey(key)));
    }

    @Override
    @SuppressWarnings("unchecked")
    @Nullable
    public <T> T get(Object key, @Nullable Class<T> type) {
        Object value = fromStoreValue(lookup(getRedisCacheKey(key)));
        if (value != null && type != null && !type.isInstance(value)) {
            throw new IllegalStateException(
                    "Cached value is not of required type [" + type.getName() + "]: " + value);
        }
        return (T) value;
    }

    @Override
    public void put(Object key, Object value, Duration timeout) {
        cacheWriter.put(this.name,redisSerializeCacheKey(key),redisSerializeCacheValue(value),timeout);
    }

    @Override
    public void put(Object key, Object value, long seconds) {
        Duration timeoutDuration = null;
        if (seconds > 0) {
            timeoutDuration = Duration.ofSeconds(seconds);
        }else{
            timeoutDuration = Duration.ZERO;
        }
        this.put(key,value,timeoutDuration);
    }

    /**
     * 不存在 放入
     * 存在 获取之前的值
     *
     * @return ValueWrapper
     * */
    @Override
    public ValueWrapper putIfAbsent(Object key, @Nullable Object value){
        ValueWrapper existingValue = get(getRedisCacheKey(key));
        if (existingValue == null) {
            put(key, value);
            existingValue = new SimpleValueWrapper(value);
        }
        return existingValue;
    }

    @Override
    public ValueWrapper putIfAbsent(Object key, @Nullable Object value,Duration timeout){
        ValueWrapper existingValue = get(getRedisCacheKey(key));
        if (existingValue == null) {
            put(key, value, timeout);
            existingValue = new SimpleValueWrapper(value);
        }
        return existingValue;
    }

    @Override
    public ValueWrapper putIfAbsent(Object key, @Nullable Object value,long seconds){
        Duration timeoutDuration;
        if (seconds > 0) {
            timeoutDuration = Duration.ofSeconds(seconds);
        }else{
            timeoutDuration = Duration.ZERO;
        }
        ValueWrapper existingValue = get(getRedisCacheKey(key));
        if (existingValue == null) {
            this.put(key,value,timeoutDuration);
            existingValue = new SimpleValueWrapper(value);
        }
        return existingValue;
    }

    @Override
    public boolean containKey(Object key) {
        return cacheWriter.execute(this.name, conection-> conection.exists(conversionService.convert(createCacheKey(getRedisCacheKey(key)), byte[].class)));
    }


    @Override
    public List<String> getKeys(String key) {
        return this.scanKeys(key);
    }

    @Override
    public List<String> scanKeys(String key) {
        return cacheWriter.execute(this.name,connection->{
            byte[] pattern = conversionService.convert(createCacheKey(getRedisCacheKey(key)), byte[].class);
            List<String> keys = new ArrayList<>();
            List<String> finalKeys = new ArrayList<>();
            connection.scan(new ScanOptions.ScanOptionsBuilder().match(redisDeserializeCacheKey(pattern)).count(Integer.MAX_VALUE).build())
                    .forEachRemaining(next-> finalKeys.add(redisDeserializeCacheKey(next)));
            keys = Optional.ofNullable(finalKeys).orElse(Collections.emptyList());
            return keys;
        });
    }

    @Override
    public void evict(Object key) {
        this.cacheWriter.remove(this.name, conversionService.convert(createCacheKey(getRedisCacheKey(key)), byte[].class));
    }

    @Override
    public void evictKeys(String key) {
        byte[] pattern = conversionService.convert(createCacheKey(getRedisCacheKey(key)), byte[].class);
        cacheWriter.clean(this.name, pattern);
    }

    @Override
    public Map<String, Object> getKeysAndValues(String key) {
        return cacheWriter.execute(this.name,connection->{
            byte[] pattern = conversionService.convert(createCacheKey(getRedisCacheKey(key)), byte[].class);
            Map<String,Object> mp = Optional.ofNullable(connection.keys(pattern)).orElse(Collections.emptySet())
                    .stream().collect(Collectors.toMap(m->redisDeserializeCacheKey(m),m->this.deserializeCacheValue(connection.get(m))));
            return mp;
        });
    }

    @Override
    public Long getKeyExpireTime(Object key) {
        Long timeout = cacheWriter.execute(this.name,conection-> conection.ttl(conversionService.convert(createCacheKey(getRedisCacheKey(key)), byte[].class)));
        return timeout;
    }

    /***
     * key的序列化方法
     * */
    private byte[] redisSerializeCacheKey(Object key) {
        String cacheKey = getRedisCacheKey(key);
        cacheKey = createCacheKey(cacheKey);

        return serializeCacheKey(cacheKey);
    }

    private String getRedisCacheKey(Object key){
        // zp 2022年9月7日13:29:11 增加处理查询的出来的key的问题 .
        // todo 自己定义key 禁止使用 '::'进行分割，否则会被清理掉
        String cacheKey = convertKey(key);
        if (cacheConfig.usePrefix()) {
            final String keyPrefixFor = cacheConfig.getKeyPrefixFor(this.name);
            boolean flag = StringUtil.startWith(cacheKey, StringPool.ASTERISK);
            if(flag){
                cacheKey = StringUtil.removePrefix(cacheKey, StringPool.ASTERISK);
            }
            cacheKey= StringUtil.removePrefix(cacheKey, keyPrefixFor);
            if(flag){
                cacheKey = StringUtil.addPrefixIfNot(cacheKey,StringPool.ASTERISK);
            }
        }
        return cacheKey;
    }

    /**
     * value 的序列化的方法
     * */
    private byte[] redisSerializeCacheValue(Object value){
        return serializeCacheValue(value);
    }

    /***
     * key 发序列化方法
     * */
    private String redisDeserializeCacheKey(byte[] keyByte){
        return cacheConfig.getKeySerializationPair().read(ByteBuffer.wrap(keyByte));
    }


}
