fix: 조회수 반영 시 유실 및 재처리 중복 방지

src/main/java/com/example/solidconnection/community/post/domain/ViewCountEvent.java

@@ -0,0 +1,15 @@
+package com.example.solidconnection.community.post.domain;
+
+public record ViewCountEvent(
+        String streamId,
+        String flushId,
+        long postId,
+        long delta
+) {
+
+    public ViewCountEvent {
+        if (flushId == null || flushId.isBlank() || flushId.length() > 100 || postId <= 0 || delta <= 0) {
+            throw new IllegalArgumentException("Invalid view count event");
+        }
+    }
+}

src/main/java/com/example/solidconnection/community/post/domain/ViewCountFlush.java

@@ -0,0 +1,32 @@
+package com.example.solidconnection.community.post.domain;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+import jakarta.persistence.UniqueConstraint;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+@Entity
+@Table(name = "view_count_flush", uniqueConstraints = @UniqueConstraint(name = "uk_view_count_flush_id", columnNames = "flush_id"))
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+public class ViewCountFlush {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    @Column(name = "id", nullable = false)
+    private Long id;
+
+    @Column(name = "flush_id", nullable = false, length = 100)
+    private String flushId;
+
+    // Receipts survive physical post deletion and late redelivery.
+    @Column(name = "post_id", nullable = false)
+    private long postId;
+
+    @Column(name = "delta", nullable = false)
+    private long delta;
+}

src/main/java/com/example/solidconnection/community/post/repository/ViewCountFlushRepository.java

@@ -0,0 +1,28 @@
+package com.example.solidconnection.community.post.repository;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Repository;
+
+@Repository
+@RequiredArgsConstructor
+public class ViewCountFlushRepository {
+
+    private final JdbcTemplate jdbcTemplate;
+
+    public boolean recordIfNew(
+            String flushId,
+            long postId,
+            long delta
+    ) {
+        try {
+            jdbcTemplate.update("INSERT INTO view_count_flush (flush_id, post_id, delta) VALUES (?, ?, ?)",
+                    flushId, postId, delta);
+            return true;
+        } catch (DuplicateKeyException exception) {
+            // A JDBC duplicate does not mark the JPA transaction rollback-only.
+            return false;
+        }
+    }
+}

src/main/java/com/example/solidconnection/community/post/repository/ViewCountStreamRepository.java

@@ -0,0 +1,111 @@
+package com.example.solidconnection.community.post.repository;
+
+import com.example.solidconnection.community.post.domain.ViewCountEvent;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.connection.stream.StreamReadOptions;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.stereotype.Repository;
+
+@Repository
+@Slf4j
+public class ViewCountStreamRepository {
+
+    public static final String ACTIVE_KEY = "post:view:{views}:active";
+    public static final String STREAM_KEY = "post:view:{views}:events";
+    public static final String GROUP = "view-count-db";
+    private static final int BATCH_SIZE = 50;
+    private static final RedisScript<Long> FLUSH = RedisScript.of(new ClassPathResource("scripts/flushViewCounts.lua"), Long.class);
+    private static final RedisScript<List> CLAIM = RedisScript.of(new ClassPathResource("scripts/claimViewCounts.lua"), List.class);
+    private static final RedisScript<Long> ACK = RedisScript.of(new ClassPathResource("scripts/ackViewCount.lua"), Long.class);
+    private static final RedisScript<Long> INIT_GROUP = RedisScript.of("""
+            local result = redis.pcall('XGROUP', 'CREATE', KEYS[1], ARGV[1], '0', 'MKSTREAM')
+            if type(result) == 'table' and result.err and not string.find(result.err, 'BUSYGROUP', 1, true) then
+                return redis.error_reply(result.err)
+            end
+            return 1
+            """, Long.class);
+
+    private final StringRedisTemplate redis;
+    private String claimCursor = "0-0";
+
+    public ViewCountStreamRepository(StringRedisTemplate redis) {
+        this.redis = redis;
+    }
+
+    public void increment(long postId) {
+        redis.opsForHash().increment(ACTIVE_KEY, Long.toString(postId), 1);
+    }
+
+    public void deleteActive(long postId) {
+        redis.opsForHash().delete(ACTIVE_KEY, Long.toString(postId));
+    }
+
+    public long flush() {
+        Long count = redis.execute(FLUSH, List.of(ACTIVE_KEY, STREAM_KEY), UUID.randomUUID().toString(), Integer.toString(BATCH_SIZE));
+        return count == null ? 0 : count;
+    }
+
+    public void ensureGroup() {
+        redis.execute(INIT_GROUP, List.of(STREAM_KEY), GROUP);
+    }
+
+    public List<ViewCountEvent> readNew(String consumer) {
+        List<MapRecord<String, Object, Object>> records = redis.opsForStream().read(
+                Consumer.from(GROUP, consumer), StreamReadOptions.empty().count(BATCH_SIZE),
+                StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()));
+        if (records == null) {
+            return List.of();
+        }
+        return records.stream().flatMap(record -> {
+            Map<Object, Object> fields = record.getValue();
+            return parseEvent(record.getId().getValue(), Objects.toString(fields.get("flushId"), ""),
+                    Objects.toString(fields.get("postId"), ""), Objects.toString(fields.get("delta"), "")).stream();
+        }).toList();
+    }
+
+    public synchronized List<ViewCountEvent> claim(String consumer, Duration minIdle) {
+        List<?> response = redis.execute(CLAIM, List.of(STREAM_KEY), GROUP, consumer,
+                Long.toString(minIdle.toMillis()), claimCursor, Integer.toString(BATCH_SIZE));
+        if (response == null || response.isEmpty()) {
+            return List.of();
+        }
+        claimCursor = response.get(0).toString();
+        List<ViewCountEvent> events = new ArrayList<>();
+        for (int i = 1; i < response.size(); i += 4) {
+            parseEvent(response.get(i).toString(), response.get(i + 1).toString(),
+                    response.get(i + 2).toString(), response.get(i + 3).toString()).ifPresent(events::add);
+        }
+        return events;
+    }
+
+    private Optional<ViewCountEvent> parseEvent(
+            String streamId,
+            String flushId,
+            String postId,
+            String delta
+    ) {
+        try {
+            return Optional.of(new ViewCountEvent(streamId, flushId, Long.parseLong(postId), Long.parseLong(delta)));
+        } catch (IllegalArgumentException exception) {
+            log.warn("Invalid view count event remains pending: streamId={}", streamId, exception);
+            return Optional.empty();
+        }
+    }
+
+    public void acknowledge(ViewCountEvent event) {
+        redis.execute(ACK, List.of(STREAM_KEY), GROUP, event.streamId());
+    }
+}

src/main/java/com/example/solidconnection/community/post/service/PostRedisManager.java

@@ -4,8 +4,8 @@
 import static com.example.solidconnection.redis.RedisConstants.VALIDATE_POST_CREATE_TTL;
 import static com.example.solidconnection.redis.RedisConstants.VALIDATE_VIEW_COUNT_KEY_PREFIX;
 import static com.example.solidconnection.redis.RedisConstants.VALIDATE_VIEW_COUNT_TTL;
-import static com.example.solidconnection.redis.RedisConstants.VIEW_COUNT_KEY_PREFIX;
 
+import com.example.solidconnection.community.post.repository.ViewCountStreamRepository;
 import com.example.solidconnection.redis.RedisService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
@@ -15,34 +15,21 @@
 public class PostRedisManager {
 
     private final RedisService redisService;
-
-    public Long getPostIdFromPostViewCountRedisKey(String key) {
-        return Long.parseLong(key.substring(VIEW_COUNT_KEY_PREFIX.getValue().length()));
-    }
-
-    public Long getAndDeleteViewCount(String key) {
-        return redisService.getAndDelete(key);
-    }
+    private final ViewCountStreamRepository streamRepository;
 
     public void deleteViewCountCache(Long postId) {
-        String key = getPostViewCountRedisKey(postId);
-        redisService.deleteKey(key);
+        streamRepository.deleteActive(postId);
     }
 
     public void incrementViewCountIfFirstAccess(long siteUserId, Long postId) {
         String validateKey = getValidatePostViewCountRedisKey(siteUserId, postId);
         boolean isFirstAccess = redisService.isPresent(validateKey, VALIDATE_VIEW_COUNT_TTL.getValue());
 
         if (isFirstAccess) {
-            String viewCountKey = getPostViewCountRedisKey(postId);
-            redisService.increaseViewCount(viewCountKey);
+            streamRepository.increment(postId);
         }
     }
 
-    public String getPostViewCountRedisKey(Long postId) {
-        return VIEW_COUNT_KEY_PREFIX.getValue() + postId;
-    }
-
     public String getValidatePostViewCountRedisKey(long siteUserId, Long postId) {
         return VALIDATE_VIEW_COUNT_KEY_PREFIX.getValue() + postId + ":" + siteUserId;
     }

src/main/java/com/example/solidconnection/community/post/service/UpdateViewCountService.java

@@ -1,30 +1,24 @@
 package com.example.solidconnection.community.post.service;
 
+import com.example.solidconnection.community.post.domain.ViewCountEvent;
 import com.example.solidconnection.community.post.repository.PostRepository;
+import com.example.solidconnection.community.post.repository.ViewCountFlushRepository;
 import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 @RequiredArgsConstructor
 @Service
-@EnableAsync
-@Slf4j
 public class UpdateViewCountService {
 
     private final PostRepository postRepository;
-    private final PostRedisManager postRedisManager;
+    private final ViewCountFlushRepository flushRepository;
 
     @Transactional
-    @Async
-    public void updateViewCount(String key) {
-        Long postId = postRedisManager.getPostIdFromPostViewCountRedisKey(key);
-        Long viewCount = postRedisManager.getAndDeleteViewCount(key);
-        if (viewCount == null) {
-            return;
+    public void apply(ViewCountEvent event) {
+        if (flushRepository.recordIfNew(event.flushId(), event.postId(), event.delta())) {
+            // Deleted posts are a completed no-op; retain the receipt against replay.
+            postRepository.increaseViewCount(event.postId(), event.delta());
         }
-        postRepository.increaseViewCount(postId, viewCount);
     }
 }

src/main/java/com/example/solidconnection/community/post/service/ViewCountConsumer.java

@@ -0,0 +1,49 @@
+package com.example.solidconnection.community.post.service;
+
+import com.example.solidconnection.community.post.domain.ViewCountEvent;
+import com.example.solidconnection.community.post.repository.ViewCountStreamRepository;
+import java.time.Duration;
+import java.util.UUID;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+@Service
+@Slf4j
+public class ViewCountConsumer {
+
+    private final ViewCountStreamRepository streamRepository;
+    private final UpdateViewCountService updateService;
+    private final Duration minIdle;
+    private final String consumer = UUID.randomUUID().toString();
+
+    public ViewCountConsumer(
+            ViewCountStreamRepository streamRepository,
+            UpdateViewCountService updateService,
+            @Value("${view.count.claim-min-idle:60s}") Duration minIdle
+    ) {
+        this.streamRepository = streamRepository;
+        this.updateService = updateService;
+        this.minIdle = minIdle;
+    }
+
+    public void consume() {
+        streamRepository.ensureGroup();
+        streamRepository.claim(consumer, minIdle).forEach(this::process);
+        streamRepository.readNew(consumer).forEach(this::process);
+    }
+
+    public void flushActiveCounters() {
+        streamRepository.flush();
+    }
+
+    public void process(ViewCountEvent event) {
+        try {
+            updateService.apply(event);
+            // The transactional proxy has committed before apply returns.
+            streamRepository.acknowledge(event);
+        } catch (RuntimeException exception) {
+            log.warn("View count event remains pending: flushId={}", event.flushId(), exception);
+        }
+    }
+}

src/main/java/com/example/solidconnection/scheduler/UpdateViewCountScheduler.java

@@ -1,36 +1,25 @@
 package com.example.solidconnection.scheduler;
 
-import static com.example.solidconnection.redis.RedisConstants.VIEW_COUNT_KEY_PATTERN;
-
-import com.example.solidconnection.community.post.service.UpdateViewCountService;
-import com.example.solidconnection.util.RedisUtils;
-import java.util.List;
+import com.example.solidconnection.community.post.service.ViewCountConsumer;
 import lombok.RequiredArgsConstructor;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.scheduling.annotation.EnableAsync;
-import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.scheduling.annotation.Scheduled;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
 
 @RequiredArgsConstructor
 @Component
-@EnableScheduling
-@EnableAsync
+@ConditionalOnProperty(name = "view.count.scheduling.enabled", havingValue = "true", matchIfMissing = true)
 public class UpdateViewCountScheduler {
 
-    private final RedisUtils redisUtils;
-    private final ThreadPoolTaskExecutor asyncExecutor;
-    private final UpdateViewCountService updateViewCountService;
+    private final ViewCountConsumer consumer;
 
-    @Async
     @Scheduled(fixedDelayString = "${view.count.scheduling.delay}")
     public void updateViewCount() {
-
-        List<String> itemViewCountKeys = redisUtils.getKeysOrderByExpiration(VIEW_COUNT_KEY_PATTERN.getValue());
-
-        itemViewCountKeys.forEach(key -> asyncExecutor.submit(() -> {
-            updateViewCountService.updateViewCount(key);
-        }));
+        try {
+            consumer.flushActiveCounters();
+        } finally {
+            // A failed flush must not prevent recovery of already persisted events.
+            consumer.consume();
+        }
     }
 }

src/main/resources/db/migration/V63__create_view_count_flush.sql

@@ -0,0 +1,8 @@
+CREATE TABLE view_count_flush (
+    id BIGINT NOT NULL AUTO_INCREMENT,
+    flush_id VARCHAR(100) NOT NULL,
+    post_id BIGINT NOT NULL,
+    delta BIGINT NOT NULL,
+    PRIMARY KEY (id),
+    CONSTRAINT uk_view_count_flush_id UNIQUE (flush_id)
+);

src/main/resources/scripts/ackViewCount.lua

@@ -0,0 +1,6 @@
+-- Only one consumer group owns this stream.
+local acknowledged = redis.call('XACK', KEYS[1], ARGV[1], ARGV[2])
+if acknowledged == 1 then
+    redis.call('XDEL', KEYS[1], ARGV[2])
+end
+return acknowledged