From 6da5e758da52924be91549416eb73500482011ba Mon Sep 17 00:00:00 2001 From: pizzazoa Date: Wed, 1 Apr 2026 20:43:10 +0900 Subject: [PATCH] =?UTF-8?q?fix():=20=EC=95=8C=EB=A6=BC=20=EB=B0=9C?= =?UTF-8?q?=EC=86=A1=20=EB=B2=84=EA=B7=B8=20=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 알림 준비를 비동기로 변경 - 배치 쿼리 추가 --- .../AnnouncementBulkNotificationService.java | 113 ++++++++++++++++++ .../AnnouncementPublicationService.java | 27 +---- .../event/AnnouncementNotificationWorker.java | 28 +++++ .../event/AnnouncementPublishedEvent.java | 3 + .../repository/NotificationRepository.java | 14 +++ .../NotificationSettingRepository.java | 14 +++ .../user/core/repository/UserRepository.java | 21 ++++ .../global/core/config/infra/AsyncConfig.java | 11 ++ 8 files changed, 208 insertions(+), 23 deletions(-) create mode 100644 src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/AnnouncementBulkNotificationService.java create mode 100644 src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/event/AnnouncementNotificationWorker.java create mode 100644 src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/event/AnnouncementPublishedEvent.java diff --git a/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/AnnouncementBulkNotificationService.java b/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/AnnouncementBulkNotificationService.java new file mode 100644 index 00000000..ee044938 --- /dev/null +++ b/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/AnnouncementBulkNotificationService.java @@ -0,0 +1,113 @@ +package org.devkor.apu.saerok_server.domain.announcement.application; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.devkor.apu.saerok_server.domain.admin.announcement.core.entity.Announcement; +import org.devkor.apu.saerok_server.domain.admin.announcement.core.repository.AnnouncementRepository; +import org.devkor.apu.saerok_server.domain.notification.application.assembly.render.NotificationRenderer; +import org.devkor.apu.saerok_server.domain.notification.application.dto.PushMessageCommand; +import org.devkor.apu.saerok_server.domain.notification.application.model.payload.SystemNotificationPayload; +import org.devkor.apu.saerok_server.domain.notification.core.entity.Notification; +import org.devkor.apu.saerok_server.domain.notification.core.entity.NotificationType; +import org.devkor.apu.saerok_server.domain.notification.core.repository.NotificationRepository; +import org.devkor.apu.saerok_server.domain.notification.core.repository.NotificationSettingRepository; +import org.devkor.apu.saerok_server.domain.notification.core.repository.UserDeviceRepository; +import org.devkor.apu.saerok_server.domain.notification.infra.fcm.FcmMessageClient; +import org.devkor.apu.saerok_server.domain.user.core.entity.User; +import org.devkor.apu.saerok_server.domain.user.core.repository.UserRepository; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +@Slf4j +@Service +@RequiredArgsConstructor +public class AnnouncementBulkNotificationService { + + private static final int CHUNK_SIZE = 100; + private static final int BATCH_INSERT_SIZE = 100; + private static final int FCM_MAX_TOKENS = 500; + + private final AnnouncementRepository announcementRepository; + private final UserRepository userRepository; + private final NotificationRepository notificationRepository; + private final UserDeviceRepository userDeviceRepository; + private final NotificationSettingRepository settingRepository; + private final NotificationRenderer renderer; + private final FcmMessageClient fcmMessageClient; + + @Transactional + public void sendBulkNotification(Long announcementId) { + Announcement ann = announcementRepository.findById(announcementId).orElse(null); + if (ann == null || !ann.isPublished() || !Boolean.TRUE.equals(ann.getSendNotification())) { + return; + } + + Long annId = ann.getId(); + NotificationType type = NotificationType.SYSTEM_PUBLISHED_ANNOUNCEMENT; + Map extras = Map.of( + "announcementId", annId, + "title", ann.getPushTitle(), + "body", ann.getPushBody(), + "inAppBody", ann.getInAppBody() + ); + + SystemNotificationPayload templatePayload = new SystemNotificationPayload(0L, type, annId, extras); + NotificationRenderer.RenderedMessage rendered = renderer.render(templatePayload); + + int offset = 0; + while (true) { + List userIds = userRepository.findActiveUserIds(offset, CHUNK_SIZE); + if (userIds.isEmpty()) break; + + processChunk(userIds, type, extras, rendered, annId); + offset += userIds.size(); + + if (userIds.size() < CHUNK_SIZE) break; + } + + log.info("Bulk announcement notification completed: announcementId={}, totalUsers={}", + annId, offset); + } + + private void processChunk(List userIds, NotificationType type, + Map extras, + NotificationRenderer.RenderedMessage rendered, + Long announcementId) { + // (a) 유저 배치 조회 + List users = userRepository.findByIds(userIds); + + // (b) 인앱 알림 batch insert + List notifications = users.stream() + .map(user -> Notification.builder() + .user(user) + .type(type) + .isRead(false) + .payload(new HashMap<>(extras)) + .build()) + .toList(); + notificationRepository.batchInsert(notifications, BATCH_INSERT_SIZE); + + // (c) 활성 디바이스 조회 (1 쿼리) + List enabledDeviceIds = settingRepository + .findEnabledDeviceIdsByUserIdsAndType(userIds, type); + if (enabledDeviceIds.isEmpty()) return; + + // (d) 토큰 조회 (1 쿼리) + List tokens = userDeviceRepository.findTokensByUserDeviceIds(enabledDeviceIds); + if (tokens.isEmpty()) return; + + // (e) FCM 전송 + PushMessageCommand cmd = PushMessageCommand.createPushMessageCommand( + rendered.pushTitle(), rendered.pushBody(), + type.name(), announcementId, 1, null); + + for (int i = 0; i < tokens.size(); i += FCM_MAX_TOKENS) { + List batch = tokens.subList(i, Math.min(i + FCM_MAX_TOKENS, tokens.size())); + fcmMessageClient.sendToDevices(batch, cmd); + } + } +} diff --git a/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/AnnouncementPublicationService.java b/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/AnnouncementPublicationService.java index 5858276b..24d6d2cd 100644 --- a/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/AnnouncementPublicationService.java +++ b/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/AnnouncementPublicationService.java @@ -3,9 +3,8 @@ import lombok.RequiredArgsConstructor; import org.devkor.apu.saerok_server.domain.admin.announcement.core.entity.Announcement; import org.devkor.apu.saerok_server.domain.admin.announcement.core.repository.AnnouncementRepository; -import org.devkor.apu.saerok_server.domain.notification.application.facade.NotifySystemService; -import org.devkor.apu.saerok_server.domain.notification.core.entity.NotificationType; -import org.devkor.apu.saerok_server.domain.user.core.repository.UserRepository; +import org.devkor.apu.saerok_server.domain.announcement.application.event.AnnouncementPublishedEvent; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -13,7 +12,6 @@ import java.time.OffsetDateTime; import java.time.ZoneId; import java.util.List; -import java.util.Map; @Service @Transactional @@ -23,8 +21,7 @@ public class AnnouncementPublicationService { private static final ZoneId KST = ZoneId.of("Asia/Seoul"); private final AnnouncementRepository announcementRepository; - private final UserRepository userRepository; - private final NotifySystemService notifySystemService; + private final ApplicationEventPublisher eventPublisher; public OffsetDateTime toKstOffset(LocalDateTime localDateTime) { if (localDateTime == null) return null; @@ -59,22 +56,6 @@ public void notifyPublishedAnnouncement(Announcement announcement) { if (!announcement.isPublished() || !Boolean.TRUE.equals(announcement.getSendNotification())) { return; } - List userIds = userRepository.findActiveUserIds(); - if (userIds.isEmpty()) { - return; - } - Map extras = Map.of( - "announcementId", announcement.getId(), - "title", announcement.getPushTitle(), - "body", announcement.getPushBody(), - "inAppBody", announcement.getInAppBody() - ); - - notifySystemService.notifyUsersDeduplicatedPush( - userIds, - NotificationType.SYSTEM_PUBLISHED_ANNOUNCEMENT, - announcement.getId(), - extras - ); + eventPublisher.publishEvent(new AnnouncementPublishedEvent(announcement.getId())); } } diff --git a/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/event/AnnouncementNotificationWorker.java b/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/event/AnnouncementNotificationWorker.java new file mode 100644 index 00000000..6bb1fe14 --- /dev/null +++ b/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/event/AnnouncementNotificationWorker.java @@ -0,0 +1,28 @@ +package org.devkor.apu.saerok_server.domain.announcement.application.event; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.devkor.apu.saerok_server.domain.announcement.application.AnnouncementBulkNotificationService; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Component; +import org.springframework.transaction.event.TransactionPhase; +import org.springframework.transaction.event.TransactionalEventListener; + +@Slf4j +@Component +@RequiredArgsConstructor +public class AnnouncementNotificationWorker { + + private final AnnouncementBulkNotificationService bulkNotificationService; + + @Async("announcementNotificationExecutor") + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + public void handle(AnnouncementPublishedEvent event) { + try { + bulkNotificationService.sendBulkNotification(event.announcementId()); + } catch (Exception e) { + log.error("Failed to send bulk announcement notification: announcementId={}", + event.announcementId(), e); + } + } +} diff --git a/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/event/AnnouncementPublishedEvent.java b/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/event/AnnouncementPublishedEvent.java new file mode 100644 index 00000000..479125aa --- /dev/null +++ b/src/main/java/org/devkor/apu/saerok_server/domain/announcement/application/event/AnnouncementPublishedEvent.java @@ -0,0 +1,3 @@ +package org.devkor.apu.saerok_server.domain.announcement.application.event; + +public record AnnouncementPublishedEvent(Long announcementId) {} diff --git a/src/main/java/org/devkor/apu/saerok_server/domain/notification/core/repository/NotificationRepository.java b/src/main/java/org/devkor/apu/saerok_server/domain/notification/core/repository/NotificationRepository.java index 422f5d8e..c284d0db 100644 --- a/src/main/java/org/devkor/apu/saerok_server/domain/notification/core/repository/NotificationRepository.java +++ b/src/main/java/org/devkor/apu/saerok_server/domain/notification/core/repository/NotificationRepository.java @@ -56,4 +56,18 @@ public void markAllAsReadByUserId(Long userId) { .setParameter("userId", userId) .executeUpdate(); } + + public void batchInsert(List notifications, int batchSize) { + for (int i = 0; i < notifications.size(); i++) { + em.persist(notifications.get(i)); + if ((i + 1) % batchSize == 0) { + em.flush(); + em.clear(); + } + } + if (!notifications.isEmpty() && notifications.size() % batchSize != 0) { + em.flush(); + em.clear(); + } + } } diff --git a/src/main/java/org/devkor/apu/saerok_server/domain/notification/core/repository/NotificationSettingRepository.java b/src/main/java/org/devkor/apu/saerok_server/domain/notification/core/repository/NotificationSettingRepository.java index e78763c7..2865f948 100644 --- a/src/main/java/org/devkor/apu/saerok_server/domain/notification/core/repository/NotificationSettingRepository.java +++ b/src/main/java/org/devkor/apu/saerok_server/domain/notification/core/repository/NotificationSettingRepository.java @@ -52,6 +52,20 @@ public List findEnabledDeviceIdsByUserAndType(Long userId, NotificationTyp .getResultList(); } + public List findEnabledDeviceIdsByUserIdsAndType(List userIds, NotificationType type) { + if (userIds == null || userIds.isEmpty()) return List.of(); + return em.createQuery(""" + select ns.userDevice.id + from NotificationSetting ns + where ns.userDevice.user.id in :userIds + and ns.type = :type + and ns.enabled = true + """, Long.class) + .setParameter("userIds", userIds) + .setParameter("type", type) + .getResultList(); + } + public void save(NotificationSetting setting) { em.persist(setting); } diff --git a/src/main/java/org/devkor/apu/saerok_server/domain/user/core/repository/UserRepository.java b/src/main/java/org/devkor/apu/saerok_server/domain/user/core/repository/UserRepository.java index 474f54f2..6db20445 100644 --- a/src/main/java/org/devkor/apu/saerok_server/domain/user/core/repository/UserRepository.java +++ b/src/main/java/org/devkor/apu/saerok_server/domain/user/core/repository/UserRepository.java @@ -49,4 +49,25 @@ public List findActiveUserIds() { .setParameter("withdrawn", SignupStatusType.WITHDRAWN) .getResultList(); } + + public List findActiveUserIds(int offset, int limit) { + return em.createQuery( + "SELECT u.id FROM User u " + + "WHERE u.deletedAt IS NULL AND u.signupStatus <> :withdrawn " + + "ORDER BY u.id", + Long.class) + .setParameter("withdrawn", SignupStatusType.WITHDRAWN) + .setFirstResult(offset) + .setMaxResults(limit) + .getResultList(); + } + + public List findByIds(List ids) { + if (ids == null || ids.isEmpty()) return List.of(); + return em.createQuery( + "SELECT u FROM User u WHERE u.id IN :ids", + User.class) + .setParameter("ids", ids) + .getResultList(); + } } diff --git a/src/main/java/org/devkor/apu/saerok_server/global/core/config/infra/AsyncConfig.java b/src/main/java/org/devkor/apu/saerok_server/global/core/config/infra/AsyncConfig.java index 7446bd32..388791f4 100644 --- a/src/main/java/org/devkor/apu/saerok_server/global/core/config/infra/AsyncConfig.java +++ b/src/main/java/org/devkor/apu/saerok_server/global/core/config/infra/AsyncConfig.java @@ -20,4 +20,15 @@ public Executor pushNotificationExecutor() { executor.initialize(); return executor; } + + @Bean(name = "announcementNotificationExecutor") + public Executor announcementNotificationExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(1); + executor.setMaxPoolSize(2); + executor.setQueueCapacity(10); + executor.setThreadNamePrefix("ann-notify-"); + executor.initialize(); + return executor; + } }