package cn.axzo.msg.center.common.redis;

import lombok.AccessLevel;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.Collection;
import java.util.concurrent.TimeUnit;

/**
 * @author cold_blade
 * @date 2023/10/16
 * @version 1.0
 */
@Slf4j
@Getter
@Component
@RequiredArgsConstructor
public class RedisUtil {

    /**
     * 使用StringRedisTemplate(，其是RedisTemplate的定制化升级)
     */
    private final StringRedisTemplate redisTemplate;

    private final KeyOps keyOps = new KeyOps();
    private final StringOps stringOps = new StringOps();
    private final LockOps lockOps = new LockOps();

    /**
     * key相关操作
     */
    @NoArgsConstructor(access = AccessLevel.PRIVATE)
    public class KeyOps {

        /**
         * 根据key, 删除redis中的对应key-value
         *  注: 若删除失败, 则返回false。
         *      若redis中，不存在该key, 那么返回的也是false。
         *      所以，不能因为返回了false,就认为redis中一定还存
         *      在该key对应的key-value。
         *
         * @param key 要删除的key
         * @return 删除是否成功
         */
        public Boolean delete(String key) {
            log.info("delete(...) => key= {}", key);
            // 返回值只可能为true/false, 不可能为null
            Boolean result = redisTemplate.delete(key);
            log.info("delete(...) => result= {}", result);
            if (result == null) {
                throw new RedisOpsResultIsNullException();
            }
            return result;
        }

        /**
         * 根据keys, 批量删除key-value
         *
         * 注: 若redis中，不存在对应的key, 那么计数不会加1, 即:
         *     redis中存在的key-value里，有名为a1、a2的key，
         *     删除时，传的集合是a1、a2、a3，那么返回结果为2。
         *
         * @param keys
         *            要删除的key集合
         * @return 删除了的key-value个数
         */
        public Long delete(Collection<String> keys) {
            log.info("delete(...) => keys= {}", keys);
            Long count = redisTemplate.delete(keys);
            log.info("delete(...) => count= {}", count);
            if (count == null) {
                throw new RedisOpsResultIsNullException();
            }
            return count;
        }

        /**
         * redis中是否存在,指定key的key-value
         *
         * @param key 指定的key
         * @return 是否存在对应的key-value
         */
        public Boolean hasKey(String key) {
            log.info("hasKey(...) => key= {}", key);
            Boolean result = redisTemplate.hasKey(key);
            log.info("hasKey(...) => result= {}", result);
            if (result == null) {
                throw new RedisOpsResultIsNullException();
            }
            return result;
        }

        /**
         * 给指定的key对应的key-value设置: 多久过时
         *
         * 注:过时后，redis会自动删除对应的key-value。
         * 注:若key不存在，那么也会返回false。
         *
         * @param key 指定的key
         * @param timeout 过时时间
         * @param unit timeout的单位
         * @return 操作是否成功
         */
        public Boolean expire(String key, Long timeout, TimeUnit unit) {
            log.info("expire(...) => key= {}, timeout= {}, unit= {}", key, timeout, unit);
            Boolean result = redisTemplate.expire(key, timeout, unit);
            log.info("expire(...) => result is= {}", result);
            if (result == null) {
                throw new RedisOpsResultIsNullException();
            }
            return result;
        }
    }

    /**
     * string相关操作
     *
     * 提示: redis中String的数据结构可参考resources/data-structure/String(字符串)的数据结构(示例一).png
     *      redis中String的数据结构可参考resources/data-structure/String(字符串)的数据结构(示例二).png
     */
    @NoArgsConstructor(access = AccessLevel.PRIVATE)
    public class StringOps {

        /**
         * 设置key-value
         * 注: 若已存在相同的key, 那么原来的key-value会被丢弃。
         *
         * @param key key
         * @param value key对应的value
         */
        public void set(String key, String value) {
            log.info("set(...) => key= {}, value= {}", key, value);
            redisTemplate.opsForValue().set(key, value);
        }

        /**
         * 设置key-value
         * 注: 若已存在相同的key, 那么原来的key-value会被丢弃
         *
         * @param key key
         * @param value key对应的value
         * @param timeout 过时时长
         * @param unit timeout的单位
         */
        public void setEx(String key, String value, Long timeout, TimeUnit unit) {
            log.info("setEx(...) => key= {}, value= {}, timeout= {}, unit= {}", key, value, timeout, unit);
            redisTemplate.opsForValue().set(key, value, timeout, unit);
        }

        /**
         * 若不存在key时, 向redis中添加key-value, 返回成功/失败。
         * 若存在，则不作任何操作, 返回false。
         *
         * @param key key
         * @param value key对应的value
         *
         * @return set是否成功
         */
        public Boolean setIfAbsent(String key, String value) {
            log.info("setIfAbsent(...) => key= {}, value= {}", key, value);
            Boolean result = redisTemplate.opsForValue().setIfAbsent(key, value);
            log.info("setIfAbsent(...) => result= {}", result);
            if (result == null) {
                throw new RedisOpsResultIsNullException();
            }
            return result;
        }

        /**
         * 若不存在key时, 向redis中添加一个(具有超时时长的)key-value, 返回成功/失败。
         * 若存在，则不作任何操作, 返回false。
         *
         * @param key key
         * @param value key对应的value
         * @param timeout 超时时长
         * @param unit timeout的单位
         *
         * @return set是否成功
         */
        public Boolean setIfAbsent(String key, String value, Long timeout, TimeUnit unit) {
            log.info("setIfAbsent(...) => key= {}, value= {}, key= {}, value= {}", key, value, timeout, unit);
            Boolean result = redisTemplate.opsForValue().setIfAbsent(key, value, timeout, unit);
            log.info("setIfAbsent(...) => result= {}", result);
            if (result == null) {
                throw new RedisOpsResultIsNullException();
            }
            return result;
        }

        /**
         * 根据key，获取到对应的value值
         * 注: 若key不存在， 则返回null。
         *
         * @param key key-value对应的key
         * @return 该key对应的值。
         */
        public String get(String key) {
            log.info("get(...) => key= {}", key);
            String result = redisTemplate.opsForValue().get(key);
            log.info("get(...) => result= {} ", result);
            return result;
        }
    }

    /**
     * redis分布式锁.
     *
     * 使用方式(示例):
     * 			Boolean flag = false;
     * 			String lockName = "sichuan:mianyang:fucheng:ds";
     * 			String lockValue = UUID.randomUUID().toString();
     * 			try {
     * 		        //	非阻塞获取(锁的最大存活时间采用默认值)
     * 				flag = RedisUtil.LockOps.getLock(lockName, lockValue);
     * 				//	非阻塞获取e.g.
     * 				flag = RedisUtil.LockOps.getLock(lockName, lockValue, 3, TimeUnit.SECONDS);
     * 			    // 阻塞获取(锁的最大存活时间采用默认值)
     * 		        flag = RedisUtil.LockOps.getLockUntilTimeout(lockName, lockValue, 2000);
     * 		        // 阻塞获取e.g.
     * 		        flag = RedisUtil.LockOps.getLockUntilTimeout(lockName, lockValue, 2, TimeUnit.SECONDS, 2000);
     * 				// your logic
     * 			    //	...
     *          } finally {
     * 				if (flag) {
     * 					RedisUtil.LockOps.releaseLock(lockName, lockValue);
     *              }
     *          }
     */
    @NoArgsConstructor(access = AccessLevel.PRIVATE)
    public class LockOps {

        /** lua脚本, 保证 释放锁脚本 的原子性(以避免, 并发场景下, 释放了别人的锁) */
        private final String RELEASE_LOCK_LUA = "if redis.call('get',KEYS[1]) == ARGV[1] "
                + "then "
                + "    return redis.call('del',KEYS[1]) "
                + "else "
                + "    return 0 "
                + "end ";

        /** 分布式锁默认(最大)存活时长 */
        public final long DEFAULT_LOCK_TIMEOUT = 3;

        /** DEFAULT_LOCK_TIMEOUT的单位 */
        public final TimeUnit DEFAULT_TIMEOUT_UNIT = TimeUnit.SECONDS;

        /**
         * 获取(分布式)锁.
         * 注: 获取结果是即时返回的、是非阻塞的。
         *
         * @see LockOps#getLock(String, String, Long, TimeUnit)
         */
        public Boolean getLock(final String key, final String value) {
            return getLock(key, value, DEFAULT_LOCK_TIMEOUT, DEFAULT_TIMEOUT_UNIT);
        }

        /**
         * 获取(分布式)锁。
         * 若成功, 则直接返回;
         * 若失败, 则进行重试, 直到成功 或 超时为止。
         * 注: 获取结果是阻塞的， 要么成功, 要么超时, 才返回。
         *
         * @param retryTimeoutLimit 重试的超时时长(ms)
         * 其它参数可详见:
         *    @see LockOps#getLock(String, String, Long, TimeUnit)
         *
         * @return 是否成功
         */
        public Boolean getLockUntilTimeout(final String key, final String value, final Long retryTimeoutLimit) {
            return getLockUntilTimeout(key, value, DEFAULT_LOCK_TIMEOUT, DEFAULT_TIMEOUT_UNIT,
                    retryTimeoutLimit);
        }

        /**
         * 获取(分布式)锁。
         * 若成功, 则直接返回;
         * 若失败, 则进行重试, 直到成功 或 超时为止。
         * 注: 获取结果是阻塞的， 要么成功, 要么超时, 才返回。
         *
         * @param retryTimeoutLimit 重试的超时时长(ms)
         * 其它参数可详见:
         *    @see LockOps#getLock(String, String, Long, TimeUnit, Boolean)
         *
         * @return 是否成功
         */
        public Boolean getLockUntilTimeout(final String key, final String value,
                                           final Long timeout, final TimeUnit unit,
                                           final Long retryTimeoutLimit) {
            log.info("getLockUntilTimeout(...) => key= {}, value= {}, timeout= {}, unit= {}, "
                    + "retryTimeoutLimit= {}ms", key, value, timeout, unit, retryTimeoutLimit);
            long startTime = Instant.now().toEpochMilli();
            long now = startTime;
            do {
                try {
                    Boolean alreadyGotLock = getLock(key, value, timeout, unit, false);
                    if (alreadyGotLock) {
                        log.info("getLockUntilTimeout(...) => consume time= {}ms, result= true", now - startTime);
                        return true;
                    }
                } catch (Exception e) {
                    log.warn("getLockUntilTimeout(...) => try to get lock failure!", e);
                }
                now = Instant.now().toEpochMilli();
            } while (now < startTime + retryTimeoutLimit);
            log.info("getLockUntilTimeout(...) => consume time= {}ms, result= false", now - startTime);
            return false;
        }

        /**
         * 获取(分布式)锁
         * 注: 获取结果是即时返回的、是非阻塞的。
         *
         * @see LockOps#getLock(String, String, Long, TimeUnit, Boolean)
         */
        public Boolean getLock(final String key, final String value, final Long timeout, final TimeUnit unit) {
            return getLock(key, value, timeout, unit, true);
        }

        /**
         * 获取(分布式)锁
         * 注: 获取结果是即时返回的、是非阻塞的。
         *
         * @param key 锁名
         * @param value
         *            锁名对应的value
         *            注: value一般采用全局唯一的值， 如: requestId、uuid等。
         *               这样， 释放锁的时候, 可以再次验证value值,
         *               保证自己上的锁只能被自己释放, 而不会被别人释放。
         *               当然, 如果锁超时时, 会被redis自动删除释放。
         * @param timeout
         *            锁的(最大)存活时长
         *            注: 一般的， 获取锁与释放锁 都是成对使用的, 在锁在达到(最大)存活时长之前，都会被主动释放。
         *                但是在某些情况下(如:程序获取锁后,释放锁前,崩了),锁得不到释放, 这时就需要等锁过
         *                了(最大)存活时长后，被redis自动删除清理了。这样就能保证redis中不会留下死数据。
         * @param unit timeout的单位
         * @param recordLog 是否记录日志
         *
         * @return 是否成功
         */
        public Boolean getLock(final String key, final String value,
                               final Long timeout, final TimeUnit unit,
                               Boolean recordLog) {
            if (recordLog) {
                log.info("getLock(...) => key= {}, value= {}, timeout= {}, unit= {}, recordLog= {}",
                        key, value, timeout, unit, recordLog);
            }
            Boolean result = redisTemplate.execute((RedisConnection connection) ->
                    connection.set(key.getBytes(StandardCharsets.UTF_8),
                            value.getBytes(StandardCharsets.UTF_8),
                            Expiration.seconds(unit.toSeconds(timeout)),
                            RedisStringCommands.SetOption.SET_IF_ABSENT)
            );
            if (recordLog) {
                log.info("getLock(...) => result= {}", result);
            }
            if (result == null) {
                throw new RedisOpsResultIsNullException();
            }
            return result;
        }

        /**
         * 释放(分布式)锁
         * 注: 此方式能(通过value的唯一性)保证: 自己加的锁, 只能被自己释放。
         * 注: 锁超时时, 也会被redis自动删除释放。
         *
         * @param key 锁名
         * @param value　锁名对应的value
         *
         * @return 释放锁是否成功
         */
        public Boolean releaseLock(final String key, final String value) {
            log.info("releaseLock(...) => key= {}, lockValue= {}", key, value);
            Boolean result = redisTemplate.execute((RedisConnection connection) ->
                    connection.eval(RELEASE_LOCK_LUA.getBytes(),
                            ReturnType.BOOLEAN, 1,
                            key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8))
            );
            log.info("releaseLock(...) => result= {}", result);
            if (result == null) {
                throw new RedisOpsResultIsNullException();
            }
            return result;
        }
    }

    public static class RedisOpsResultIsNullException extends NullPointerException {

        private static final long serialVersionUID = 7727166544003942512L;

        public RedisOpsResultIsNullException() {
            super();
        }

        public RedisOpsResultIsNullException(String message) {
            super(message);
        }
    }
}
