Merge pull request #17 from Pinit-Scheduler/feat/알림-배치처리-적용

Feat/알림 배치처리 적용

Author: GoGradually

src/main/java/me/pinitnotification/application/notification/NotificationDispatchItem.java

@@ -0,0 +1,11 @@
+package me.pinitnotification.application.notification;
+
+import me.pinitnotification.domain.notification.UpcomingScheduleNotification;
+
+import java.util.List;
+
+public record NotificationDispatchItem(
+        UpcomingScheduleNotification notification,
+        List<String> tokens
+) {
+}

src/main/java/me/pinitnotification/application/notification/NotificationDispatchQueryRepository.java

@@ -0,0 +1,8 @@
+package me.pinitnotification.application.notification;
+
+import java.time.Instant;
+import java.util.List;
+
+public interface NotificationDispatchQueryRepository {
+    List<NotificationDispatchItem> findAllDueNotificationsWithTokens(Instant now);
+}

src/main/java/me/pinitnotification/application/notification/NotificationDispatchScheduler.java

@@ -1,64 +1,85 @@
 package me.pinitnotification.application.notification;
 
+import me.pinitnotification.application.push.PushSendResult;
+import me.pinitnotification.application.push.PushService;
 import me.pinitnotification.domain.notification.UpcomingScheduleNotification;
 import me.pinitnotification.domain.notification.UpcomingScheduleNotificationRepository;
-import me.pinitnotification.domain.push.PushSubscription;
-import me.pinitnotification.domain.push.PushSubscriptionRepository;
-import me.pinitnotification.application.push.PushService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.Clock;
-import java.time.OffsetDateTime;
-import java.time.format.DateTimeParseException;
+import java.time.Instant;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 @Service
 public class NotificationDispatchScheduler {
     private static final Logger log = LoggerFactory.getLogger(NotificationDispatchScheduler.class);
 
     private final UpcomingScheduleNotificationRepository notificationRepository;
-    private final PushSubscriptionRepository pushSubscriptionRepository;
+    private final NotificationDispatchQueryRepository dispatchQueryRepository;
+    private final PushTokenCleanupService pushTokenCleanupService;
     private final PushService pushService;
     private final Clock clock;
 
     public NotificationDispatchScheduler(UpcomingScheduleNotificationRepository notificationRepository,
-                                         PushSubscriptionRepository pushSubscriptionRepository,
+                                         NotificationDispatchQueryRepository dispatchQueryRepository,
+                                         PushTokenCleanupService pushTokenCleanupService,
                                          PushService pushService,
                                          Clock clock) {
         this.notificationRepository = notificationRepository;
-        this.pushSubscriptionRepository = pushSubscriptionRepository;
+        this.dispatchQueryRepository = dispatchQueryRepository;
+        this.pushTokenCleanupService = pushTokenCleanupService;
         this.pushService = pushService;
         this.clock = clock;
     }
 
     @Scheduled(cron = "0 */10 * * * *")
     @Transactional
     public void dispatchDueNotifications() {
-        OffsetDateTime now = OffsetDateTime.now(clock);
-        List<UpcomingScheduleNotification> dueNotifications = notificationRepository.findAll().stream()
-                .filter(notification -> notification.isDue(now))
-                .toList();
+        Instant now = Instant.now(clock);
+        List<NotificationDispatchItem> dispatchItems = dispatchQueryRepository.findAllDueNotificationsWithTokens(now);
 
-        if (dueNotifications.isEmpty()) {
+        if (dispatchItems.isEmpty()) {
             return;
         }
 
-        dueNotifications.forEach(this::sendNotificationToOwner);
-        notificationRepository.deleteAllInBatch(dueNotifications);
+        Set<String> tokensToDelete = new HashSet<>();
+        dispatchItems.forEach(item -> sendNotificationToOwner(item, tokensToDelete));
+        notificationRepository.deleteAllInBatch(dispatchItems.stream().map(NotificationDispatchItem::notification).toList());
+
+        if (!tokensToDelete.isEmpty()) {
+            deleteTokensSafely(tokensToDelete);
+        }
     }
 
 
-    private void sendNotificationToOwner(UpcomingScheduleNotification notification) {
-        List<PushSubscription> subscriptions = pushSubscriptionRepository.findAllByMemberId(notification.getOwnerId());
-        if (subscriptions.isEmpty()) {
+    private void sendNotificationToOwner(NotificationDispatchItem dispatchItem, Set<String> tokensToDelete) {
+        UpcomingScheduleNotification notification = dispatchItem.notification();
+        List<String> tokens = dispatchItem.tokens();
+
+        if (tokens.isEmpty()) {
             log.info("No push tokens for owner; skip sending. ownerId={}, scheduleId={}", notification.getOwnerId(), notification.getScheduleId());
             return;
         }
 
-        subscriptions.forEach(subscription -> pushService.sendPushMessage(subscription.getToken(), notification));
+        tokens.forEach(token -> {
+            PushSendResult result = pushService.sendPushMessage(token, notification);
+            if (result.shouldDeleteToken()) {
+                tokensToDelete.add(token);
+            }
+        });
+    }
+
+    private void deleteTokensSafely(Set<String> tokensToDelete) {
+        try {
+            pushTokenCleanupService.deleteTokensInNewTransaction(tokensToDelete);
+        } catch (Exception ex) {
+            log.warn("Failed to delete invalid push tokens; notifications already removed. tokens={}", tokensToDelete.size(), ex);
+        }
     }
 }

src/main/java/me/pinitnotification/application/notification/PushTokenCleanupService.java

@@ -0,0 +1,29 @@
+package me.pinitnotification.application.notification;
+
+import me.pinitnotification.domain.push.PushSubscriptionRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.Set;
+
+@Service
+public class PushTokenCleanupService {
+    private static final Logger log = LoggerFactory.getLogger(PushTokenCleanupService.class);
+    private final PushSubscriptionRepository pushSubscriptionRepository;
+
+    public PushTokenCleanupService(PushSubscriptionRepository pushSubscriptionRepository) {
+        this.pushSubscriptionRepository = pushSubscriptionRepository;
+    }
+
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public void deleteTokensInNewTransaction(Set<String> tokens) {
+        if (tokens == null || tokens.isEmpty()) {
+            return;
+        }
+        pushSubscriptionRepository.deleteByTokens(tokens);
+        log.info("Deleted {} invalid push tokens", tokens.size());
+    }
+}

src/main/java/me/pinitnotification/application/push/PushSendResult.java

@@ -0,0 +1,22 @@
+package me.pinitnotification.application.push;
+
+public record PushSendResult(
+        boolean success,
+        boolean invalidToken
+) {
+    public static PushSendResult successResult() {
+        return new PushSendResult(true, false);
+    }
+
+    public static PushSendResult invalidTokenResult() {
+        return new PushSendResult(false, true);
+    }
+
+    public static PushSendResult failedResult() {
+        return new PushSendResult(false, false);
+    }
+
+    public boolean shouldDeleteToken() {
+        return invalidToken;
+    }
+}

src/main/java/me/pinitnotification/application/push/PushService.java

@@ -8,7 +8,8 @@ public interface PushService {
     void subscribe(Long memberId, String deviceId, String token);
 
     void unsubscribe(Long memberId, String deviceId, String token);
-    void sendPushMessage(String token, Notification notification);
+
+    PushSendResult sendPushMessage(String token, Notification notification);
 
     boolean isSubscribed(Long memberId, String deviceId);
 }

src/main/java/me/pinitnotification/domain/push/PushSubscriptionRepository.java

@@ -1,5 +1,6 @@
 package me.pinitnotification.domain.push;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
 
@@ -12,5 +13,7 @@ public interface PushSubscriptionRepository {
 
     void deleteByToken(String token);
 
+    void deleteByTokens(Collection<String> tokens);
+
     void deleteByMemberIdAndDeviceId(Long memberId, String deviceId);
 }

src/main/java/me/pinitnotification/infrastructure/fcm/FcmService.java

@@ -5,6 +5,7 @@
 import com.google.firebase.messaging.Message;
 import com.google.firebase.messaging.MessagingErrorCode;
 import lombok.extern.slf4j.Slf4j;
+import me.pinitnotification.application.push.PushSendResult;
 import me.pinitnotification.application.push.PushService;
 import me.pinitnotification.domain.notification.Notification;
 import me.pinitnotification.domain.push.PushSubscription;
@@ -34,20 +35,21 @@ public FcmService(FirebaseMessaging firebaseMessaging,
     }
 
     @Override
-    public void sendPushMessage(String token, Notification notification) {
+    public PushSendResult sendPushMessage(String token, Notification notification) {
         log.info("publish token: {}", token);
         Message message = Message.builder()
                 .setToken(token)
                 .putAllData(notification.getData())
                 .build();
         try {
             firebaseMessaging.send(message);
+            return PushSendResult.successResult();
         } catch (FirebaseMessagingException e) {
             log.error(e.getMessage(), e);
             if (e.getMessagingErrorCode() == MessagingErrorCode.UNREGISTERED || e.getMessagingErrorCode() == MessagingErrorCode.INVALID_ARGUMENT) {
-                // Todo 토큰 삭제 방식 변경 필요
-                pushSubscriptionRepository.deleteByToken(token);
+                return PushSendResult.invalidTokenResult();
             }
+            return PushSendResult.failedResult();
         }
     }
 

src/main/java/me/pinitnotification/infrastructure/persistence/notification/NotificationDispatchQueryRepositoryAdapter.java

@@ -0,0 +1,103 @@
+package me.pinitnotification.infrastructure.persistence.notification;
+
+import me.pinitnotification.application.notification.NotificationDispatchItem;
+import me.pinitnotification.application.notification.NotificationDispatchQueryRepository;
+import me.pinitnotification.domain.notification.UpcomingScheduleNotification;
+import org.springframework.jdbc.core.simple.JdbcClient;
+import org.springframework.stereotype.Repository;
+
+import java.time.Instant;
+import java.util.*;
+
+@Repository
+public class NotificationDispatchQueryRepositoryAdapter implements NotificationDispatchQueryRepository {
+    private static final String FIND_DUE_WITH_TOKENS_SQL = """
+            SELECT n.public_id,
+                   n.owner_id,
+                   n.schedule_id,
+                   n.schedule_title,
+                   n.schedule_start_time,
+                   n.idempotent_key,
+                   ps.token
+            FROM upcoming_schedule_notification n
+                     LEFT JOIN push_subscription ps ON ps.member_id = n.owner_id
+            WHERE n.schedule_start_time IS NOT NULL
+              AND n.schedule_start_time <= ?
+            """;
+
+    private final JdbcClient jdbcClient;
+
+    public NotificationDispatchQueryRepositoryAdapter(JdbcClient jdbcClient) {
+        this.jdbcClient = jdbcClient;
+    }
+
+    @Override
+    public List<NotificationDispatchItem> findAllDueNotificationsWithTokens(Instant now) {
+        List<DispatchRow> rows = jdbcClient.sql(FIND_DUE_WITH_TOKENS_SQL)
+                .param(now.toString())
+                .query((rs, rowNum) -> new DispatchRow(
+                        UUID.fromString(rs.getString("public_id")),
+                        rs.getLong("owner_id"),
+                        rs.getLong("schedule_id"),
+                        rs.getString("schedule_title"),
+                        rs.getString("schedule_start_time"),
+                        rs.getString("idempotent_key"),
+                        rs.getString("token")
+                ))
+                .list();
+
+        if (rows.isEmpty()) {
+            return List.of();
+        }
+
+        Map<UUID, DispatchAccumulator> aggregated = new LinkedHashMap<>();
+        for (DispatchRow row : rows) {
+            DispatchAccumulator accumulator = aggregated.computeIfAbsent(
+                    row.notificationId,
+                    id -> new DispatchAccumulator(
+                            toDomain(row),
+                            new ArrayList<>()
+                    )
+            );
+            if (row.token != null) {
+                accumulator.tokens.add(row.token);
+            }
+        }
+
+        return aggregated.values().stream()
+                .map(accumulator -> new NotificationDispatchItem(accumulator.notification, List.copyOf(accumulator.tokens)))
+                .toList();
+    }
+
+    private UpcomingScheduleNotification toDomain(DispatchRow row) {
+        return new UpcomingScheduleNotification(
+                row.notificationId,
+                row.ownerId,
+                row.scheduleId,
+                row.scheduleTitle,
+                row.scheduleStartTime,
+                row.idempotentKey
+        );
+    }
+
+    private record DispatchRow(
+            UUID notificationId,
+            Long ownerId,
+            Long scheduleId,
+            String scheduleTitle,
+            String scheduleStartTime,
+            String idempotentKey,
+            String token
+    ) {
+    }
+
+    private static class DispatchAccumulator {
+        private final UpcomingScheduleNotification notification;
+        private final List<String> tokens;
+
+        private DispatchAccumulator(UpcomingScheduleNotification notification, List<String> tokens) {
+            this.notification = notification;
+            this.tokens = tokens;
+        }
+    }
+}

src/main/java/me/pinitnotification/infrastructure/persistence/notification/UpcomingScheduleNotificationEntity.java

@@ -15,6 +15,9 @@
         uniqueConstraints = {
                 @UniqueConstraint(name = "uk_schedule_owner", columnNames = {"schedule_id", "owner_id"}),
                 @UniqueConstraint(name = "uk_idempotent_key", columnNames = {"idempotent_key"})
+        },
+        indexes = {
+                @Index(name = "idx_upcoming_start_time", columnList = "schedule_start_time")
         }
 )
 @Getter