/*
 * Copyright 2017-present the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.springframework.data.redis.cache;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

import org.jspecify.annotations.Nullable;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.util.Assert;

/**
 * {@link RedisCacheWriter} provides low-level access to Redis commands ({@code SET, SETNX, GET, EXPIRE,...}) used for
 * caching.
 * <p>
 * The {@link RedisCacheWriter} may be shared by multiple cache implementations and is responsible for reading/writing
 * binary data from/to Redis. The implementation honors potential cache lock flags that might be set.
 * <p>
 * The default {@link RedisCacheWriter} implementation can be customized with {@link BatchStrategy} to tune performance
 * behavior.
 *
 * @author Christoph Strobl
 * @author Mark Paluch
 * @author John Blum
 * @since 2.0
 */
public interface RedisCacheWriter extends CacheStatisticsProvider {

	/**
	 * Create new {@link RedisCacheWriter} configure it through {@link RedisCacheWriterConfigurer}. The cache writer
	 * defaults does not lock the cache by default using {@link BatchStrategies#keys()}.
	 *
	 * @param connectionFactory the connection factory to use.
	 * @param configurerConsumer a configuration function that configures {@link RedisCacheWriterConfigurer}.
	 * @return new instance of {@link DefaultRedisCacheWriter}.
	 * @since 4.0
	 */
	static RedisCacheWriter create(RedisConnectionFactory connectionFactory,
			Consumer<RedisCacheWriterConfigurer> configurerConsumer) {
		return DefaultRedisCacheWriter.create(connectionFactory, configurerConsumer);
	}

	/**
	 * Create new {@link RedisCacheWriter} without locking behavior using {@link BatchStrategies#keys()}.
	 *
	 * @param connectionFactory must not be {@literal null}.
	 * @return new instance of {@link DefaultRedisCacheWriter}.
	 */
	static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
		return nonLockingRedisCacheWriter(connectionFactory, BatchStrategies.keys());
	}

	/**
	 * Create new {@link RedisCacheWriter} without locking behavior.
	 *
	 * @param connectionFactory must not be {@literal null}.
	 * @param batchStrategy must not be {@literal null}.
	 * @return new instance of {@link DefaultRedisCacheWriter}.
	 * @since 2.6
	 */
	static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connectionFactory,
			BatchStrategy batchStrategy) {
		return create(connectionFactory, config -> config.batchStrategy(batchStrategy));
	}

	/**
	 * Create new {@link RedisCacheWriter} with locking behavior using {@link BatchStrategies#keys()}.
	 *
	 * @param connectionFactory must not be {@literal null}.
	 * @return new instance of {@link DefaultRedisCacheWriter}.
	 */
	static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
		return lockingRedisCacheWriter(connectionFactory, BatchStrategies.keys());
	}

	/**
	 * Create new {@link RedisCacheWriter} with locking behavior.
	 *
	 * @param connectionFactory must not be {@literal null}.
	 * @param batchStrategy must not be {@literal null}.
	 * @return new instance of {@link DefaultRedisCacheWriter}.
	 * @since 2.6
	 */
	static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory,
			BatchStrategy batchStrategy) {

		return create(connectionFactory,
				it -> it.batchStrategy(batchStrategy).cacheLocking(CacheLockingConfigurer::enable));
	}

	/**
	 * Create new {@link RedisCacheWriter} with locking behavior.
	 *
	 * @param connectionFactory must not be {@literal null}.
	 * @param sleepTime sleep time between lock access attempts, must not be {@literal null}.
	 * @param lockTtlFunction TTL function to compute the Lock TTL. The function is called with contextual keys and values
	 *          (such as the cache name on cleanup or the actual key/value on put requests); must not be {@literal null}.
	 * @param batchStrategy must not be {@literal null}.
	 * @return new instance of {@link DefaultRedisCacheWriter}.
	 * @since 3.2
	 */
	static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime,
			TtlFunction lockTtlFunction, BatchStrategy batchStrategy) {

		return create(connectionFactory, it -> it.batchStrategy(batchStrategy)
				.enableLocking(locking -> locking.sleepTime(sleepTime).lockTimeout(lockTtlFunction)));
	}

	/**
	 * Get the binary value representation from Redis stored for the given key.
	 *
	 * @param name must not be {@literal null}.
	 * @param key must not be {@literal null}.
	 * @return {@literal null} if key does not exist.
	 * @see #get(String, byte[], Duration)
	 */
	byte @Nullable [] get(String name, byte[] key);

	/**
	 * Get the binary value representation from Redis stored for the given key and set the given {@link Duration TTL
	 * expiration} for the cache entry.
	 *
	 * @param name must not be {@literal null}.
	 * @param key must not be {@literal null}.
	 * @param ttl {@link Duration} specifying the {@literal expiration timeout} for the cache entry.
	 * @return {@literal null} if key does not exist or has {@literal expired}.
	 */
	default byte @Nullable [] get(String name, byte[] key, @Nullable Duration ttl) {
		return get(name, key);
	}

	/**
	 * Get the binary value representation from Redis stored for the given key and set the given {@link Duration TTL
	 * expiration} for the cache entry, obtaining the value from {@code valueLoader} if necessary.
	 * <p>
	 * If possible (and configured for locking), implementations should ensure that the loading operation is synchronized
	 * so that the specified {@code valueLoader} is only called once in case of concurrent access on the same key.
	 *
	 * @param name must not be {@literal null}.
	 * @param key must not be {@literal null}.
	 * @param valueLoader value loader that creates the value if the cache lookup has been not successful.
	 * @param ttl {@link Duration} specifying the {@literal expiration timeout} for the cache entry.
	 * @param timeToIdleEnabled {@literal true} to enable Time to Idle when retrieving the value.
	 * @since 3.4
	 */
	default byte[] get(String name, byte[] key, Supplier<byte[]> valueLoader, @Nullable Duration ttl,
			boolean timeToIdleEnabled) {

		byte[] bytes = timeToIdleEnabled ? get(name, key, ttl) : get(name, key);

		if (bytes == null) {
			bytes = valueLoader.get();
			put(name, key, bytes, ttl);
		}

		return bytes;
	}

	/**
	 * Determines whether the asynchronous {@link #retrieve(String, byte[])} and
	 * {@link #retrieve(String, byte[], Duration)} cache operations are supported by the implementation.
	 * <p>
	 * The main factor for whether the {@literal retrieve} operation can be supported will primarily be determined by the
	 * Redis driver in use at runtime.
	 * <p>
	 * Returns {@literal false} by default. This will have an effect of {@link RedisCache#retrieve(Object)} and
	 * {@link RedisCache#retrieve(Object, Supplier)} throwing an {@link UnsupportedOperationException}.
	 *
	 * @return {@literal true} if asynchronous {@literal retrieve} operations are supported by the implementation.
	 * @since 3.2
	 */
	default boolean supportsAsyncRetrieve() {
		return false;
	}

	/**
	 * Asynchronously retrieves the {@link CompletableFuture value} to which the {@link RedisCache} maps the given
	 * {@code byte[] key}.
	 * <p>
	 * This operation is non-blocking.
	 *
	 * @param name {@link String} with the name of the {@link RedisCache}.
	 * @param key {@code byte[] key} mapped to the {@link CompletableFuture value} in the {@link RedisCache}.
	 * @return the {@link CompletableFuture value} to which the {@link RedisCache} maps the given {@code byte[] key}.
	 * @see #retrieve(String, byte[], Duration)
	 * @since 3.2
	 */
	default CompletableFuture<byte[]> retrieve(String name, byte[] key) {
		return retrieve(name, key, null);
	}

	/**
	 * Asynchronously retrieves the {@link CompletableFuture value} to which the {@link RedisCache} maps the given
	 * {@code byte[] key} setting the {@link Duration TTL expiration} for the cache entry.
	 * <p>
	 * This operation is non-blocking.
	 *
	 * @param name {@link String} with the name of the {@link RedisCache}.
	 * @param key {@code byte[] key} mapped to the {@link CompletableFuture value} in the {@link RedisCache}.
	 * @param ttl {@link Duration} specifying the {@literal expiration timeout} for the cache entry.
	 * @return the {@link CompletableFuture value} to which the {@link RedisCache} maps the given {@code byte[] key}.
	 * @since 3.2
	 */
	CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Duration ttl);

	/**
	 * Write the given key/value pair to Redis and set the expiration time if defined.
	 *
	 * @param name cache name must not be {@literal null}.
	 * @param key key for the cache entry. Must not be {@literal null}.
	 * @param value value stored for the key. Must not be {@literal null}.
	 * @param ttl optional expiration time. Can be {@literal null}.
	 */
	void put(String name, byte[] key, byte[] value, @Nullable Duration ttl);

	/**
	 * Store the given key/value pair asynchronously to Redis and set the expiration time if defined.
	 * <p>
	 * This operation is non-blocking.
	 *
	 * @param name cache name must not be {@literal null}.
	 * @param key key for the cache entry. Must not be {@literal null}.
	 * @param value value stored for the key. Must not be {@literal null}.
	 * @param ttl optional expiration time. Can be {@literal null}.
	 * @since 3.2
	 */
	CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nullable Duration ttl);

	/**
	 * Write the given value to Redis if the key does not already exist.
	 *
	 * @param name cache name must not be {@literal null}.
	 * @param key key for the cache entry. Must not be {@literal null}.
	 * @param value value stored for the key. Must not be {@literal null}.
	 * @param ttl optional expiration time. Can be {@literal null}.
	 * @return {@literal null} if the value has been written, the value stored for the key if it already exists.
	 */
	byte @Nullable [] putIfAbsent(String name, byte[] key, byte[] value, @Nullable Duration ttl);

	/**
	 * Remove the given key from Redis.
	 * <p>
	 * Actual eviction may be performed in an asynchronous or deferred fashion, with subsequent lookups possibly still
	 * seeing the entry.
	 *
	 * @param name cache name must not be {@literal null}.
	 * @param key key for the cache entry. Must not be {@literal null}.
	 * @deprecated since 4.0 in favor of {@link #evict(String, byte[])}
	 */
	@Deprecated(since = "4.0", forRemoval = true)
	default void remove(String name, byte[] key) {
		evict(name, key);
	}

	/**
	 * Remove the given key from Redis.
	 * <p>
	 * Actual eviction may be performed in an asynchronous or deferred fashion, with subsequent lookups possibly still
	 * seeing the entry.
	 *
	 * @param name cache name must not be {@literal null}.
	 * @param key key for the cache entry. Must not be {@literal null}.
	 * @since 4.0
	 */
	void evict(String name, byte[] key);

	/**
	 * Remove the given key from Redis if it is present, expecting the key to be immediately invisible for subsequent
	 * lookups.
	 *
	 * @param name cache name must not be {@literal null}.
	 * @param key key for the cache entry. Must not be {@literal null}.
	 * @return {@code true} if the cache was known to have a mapping for this key before, {@code false} if it did not (or
	 *         if prior presence could not be determined).
	 * @since 4.0
	 */
	default boolean evictIfPresent(String name, byte[] key) {
		evict(name, key);
		return false;
	}

	/**
	 * Remove all keys following the given pattern.
	 * <p>
	 * Actual clearing may be performed in an asynchronous or deferred fashion, with subsequent lookups possibly still
	 * seeing the entries.
	 *
	 * @param name cache name must not be {@literal null}.
	 * @param pattern pattern for the keys to remove. Must not be {@literal null}.
	 * @deprecated since 4.0 in favor of {@link #clear(String, byte[])}
	 */
	@Deprecated(since = "4.0", forRemoval = true)
	default void clean(String name, byte[] pattern) {
		clear(name, pattern);
	}

	/**
	 * Remove all keys following the given pattern.
	 * <p>
	 * Actual clearing may be performed in an asynchronous or deferred fashion, with subsequent lookups possibly still
	 * seeing the entries.
	 *
	 * @param name cache name must not be {@literal null}.
	 * @param pattern pattern for the keys to remove. Must not be {@literal null}.
	 * @since 4.0
	 */
	void clear(String name, byte[] pattern);

	/**
	 * Remove all keys following the given pattern expecting all entries to be immediately invisible for subsequent
	 * lookups.
	 *
	 * @param name cache name must not be {@literal null}.
	 * @param pattern pattern for the keys to remove. Must not be {@literal null}.
	 * @return {@code true} if the cache was known to have mappings before, {@code false} if it did not (or if prior
	 *         presence of entries could not be determined).
	 */
	default boolean invalidate(String name, byte[] pattern) {
		clear(name, pattern);
		return false;
	}

	/**
	 * Reset all statistics counters and gauges for this cache.
	 *
	 * @since 2.4
	 */
	void clearStatistics(String name);

	/**
	 * Executes the given {@link Function} with a {@link RedisConnection}.
	 *
	 * @param callback the callback action to invoke with a connection.
	 * @return return value of the callback.
	 * @param <T>
	 * @throws UnsupportedOperationException if the cache writer does not support direct access to
	 *           {@link RedisConnection}.
	 * @since 4.1
	 */
	default <T extends @Nullable Object> T execute(Function<RedisConnection, T> callback) {
		throw new UnsupportedOperationException("execute(...) is not supported by this RedisCacheWriter");
	}

	/**
	 * Obtain a {@link RedisCacheWriter} using the given {@link CacheStatisticsCollector} to collect metrics.
	 *
	 * @param cacheStatisticsCollector must not be {@literal null}.
	 * @return new instance of {@link RedisCacheWriter}.
	 */
	RedisCacheWriter withStatisticsCollector(CacheStatisticsCollector cacheStatisticsCollector);

	/**
	 * Interface that allows for configuring a {@link RedisCacheWriter}.
	 *
	 * @author Mark Paluch
	 * @since 4.0
	 */
	interface RedisCacheWriterConfigurer {

		/**
		 * Configure the {@link CacheStatisticsCollector} to use. This is useful for plugging in and/or customizing
		 * statistics collection.
		 */
		default RedisCacheWriterConfigurer collectStatistics() {
			return collectStatistics(CacheStatisticsCollector.create());
		}

		/**
		 * Configure the {@link CacheStatisticsCollector} to use. This is useful for plugging in and/or customizing
		 * statistics collection.
		 * <p>
		 * If no statistics collector is specified, no statistics will be recorded. Statistics collection can be
		 * reconfigured on the built RedisCacheWriter by invoking
		 * {@code RedisCacheWriter#withStatisticsCollector(CacheStatisticsCollector)}.
		 *
		 * @param cacheStatisticsCollector the statistics collector to use.
		 */
		RedisCacheWriterConfigurer collectStatistics(CacheStatisticsCollector cacheStatisticsCollector);

		/**
		 * Configure the {@link BatchStrategy} when clearing the cache (i.e. bulk removal of cache keys).
		 * <p>
		 * If no batch strategy is specified, the RedisCacheWriter uses {@link BatchStrategies#keys()};
		 *
		 * @param batchStrategy the batch strategy to use.
		 */
		RedisCacheWriterConfigurer batchStrategy(BatchStrategy batchStrategy);

		/**
		 * Enable cache locking to synchronize cache access across multiple cache instances.
		 */
		default RedisCacheWriterConfigurer enableLocking() {
			return cacheLocking(it -> it.enable(config -> {}));
		}

		/**
		 * Enable cache locking to synchronize cache access across multiple cache instances.
		 *
		 * @param configurerConsumer a configuration function that configures {@link CacheLockingConfiguration}.
		 */
		default RedisCacheWriterConfigurer enableLocking(Consumer<CacheLockingConfiguration> configurerConsumer) {
			return cacheLocking(it -> it.enable(configurerConsumer));
		}

		/**
		 * Configure cache locking to synchronize cache access across multiple cache instances.
		 *
		 * @param configurerConsumer a configuration function that configures {@link CacheLockingConfigurer}.
		 */
		RedisCacheWriterConfigurer cacheLocking(Consumer<CacheLockingConfigurer> configurerConsumer);

		/**
		 * Use immediate writes (i.e. write operations such as
		 * {@link RedisCacheWriter#put(String, byte[], byte[], Duration)} or {@link #clear(String, byte[])}) shall apply
		 * immediately.
		 * <p>
		 * Several {@link org.springframework.cache.Cache} operations can be performed asynchronously or deferred and this
		 * is the default behavior for {@link RedisCacheWriter}. Enable immediate writes in case a particular cache requires
		 * stronger consistency (i.e. Cache writes must be visible immediately).
		 * <p>
		 * When using a {@link org.springframework.data.redis.connection.ReactiveRedisConnectionFactory reactive Redis
		 * driver}, immediate writes lead to blocking.
		 */
		default RedisCacheWriterConfigurer immediateWrites() {
			return immediateWrites(true);
		}

		/**
		 * Configure whether to use immediate writes (i.e. write operations such as
		 * {@link RedisCacheWriter#put(String, byte[], byte[], Duration)} or {@link #clear(String, byte[])}) shall apply
		 * immediately.
		 * <p>
		 * Several {@link org.springframework.cache.Cache} operations can be performed asynchronously or deferred and this
		 * is the default behavior for {@link RedisCacheWriter}. Enable immediate writes in case a particular cache requires
		 * stronger consistency (i.e. Cache writes must be visible immediately).
		 * <p>
		 * When using a {@link org.springframework.data.redis.connection.ReactiveRedisConnectionFactory reactive Redis
		 * driver}, immediate writes lead to blocking.
		 *
		 * @param enableImmediateWrites whether write operations must be visible immediately.
		 */
		RedisCacheWriterConfigurer immediateWrites(boolean enableImmediateWrites);

	}

	/**
	 * Interface that allows for configuring cache locking.
	 *
	 * @author Mark Paluch
	 * @since 4.0
	 */
	interface CacheLockingConfigurer {

		/**
		 * Disable cache locking (default).
		 */
		void disable();

		/**
		 * Enable cache locking with a default sleep time of {@code 50 milliseconds} and persistent lock keys.
		 */
		default void enable() {
			enable(it -> {});
		}

		/**
		 * Enable cache locking.
		 */
		void enable(Consumer<CacheLockingConfiguration> configurationConsumer);

	}

	/**
	 * Interface that allows for configuring cache locking options.
	 *
	 * @author Mark Paluch
	 * @since 4.0
	 */
	interface CacheLockingConfiguration {

		/**
		 * Configure the sleep time between cache lock checks. Sleep time is applied to reattempt lock checks if a cache key
		 * is locked.
		 *
		 * @param sleepTime the sleep time, must not be {@literal null} and must be greater {@link Duration#ZERO}.
		 */
		CacheLockingConfiguration sleepTime(Duration sleepTime);

		/**
		 * Configure a {@link TtlFunction} to compute the lock timeout.
		 * <p>
		 * If no TTL function is specified, the RedisCacheWriter persistent lock keys. Persistent lock keys need to be
		 * removed in case of failures (e.g. Redis crashes before a lock key is removed). Expiring lock keys can become
		 * subject to GC timing if lock keys expire while a garbage collection halts the JVM.
		 *
		 * @param ttlFunction the lock timeout function.
		 */
		CacheLockingConfiguration lockTimeout(TtlFunction ttlFunction);

	}

	/**
	 * Function to compute the time to live from the cache {@code key} and {@code value}.
	 *
	 * @author Mark Paluch
	 * @since 3.2
	 */
	@FunctionalInterface
	interface TtlFunction {

		Duration NO_EXPIRATION = Duration.ZERO;

		/**
		 * Creates a {@literal Singleton} {@link TtlFunction} using the given {@link Duration}.
		 *
		 * @param duration the time to live. Can be {@link Duration#ZERO} for persistent values (i.e. cache entry does not
		 *          expire).
		 * @return a singleton {@link TtlFunction} using {@link Duration}.
		 */
		static TtlFunction just(Duration duration) {

			Assert.notNull(duration, "TTL Duration must not be null");

			return new FixedDurationTtlFunction(duration);
		}

		/**
		 * Returns a {@link TtlFunction} to create persistent entries that do not expire.
		 *
		 * @return a {@link TtlFunction} to create persistent entries that do not expire.
		 */
		static TtlFunction persistent() {
			return just(NO_EXPIRATION);
		}

		/**
		 * Compute a {@link Duration time-to-live (TTL)} using the cache {@code key} and {@code value}.
		 * <p>
		 * The {@link Duration time-to-live (TTL)} is computed on each write operation. Redis uses millisecond granularity
		 * for timeouts. Any more granular values (e.g. micros or nanos) are not considered and will be truncated due to
		 * rounding. Returning {@link Duration#ZERO}, or a value less than {@code Duration.ofMillis(1)}, results in a
		 * persistent value that does not expire.
		 *
		 * @param key the cache key.
		 * @param value the cache value. Can be {@literal null} if the cache supports {@literal null} value caching.
		 * @return the computed {@link Duration time-to-live (TTL)}. Can be {@link Duration#ZERO} for persistent values
		 *         (i.e. cache entry does not expire).
		 */
		Duration getTimeToLive(Object key, @Nullable Object value);

	}

}