import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;

import java.time.Duration;
import java.util.function.Supplier;

/**
 * A simple implementation for Distributed Lock based on Redis.
 * <p>
 * The implementation relies on Redis SETNX (SET if Not eXists).
 * if one process acquires the lock, the other processes won't wait, and simply an exception will be thrown.
 * <p>
 * Usually, you would use the Database as a lock mechanism (using unique keys for example).
 * However, sometimes, you cannot use the database (for any reason), or you need a quick solution, hence you can use this class.
 *
 * @see <a href="https://redis.com/glossary/redis-lock/">Redis Lock</a>
 */
@Slf4j
public class Lock {

    private static final String KEY_PREFIX = "idempotent-";
    private static final Options DEFAULT_OPTIONS = new Options();

    private static RedisTemplate<String, Object> redisTemplate;

    /**
     * Run with default options
     */
    public static void run(String key, Runnable action) {
        run(key, DEFAULT_OPTIONS, action);
    }

    public static void run(String key, Options options, Runnable action) {
        run(key, options, runnableToSupplier(action));
    }

    /**
     * Run with default options
     */
    public static <T> T run(String key, Supplier<T> action) {
        return run(key, DEFAULT_OPTIONS, action);
    }

    public static <T> T run(String key, Options options, Supplier<T> action) {
        initRedisTemplate();

        writeKey(key, options.duration, options.errorKey);

        try {
            return action.get();
        } finally {
            if (!options.keepLock) {
                removeKey(key);
            }
        }
    }

    /**
     * Use {@link Lock#options()} to create default {@link Options} object.
     */
    @NoArgsConstructor(access = AccessLevel.PRIVATE)
    public static class Options {
        private Duration duration = Duration.ofMinutes(5);
        private boolean keepLock = false;
        private String errorKey = "operation_in_progress";

        /**
         * the maximum duration to lock, default is 5 minutes
         */
        public Options duration(Duration d) {
            this.duration = d;
            return this;
        }

        /**
         * keep the lock for the whole duration regardless of whether the execution is done or not, default is false.
         */
        public Options keepLock(boolean b) {
            this.keepLock = b;
            return this;
        }

        /**
         * error key to thrown in case of operation is in progress, default is "operation_in_progress".
         */
        public Options errorKey(String key) {
            this.errorKey = key;
            return this;
        }
    }

    public static Options options() {
        return new Options();
    }

    @SuppressWarnings({"unchecked"})
    private static void initRedisTemplate() {
        if (redisTemplate == null) {
            redisTemplate = AppContextUtil.getBean(RedisTemplate.class, String.class, Object.class);
        }
    }

    private static void writeKey(String key, Duration duration, String errorMessage) {
        log.trace("write key: {}", key);

        var ok = redisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + key, "1", duration);
        if (ok == null) {
            throw new RuntimeException("Lock cannot used inside Redis pipeline/transaction");
        }
        if (!ok) {
            throw new BusinessException(errorMessage, "key", key);
        }
    }

    private static void removeKey(String key) {
        log.trace("remove key: {}", key);
        redisTemplate.opsForValue().getAndDelete(KEY_PREFIX + key);
    }

    private static Supplier<Void> runnableToSupplier(Runnable action) {
        return () -> {
            action.run();
            return null;
        };
    }
}