#168 [Feat] 투표 SSE 및 댓글 포인트

src/main/java/com/swyp/picke/domain/battle/repository/BattleRepository.java

@@ -56,6 +56,7 @@ public interface BattleRepository extends JpaRepository<Battle, Long> {
     // 6. 전체 배틀 목록 조회
     Page<Battle> findByDeletedAtIsNullOrderByCreatedAtDesc(Pageable pageable);
     Page<Battle> findByStatusAndDeletedAtIsNullOrderByCreatedAtDesc(BattleStatus status, Pageable pageable);
+    List<Battle> findByStatusAndDeletedAtIsNull(BattleStatus status);
 
     // 기본 조회용
     List<Battle> findByTargetDateAndStatusAndDeletedAtIsNull(LocalDate date, BattleStatus status);

src/main/java/com/swyp/picke/domain/perspective/controller/BestCommentSchedulerTestController.java

@@ -0,0 +1,34 @@
+package com.swyp.picke.domain.perspective.controller;
+
+import com.swyp.picke.domain.perspective.scheduler.BestCommentScheduler;
+import com.swyp.picke.global.common.response.ApiResponse;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@Tag(name = "[Test] BestCommentScheduler", description = "스케줄러 테스트 API")
+@RestController
+@RequestMapping("/api/test/scheduler")
+@RequiredArgsConstructor
+public class BestCommentSchedulerTestController {
+
+    private final BestCommentScheduler bestCommentScheduler;
+
+    @Operation(summary = "베스트 댓글 정산 전체 실행", description = "PUBLISHED 상태 배틀 전체를 대상으로 베스트 댓글 포인트 정산을 즉시 실행합니다.")
+    @PostMapping("/best-comment")
+    public ApiResponse<String> runAll() {
+        bestCommentScheduler.awardBestComments();
+        return ApiResponse.onSuccess("베스트 댓글 정산 완료");
+    }
+
+    @Operation(summary = "베스트 댓글 정산 단건 실행", description = "특정 battleId에 대해서만 베스트 댓글 포인트 정산을 즉시 실행합니다.")
+    @PostMapping("/best-comment/battles/{battleId}")
+    public ApiResponse<String> runByBattle(@PathVariable Long battleId) {
+        bestCommentScheduler.processBattle(battleId);
+        return ApiResponse.onSuccess("battleId=" + battleId + " 베스트 댓글 정산 완료");
+    }
+}

src/main/java/com/swyp/picke/domain/perspective/repository/PerspectiveCommentRepository.java

@@ -22,4 +22,7 @@ public interface PerspectiveCommentRepository extends JpaRepository<PerspectiveC
     long countByUserId(Long userId);
 
     void deleteAllByPerspective(Perspective perspective);
+
+    @Query("SELECT c FROM PerspectiveComment c WHERE c.perspective.battle.id = :battleId AND c.hidden = false AND c.likeCount >= :minLikeCount ORDER BY c.likeCount DESC")
+    List<PerspectiveComment> findTopCommentsByBattleId(@Param("battleId") Long battleId, @Param("minLikeCount") int minLikeCount, Pageable pageable);
 }

src/main/java/com/swyp/picke/domain/perspective/scheduler/BestCommentScheduler.java

@@ -0,0 +1,68 @@
+package com.swyp.picke.domain.perspective.scheduler;
+
+import com.swyp.picke.domain.battle.entity.Battle;
+import com.swyp.picke.domain.battle.enums.BattleStatus;
+import com.swyp.picke.domain.battle.repository.BattleRepository;
+import com.swyp.picke.domain.perspective.entity.PerspectiveComment;
+import com.swyp.picke.domain.perspective.repository.PerspectiveCommentRepository;
+import com.swyp.picke.domain.user.enums.CreditType;
+import com.swyp.picke.domain.user.service.CreditService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class BestCommentScheduler {
+
+    private static final int MIN_LIKE_COUNT = 10;
+    private static final int TOP_N = 3;
+
+    private final BattleRepository battleRepository;
+    private final PerspectiveCommentRepository perspectiveCommentRepository;
+    private final CreditService creditService;
+
+    @Scheduled(cron = "0 0 0 * * MON")
+    public void awardBestComments() {
+        log.info("[BestCommentScheduler] 베스트 댓글 포인트 정산 시작");
+
+        List<Battle> battles = battleRepository.findByStatusAndDeletedAtIsNull(BattleStatus.PUBLISHED);
+
+        for (Battle battle : battles) {
+            try {
+                processBattle(battle.getId());
+            } catch (Exception e) {
+                log.error("[BestCommentScheduler] battleId={} 처리 중 오류 발생: {}", battle.getId(), e.getMessage());
+            }
+        }
+
+        log.info("[BestCommentScheduler] 베스트 댓글 포인트 정산 완료");
+    }
+
+    @Transactional
+    public void processBattle(Long battleId) {
+        List<PerspectiveComment> topComments = perspectiveCommentRepository.findTopCommentsByBattleId(
+                battleId,
+                MIN_LIKE_COUNT,
+                PageRequest.of(0, TOP_N)
+        );
+
+        if (topComments.isEmpty()) {
+            return;
+        }
+
+        for (PerspectiveComment comment : topComments) {
+            Long userId = comment.getUser().getId();
+            Long commentId = comment.getId();
+
+            creditService.addCredit(userId, CreditType.BEST_COMMENT, commentId);
+            log.info("[BestCommentScheduler] 포인트 지급 - battleId={}, commentId={}, userId={}", battleId, commentId, userId);
+        }
+    }
+}

src/main/java/com/swyp/picke/domain/user/enums/CreditType.java

@@ -7,7 +7,7 @@ public enum CreditType {
     BATTLE_VOTE(10),
     QUIZ_VOTE(5),
     MAJORITY_WIN(20),
-    BEST_COMMENT(100),
+    BEST_COMMENT(50),
     TOPIC_SUGGEST(30),
     TOPIC_ADOPTED(1000),
     AD_REWARD(50),

src/main/java/com/swyp/picke/domain/vote/controller/VoteController.java

@@ -11,10 +11,14 @@
 import com.swyp.picke.domain.vote.service.BattleVoteService;
 import com.swyp.picke.domain.vote.service.PollVoteService;
 import com.swyp.picke.domain.vote.service.QuizVoteService;
+import com.swyp.picke.domain.vote.sse.SseEmitterRegistry;
 import com.swyp.picke.global.common.response.ApiResponse;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
+import java.io.IOException;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.MediaType;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.security.core.annotation.AuthenticationPrincipal;
 import org.springframework.web.bind.annotation.DeleteMapping;
@@ -24,7 +28,9 @@
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
+@Slf4j
 @Tag(name = "투표 API", description = "배틀/퀴즈/투표 투표 처리")
 @RestController
 @RequestMapping("/api/v1")
@@ -34,6 +40,7 @@ public class VoteController {
     private final BattleVoteService battleVoteService;
     private final QuizVoteService quizVoteService;
     private final PollVoteService pollVoteService;
+    private final SseEmitterRegistry sseEmitterRegistry;
 
     @Operation(summary = "[퀴즈] 답안 제출")
     @PostMapping("/battles/{battleId}/quiz-vote")
@@ -99,6 +106,25 @@ public ApiResponse<VoteStatsResponse> getVoteStats(@PathVariable Long battleId)
         return ApiResponse.onSuccess(battleVoteService.getVoteStats(battleId));
     }
 
+    @Operation(summary = "[배틀] 투표 통계 실시간 구독", description = "post 투표 완료 후 투표 % 실시간 업데이트를 SSE로 수신합니다. 페이지 이탈 시 클라이언트에서 EventSource.close()를 호출해야 합니다.")
+    @GetMapping(value = "/battles/{battleId}/vote-stats/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public SseEmitter streamVoteStats(
+            @PathVariable Long battleId,
+            @AuthenticationPrincipal Long userId
+    ) {
+        SseEmitter emitter = sseEmitterRegistry.register(battleId, userId);
+
+        try {
+            VoteStatsResponse stats = battleVoteService.getVoteStats(battleId);
+            emitter.send(SseEmitter.event().name("vote-stats").data(stats));
+        } catch (IOException e) {
+            log.warn("SSE 초기 데이터 전송 실패 - battleId: {}, userId: {}", battleId, userId);
+            sseEmitterRegistry.remove(battleId, userId);
+        }
+
+        return emitter;
+    }
+
     @Operation(summary = "[배틀] 내 투표 내역 조회", description = "특정 배틀에 대한 내 사전/사후 투표 내역과 현재 상태를 조회합니다.")
     @GetMapping("/battles/{battleId}/votes/me")
     public ApiResponse<MyVoteResponse> getMyVote(

src/main/java/com/swyp/picke/domain/vote/service/BattleVoteServiceImpl.java

@@ -16,12 +16,14 @@
 import com.swyp.picke.domain.vote.dto.response.VoteStatsResponse;
 import com.swyp.picke.domain.vote.entity.BattleVote;
 import com.swyp.picke.domain.vote.repository.BattleVoteRepository;
+import com.swyp.picke.domain.vote.sse.VoteUpdatedEvent;
 import com.swyp.picke.global.common.exception.CustomException;
 import com.swyp.picke.global.common.exception.ErrorCode;
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Optional;
 import lombok.RequiredArgsConstructor;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -35,6 +37,7 @@ public class BattleVoteServiceImpl implements BattleVoteService {
     private final BattleOptionRepository battleOptionRepository;
     private final UserRepository userRepository;
     private final UserBattleService userBattleService;
+    private final ApplicationEventPublisher eventPublisher;
 
     @Override
     public BattleOption findPreVoteOption(Long battleId, Long userId) {
@@ -151,6 +154,7 @@ public VoteResultResponse postVote(Long battleId, Long userId, VoteRequest reque
 
         vote.doPostVote(option);
         userBattleService.upsertStep(user, battle, UserBattleStep.COMPLETED);
+        eventPublisher.publishEvent(new VoteUpdatedEvent(battleId));
 
         return new VoteResultResponse(vote.getId(), UserBattleStep.COMPLETED);
     }

src/main/java/com/swyp/picke/domain/vote/sse/SseEmitterRegistry.java

@@ -0,0 +1,77 @@
+package com.swyp.picke.domain.vote.sse;
+
+import com.swyp.picke.domain.vote.dto.response.VoteStatsResponse;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Slf4j
+@Component
+public class SseEmitterRegistry {
+
+    private static final long TIMEOUT_MS = 180_000L; // 3분
+
+    // battleId -> (userId -> SseEmitter)
+    private final Map<Long, Map<Long, SseEmitter>> emitters = new ConcurrentHashMap<>();
+
+    public SseEmitter register(Long battleId, Long userId) {
+        SseEmitter emitter = new SseEmitter(TIMEOUT_MS);
+
+        // 기존 연결이 있으면 먼저 제거
+        remove(battleId, userId);
+
+        emitters.computeIfAbsent(battleId, k -> new ConcurrentHashMap<>())
+                .put(userId, emitter);
+
+        emitter.onCompletion(() -> remove(battleId, userId));
+        emitter.onTimeout(() -> remove(battleId, userId));
+        emitter.onError(e -> remove(battleId, userId));
+
+        return emitter;
+    }
+
+    public void remove(Long battleId, Long userId) {
+        Map<Long, SseEmitter> battleEmitters = emitters.get(battleId);
+        if (battleEmitters == null) return;
+
+        battleEmitters.remove(userId);
+        if (battleEmitters.isEmpty()) {
+            emitters.remove(battleId);
+        }
+    }
+
+    public void sendToAll(Long battleId, VoteStatsResponse stats) {
+        Map<Long, SseEmitter> battleEmitters = emitters.get(battleId);
+        if (battleEmitters == null || battleEmitters.isEmpty()) return;
+
+        battleEmitters.forEach((userId, emitter) -> {
+            try {
+                emitter.send(SseEmitter.event()
+                        .name("vote-stats")
+                        .data(stats));
+            } catch (IOException e) {
+                log.warn("SSE 전송 실패 - battleId: {}, userId: {}", battleId, userId);
+                remove(battleId, userId);
+            }
+        });
+    }
+
+    // nginx/프록시 idle 끊김 방지용 heartbeat (30초마다)
+    @Scheduled(fixedDelay = 30_000)
+    public void sendHeartbeat() {
+        emitters.forEach((battleId, battleEmitters) ->
+                battleEmitters.forEach((userId, emitter) -> {
+                    try {
+                        emitter.send(SseEmitter.event().comment("ping"));
+                    } catch (IOException e) {
+                        remove(battleId, userId);
+                    }
+                })
+        );
+    }
+}

src/main/java/com/swyp/picke/domain/vote/sse/VoteEventListener.java

@@ -0,0 +1,24 @@
+package com.swyp.picke.domain.vote.sse;
+
+import com.swyp.picke.domain.vote.dto.response.VoteStatsResponse;
+import com.swyp.picke.domain.vote.service.BattleVoteService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+@Component
+@RequiredArgsConstructor
+public class VoteEventListener {
+
+    private final BattleVoteService battleVoteService;
+    private final SseEmitterRegistry sseEmitterRegistry;
+
+    @Async
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void onVoteUpdated(VoteUpdatedEvent event) {
+        VoteStatsResponse stats = battleVoteService.getVoteStats(event.battleId());
+        sseEmitterRegistry.sendToAll(event.battleId(), stats);
+    }
+}

src/main/java/com/swyp/picke/domain/vote/sse/VoteUpdatedEvent.java

@@ -0,0 +1,4 @@
+package com.swyp.picke.domain.vote.sse;
+
+public record VoteUpdatedEvent(Long battleId) {
+}

src/main/java/com/swyp/picke/global/config/AsyncConfig.java

@@ -2,8 +2,10 @@
 
 import org.springframework.context.annotation.Configuration;
 import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 @Configuration
 @EnableAsync
+@EnableScheduling
 public class AsyncConfig {
 }