tasteby/backend-java/src/main/java/com/tasteby/service/CacheService.java

package com.tasteby.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

@Service
public class CacheService {

    private static final Logger log = LoggerFactory.getLogger(CacheService.class);
    private static final String PREFIX = "tasteby:";
    private static final String SCAN_PATTERN = PREFIX + "*";
    private static final int SCAN_BATCH = 500;

    private final StringRedisTemplate redis;
    private final ObjectMapper mapper;
    private final Duration ttl;

    // #336 — disabled/errorCount/lastError는 헬스체크와 다른 호출 스레드 사이에서 안전하게 공유.
    private volatile boolean disabled = false;
    private final AtomicLong errorCount = new AtomicLong(0);
    private volatile String lastError = null;

    public CacheService(StringRedisTemplate redis, ObjectMapper mapper,
                        @Value("${app.cache.ttl-seconds:600}") int ttlSeconds) {
        this.redis = redis;
        this.mapper = mapper;
        this.ttl = Duration.ofSeconds(ttlSeconds);
        this.disabled = !pingOk();
        if (!disabled) log.info("Redis connected");
    }

    public String makeKey(String... parts) {
        if (parts == null || parts.length == 0) {
            throw new IllegalArgumentException("makeKey requires at least one part");
        }
        for (String p : parts) {
            if (p == null) throw new IllegalArgumentException("makeKey parts must not be null");
        }
        return PREFIX + String.join(":", parts);
    }

    public <T> T get(String key, Class<T> type) {
        if (disabled) return null;
        try {
            String val = redis.opsForValue().get(key);
            if (val != null) {
                return mapper.readValue(val, type);
            }
        } catch (Exception e) {
            recordError("get", e);
        }
        return null;
    }

    public String getRaw(String key) {
        if (disabled) return null;
        try {
            return redis.opsForValue().get(key);
        } catch (Exception e) {
            recordError("getRaw", e);
            return null;
        }
    }

    public void set(String key, Object value) {
        if (disabled) return;
        try {
            String json = mapper.writeValueAsString(value);
            redis.opsForValue().set(key, json, ttl);
        } catch (JsonProcessingException e) {
            recordError("set:serialize", e);
        } catch (Exception e) {
            recordError("set", e);
        }
    }

    /**
     * #336 — KEYS 블로킹 명령 대체.
     * SCAN으로 cursor 순회 후 UNLINK(논블로킹 삭제)로 일괄 삭제.
     */
    public void flush() {
        if (disabled) return;
        Integer count = redis.execute((org.springframework.data.redis.core.RedisCallback<Integer>) conn -> {
            List<byte[]> batch = new ArrayList<>(SCAN_BATCH);
            int deleted = 0;
            try (Cursor<byte[]> cursor = conn.keyCommands().scan(
                    ScanOptions.scanOptions().match(SCAN_PATTERN).count(SCAN_BATCH).build())) {
                while (cursor.hasNext()) {
                    batch.add(cursor.next());
                    if (batch.size() >= SCAN_BATCH) {
                        deleted += unlinkBatch(conn, batch);
                        batch.clear();
                    }
                }
                if (!batch.isEmpty()) {
                    deleted += unlinkBatch(conn, batch);
                }
            } catch (Exception e) {
                recordError("flush:scan", e);
            }
            return deleted;
        });
        log.info("Cache flushed ({} keys via SCAN+UNLINK)", count == null ? 0 : count);
    }

    private int unlinkBatch(org.springframework.data.redis.connection.RedisConnection conn, List<byte[]> keys) {
        try {
            Long n = conn.keyCommands().unlink(keys.toArray(new byte[0][]));
            return n == null ? 0 : n.intValue();
        } catch (Exception e) {
            // UNLINK 미지원 환경 대비 DEL 폴백
            recordError("flush:unlink", e);
            try {
                Long n = conn.keyCommands().del(keys.toArray(new byte[0][]));
                return n == null ? 0 : n.intValue();
            } catch (Exception delErr) {
                recordError("flush:del", delErr);
                return 0;
            }
        }
    }

    public void del(String key) {
        if (disabled) return;
        try {
            redis.delete(key);
        } catch (Exception e) {
            recordError("del", e);
        }
    }

    /**
     * #336 — Redis 다운 → disabled=true, 재기동되면 자동으로 disabled=false.
     * 30초마다 ping 한 번(<1ms)이라 부하 미미.
     */
    @Scheduled(fixedDelay = 30_000L)
    public void checkHealth() {
        boolean ok = pingOk();
        if (ok && disabled) {
            disabled = false;
            log.info("Redis recovered, caching re-enabled");
        } else if (!ok && !disabled) {
            disabled = true;
            log.warn("Redis lost, caching disabled");
        }
    }

    private boolean pingOk() {
        RedisConnectionFactory factory = redis.getConnectionFactory();
        if (factory == null) return false;
        try (var conn = factory.getConnection()) {
            conn.ping();
            return true;
        } catch (Exception e) {
            lastError = "ping: " + e.getMessage();
            return false;
        }
    }

    private void recordError(String op, Exception e) {
        long n = errorCount.incrementAndGet();
        String msg = e.getMessage();
        lastError = op + ": " + (msg == null ? e.getClass().getSimpleName() : msg);
        // 한 번씩만 WARN, 나머지는 DEBUG로 (운영 로그 폭주 방지 — 단순한 throttle)
        if (n == 1 || n % 100 == 0) {
            log.warn("Cache {} error #{}: {}", op, n, lastError);
        } else {
            log.debug("Cache {} error #{}: {}", op, n, lastError);
        }
    }

    public boolean isDisabled() {
        return disabled;
    }

    public CacheStats getStats() {
        return new CacheStats(disabled, errorCount.get(), lastError);
    }

    public record CacheStats(boolean disabled, long errorCount, String lastError) {}
}