package com.cygnus.cloud.cache;

import java.time.Duration;

import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.stereotype.Service;

import reactor.core.publisher.Mono;

@Service
public class ReactiveCacheService {

    private final ReactiveStringRedisTemplate redis;
    private final CacheProperties properties;

    public ReactiveCacheService(ReactiveStringRedisTemplate redis, CacheProperties properties) {
        this.redis = redis;
        this.properties = properties;
    }

    public Mono<String> get(String namespace, String key) {
        return redis.opsForValue().get(cacheKey(namespace, key));
    }

    public Mono<Boolean> put(String namespace, String key, String value) {
        return put(namespace, key, value, properties.defaultTtl());
    }

    public Mono<Boolean> put(String namespace, String key, String value, Duration ttl) {
        return redis.opsForValue().set(cacheKey(namespace, key), value, ttl);
    }

    public Mono<Boolean> putIfAbsent(String namespace, String key, String value, Duration ttl) {
        return redis.opsForValue().setIfAbsent(cacheKey(namespace, key), value, ttl);
    }

    public Mono<Boolean> evict(String namespace, String key) {
        return redis.delete(cacheKey(namespace, key)).map(deleted -> deleted > 0);
    }

    public Mono<Long> increment(String namespace, String key, Duration ttl) {
        String fullKey = cacheKey(namespace, key);
        return redis.opsForValue()
                .increment(fullKey)
                .flatMap(count -> count == 1
                        ? redis.expire(fullKey, ttl).thenReturn(count)
                        : Mono.just(count));
    }

    private String cacheKey(String namespace, String key) {
        return properties.keyPrefix() + ':' + namespace + ':' + key;
    }
}