Merge pull request #253 from FunD-StockProject/fix/notification-only-one-push

Fix: 알림 시도때도 없이 발생해서 푸시 알림으로 가던 현상 해결

Author: MuuiGong

src/main/java/com/fund/stockProject/notification/repository/OutboxRepository.java

@@ -23,6 +23,20 @@ public interface OutboxRepository extends JpaRepository<OutboxEvent, Integer> {
     @Query("SELECT e FROM OutboxEvent e WHERE e.status IN :statuses AND (e.scheduledAt IS NULL OR e.scheduledAt <= :now)")
     Page<OutboxEvent> findReadyToProcessInStatuses(@Param("statuses") List<String> statuses, @Param("now") Instant now, Pageable pageable);
 
+    @Query("""
+        SELECT e FROM OutboxEvent e
+        WHERE e.status = 'PENDING'
+        AND e.scheduledAt IS NULL
+    """)
+    Page<OutboxEvent> findReadyImmediateEvents(Pageable pageable);
+
+    @Query("""
+        SELECT e FROM OutboxEvent e
+        WHERE e.status = 'READY_TO_SEND'
+        AND e.scheduledAt <= :now
+    """)
+    Page<OutboxEvent> findReadyScheduledEvents(@Param("now") Instant now, Pageable pageable);
+
     @Query("SELECT e FROM OutboxEvent e WHERE e.status = :status AND e.nextAttemptAt <= :now")
     List<OutboxEvent> findRetryableEvents(@Param("status") String status, @Param("now") Instant now);
 

src/main/java/com/fund/stockProject/notification/service/OutboxDispatcher.java

@@ -43,31 +43,45 @@ public OutboxDispatcher(
     }
 
     /**
-     * 즉시 발송 처리 (15분마다)
+     * 즉시 알림 발송 처리
      */
-    @Scheduled(fixedDelay = 900000) // 15분 = 900,000ms
+    @Scheduled(fixedDelay = 60000) // 1분
     public void dispatchImmediate() {
         try {
-            // 스케줄러에서는 트랜잭션을 사용하지 않고, 개별 처리에서만 사용
-            var batch = outboxRepo.findReadyToProcessInStatuses(
-                    List.of("PENDING", "READY_TO_SEND"),
-                    Instant.now(),
-                    PageRequest.of(0, 100)
-            );
+            var batch = outboxRepo.findReadyImmediateEvents(PageRequest.of(0, 100));
 
-            log.debug("Found {} events to process immediately", batch.getContent().size());
+            log.debug("Found {} immediate events to process", batch.getContent().size());
 
             for (OutboxEvent e : batch.getContent()) {
                 if (!"ALERT_CREATED".equals(e.getType())) continue;
 
-                // 각 이벤트를 개별 트랜잭션으로 처리
-                processEventWithTransaction(e);
+                processEvent(e);
             }
         } catch (Exception e) {
             log.error("Error in dispatchImmediate scheduler", e);
         }
     }
 
+    /**
+     * 예약 알림 발송 처리 (9시대 매분)
+     */
+    @Scheduled(cron = "0 * 9 * * *", zone = "Asia/Seoul")
+    public void dispatchScheduled() {
+        try {
+            var batch = outboxRepo.findReadyScheduledEvents(Instant.now(), PageRequest.of(0, 200));
+
+            log.debug("Found {} scheduled events to process", batch.getContent().size());
+
+            for (OutboxEvent e : batch.getContent()) {
+                if (!"ALERT_CREATED".equals(e.getType())) continue;
+
+                processEvent(e);
+            }
+        } catch (Exception e) {
+            log.error("Error in dispatchScheduled scheduler", e);
+        }
+    }
+
     /**
      * 재시도 이벤트 처리 (5분마다)
      */
@@ -81,27 +95,13 @@ public void dispatchRetry() {
             for (OutboxEvent e : retryEvents) {
                 if (!"ALERT_CREATED".equals(e.getType())) continue;
 
-                // 각 이벤트를 개별 트랜잭션으로 처리
-                processEventWithTransaction(e);
+                processEvent(e);
             }
         } catch (Exception e) {
             log.error("Error in dispatchRetry scheduler", e);
         }
     }
 
-    /**
-     * 개별 이벤트를 트랜잭션으로 처리
-     */
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
-    public void processEventWithTransaction(OutboxEvent e) {
-        try {
-            processEvent(e);
-        } catch (Exception ex) {
-            log.error("Failed to process event {}: {}", e.getId(), ex.getMessage());
-            handleError(e, ex);
-        }
-    }
-
     /**
      * 이벤트 처리 공통 로직
      */
@@ -132,6 +132,7 @@ private void processEvent(OutboxEvent e) {
             }
 
             e.setStatus("PROCESSED");
+            outboxRepo.save(e);
             log.info("Notification sent successfully: userId={}, notificationId={}", userId, nId);
 
         } catch (Exception ex) {
@@ -150,6 +151,7 @@ private void handleError(OutboxEvent e, Exception ex) {
         long backoffSec = Math.min(300, 1L << Math.min(10, retryCount));
         e.setNextAttemptAt(Instant.now().plusSeconds(backoffSec));
         e.setStatus("RETRY");
+        outboxRepo.save(e);
 
         log.warn("Notification dispatch failed, will retry: eventId={}, retryCount={}, nextAttempt={}, error={}", 
                 e.getId(), retryCount, e.getNextAttemptAt(), ex.getMessage());

src/main/java/com/fund/stockProject/notification/service/StockScoreAlertService.java

@@ -1,5 +1,7 @@
 package com.fund.stockProject.notification.service;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fund.stockProject.notification.domain.NotificationType;
 import com.fund.stockProject.notification.entity.Notification;
 import com.fund.stockProject.notification.entity.OutboxEvent;
@@ -9,23 +11,28 @@
 import com.fund.stockProject.preference.entity.Preference;
 import com.fund.stockProject.preference.repository.PreferenceRepository;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalTime;
 import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 @Service
 @RequiredArgsConstructor
+@Slf4j
 public class StockScoreAlertService {
     private final NotificationRepository notificationRepo;
     private final OutboxRepository outboxRepo;
     private final NotificationService notificationService;
     private final PreferenceRepository preferenceRepo;
+    private final ObjectMapper objectMapper;
 
     private static final int THRESHOLD_ABS = 15;
 
@@ -89,10 +96,102 @@ public void sendDailyScoreAlerts() {
         // PENDING 상태이고 scheduledAt이 현재 시간 이전인 알림들을 발송
         List<OutboxEvent> pendingEvents = outboxRepo.findByStatusAndScheduledAtBefore(
             "PENDING", Instant.now());
-        
+
+        if (pendingEvents.isEmpty()) {
+            return;
+        }
+
+        Map<Integer, List<OutboxEvent>> scoreSpikeEventsByUser = new HashMap<>();
+        int readyCount = 0;
+        int suppressedCount = 0;
+
         for (OutboxEvent event : pendingEvents) {
-            // OutboxDispatcher에서 처리하도록 상태 업데이트
+            Map<String, Object> payload = parsePayload(event.getPayload());
+            if (payload == null) {
+                event.setStatus("READY_TO_SEND");
+                readyCount++;
+                continue;
+            }
+
+            String type = String.valueOf(payload.get("type"));
+            Integer userId = toInteger(payload.get("userId"));
+
+            if (NotificationType.SCORE_SPIKE.name().equals(type) && userId != null) {
+                scoreSpikeEventsByUser.computeIfAbsent(userId, key -> new ArrayList<>()).add(event);
+                continue;
+            }
+
             event.setStatus("READY_TO_SEND");
+            readyCount++;
+        }
+
+        for (Map.Entry<Integer, List<OutboxEvent>> entry : scoreSpikeEventsByUser.entrySet()) {
+            List<OutboxEvent> userEvents = entry.getValue();
+            if (userEvents.isEmpty()) {
+                continue;
+            }
+
+            OutboxEvent representative = userEvents.get(0);
+            representative.setStatus("READY_TO_SEND");
+            readyCount++;
+
+            if (userEvents.size() > 1) {
+                updateRepresentativeMessage(representative, userEvents.size());
+                for (int i = 1; i < userEvents.size(); i++) {
+                    userEvents.get(i).setStatus("PROCESSED");
+                    suppressedCount++;
+                }
+            }
+        }
+
+        log.info("Daily score alerts prepared: totalPending={}, readyToSend={}, suppressed={}",
+            pendingEvents.size(), readyCount, suppressedCount);
+    }
+
+    private void updateRepresentativeMessage(OutboxEvent representative, int totalCount) {
+        Map<String, Object> payload = parsePayload(representative.getPayload());
+        if (payload == null) {
+            return;
+        }
+
+        Integer notificationId = toInteger(payload.get("notificationId"));
+        if (notificationId == null) {
+            return;
+        }
+
+        notificationRepo.findById(notificationId).ifPresent(notification -> {
+            notification.setTitle("북마크 종목 점수 급변 알림");
+            notification.setBody("북마크한 종목 " + totalCount + "개의 점수가 크게 변했습니다.");
+            notificationRepo.save(notification);
+        });
+    }
+
+    private Map<String, Object> parsePayload(String payload) {
+        if (payload == null || payload.isBlank()) {
+            return null;
+        }
+
+        try {
+            return objectMapper.readValue(payload, new TypeReference<>() {});
+        } catch (Exception e) {
+            log.warn("Failed to parse outbox payload: {}", e.getMessage());
+            return null;
+        }
+    }
+
+    private Integer toInteger(Object value) {
+        if (value == null) {
+            return null;
+        }
+
+        if (value instanceof Number number) {
+            return number.intValue();
+        }
+
+        try {
+            return Integer.valueOf(value.toString());
+        } catch (NumberFormatException e) {
+            return null;
         }
     }
 }