feat: deliver notification SSE across instances via Redis pub/sub

momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisExecutorConfig.java

@@ -0,0 +1,41 @@
+package com.momogo.api.notification.redis;
+
+import java.util.concurrent.ThreadPoolExecutor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+/*
+ * 알림 Redis Pub/Sub 전용 스레드 풀 설정.
+ */
+@Configuration
+public class NotificationRedisExecutorConfig {
+
+  // 개별 Pub/Sub 메시지 비동기 처리 전용 스레드 풀
+  @Bean(name = "notificationRedisTaskExecutor")
+  public ThreadPoolTaskExecutor notificationRedisTaskExecutor() {
+    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+    executor.setCorePoolSize(10);
+    executor.setMaxPoolSize(50);
+    executor.setQueueCapacity(500);
+    executor.setThreadNamePrefix("notification-redis-task-");
+    // 큐+최대스레드가 모두 꽉 찼을 때 기본 정책(AbortPolicy)은 메시지를 버린다.
+    // CallerRunsPolicy로 바꿔서, 꽉 찼을 땐 발행자 스레드가 대신 처리하게 해
+    // 처리 속도가 느려지더라도 메시지가 유실되지 않도록 한다.
+    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+    executor.initialize();
+    return executor;
+  }
+
+  // Redis SUBSCRIBE 커넥션 유지 전용 독립 스레드 풀
+  @Bean(name = "notificationRedisSubscriptionExecutor")
+  public ThreadPoolTaskExecutor notificationRedisSubscriptionExecutor() {
+    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+    executor.setCorePoolSize(2);
+    executor.setMaxPoolSize(5);
+    executor.setQueueCapacity(10);
+    executor.setThreadNamePrefix("notification-redis-sub-");
+    executor.initialize();
+    return executor;
+  }
+}

momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisPubSubConfig.java

@@ -0,0 +1,61 @@
+package com.momogo.api.notification.redis;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.concurrent.Executor;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
+
+/*
+ * 알림 Redis Pub/Sub 채널/리스너/컨테이너 설정.
+ */
+@Slf4j
+@Configuration
+@RequiredArgsConstructor
+public class NotificationRedisPubSubConfig {
+
+  // 알림 SSE 중계에 사용할 Redis 채널 토픽 정의
+  @Bean
+  public ChannelTopic notificationTopic() {
+    return new ChannelTopic("notification-sse-channel");
+  }
+
+  // Redis 메시지를 수신할 어댑터 설정. Jackson2JsonRedisSerializer로 역직렬화를 위임해
+  // subscriber가 수동으로 JSON을 파싱하지 않고 NotificationSseMessage를 바로 받도록 함.
+  // Jackson2JsonRedisSerializer(Class)는 내부적으로 순정 ObjectMapper를 새로 만들어서
+  // JavaTimeModule이 없어 OffsetDateTime(NotificationResponse.createdAt)을 못 읽는다.
+  // 앱이 이미 쓰는(JavaTimeModule 등록된) ObjectMapper 빈을 그대로 주입해 사용한다.
+  @Bean
+  public MessageListenerAdapter notificationListenerAdapter(
+      NotificationRedisSubscriber subscriber, ObjectMapper objectMapper) {
+    MessageListenerAdapter adapter = new MessageListenerAdapter(subscriber, "handleMessage");
+    adapter.setSerializer(new Jackson2JsonRedisSerializer<>(objectMapper, NotificationSseMessage.class));
+    return adapter;
+  }
+
+  @Bean
+  public RedisMessageListenerContainer notificationRedisMessageListenerContainer(
+      RedisConnectionFactory connectionFactory,
+      MessageListenerAdapter notificationListenerAdapter,
+      ChannelTopic notificationTopic,
+      @Qualifier("notificationRedisTaskExecutor") Executor taskExecutor,
+      @Qualifier("notificationRedisSubscriptionExecutor") Executor subscriptionExecutor
+  ) {
+    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+    container.setConnectionFactory(connectionFactory);
+    container.addMessageListener(notificationListenerAdapter, notificationTopic);
+    container.setTaskExecutor(taskExecutor);
+    container.setSubscriptionExecutor(subscriptionExecutor);
+    container.setErrorHandler(e ->
+        log.error("[Notification Redis Pub/Sub Error] 비동기 메시지 수신/처리 중 오류 발생", e)
+    );
+    return container;
+  }
+}

momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisSubscriber.java

@@ -0,0 +1,56 @@
+package com.momogo.api.notification.redis;
+
+import com.momogo.api.notification.registry.NotificationEmitterRegistry;
+import com.momogo.core.domain.notification.dto.response.NotificationResponse;
+import java.io.IOException;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+/*
+ * NotificationRedisPubSubConfig의 MessageListenerAdapter가 호출하는 구독자.
+ * Redis 채널로 발행된 알림을 받아, 이 인스턴스에 실제로 연결된 유저가 있으면 SSE로 전송
+ * 연결된 유저가 없으면(다른 인스턴스에 연결된 유저면) 조용히 무시
+ *
+ * 역직렬화는 MessageListenerAdapter에 등록한 Jackson2JsonRedisSerializer가 대신 처리하므로
+ * 여기서는 이미 NotificationSseMessage 객체로 전달받는다. 역직렬화 자체가 실패하면
+ * RedisMessageListenerContainer의 errorHandler가 잡아서 로그만 남긴다.
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class NotificationRedisSubscriber {
+
+  private final NotificationEmitterRegistry emitterRegistry;
+
+  public void handleMessage(NotificationSseMessage message) {
+    if (message == null || message.userId() == null || message.notification() == null
+        || message.notification().id() == null) {
+      log.warn("[NotificationRedisSubscriber] 필수 필드가 없는 Redis 메시지를 무시합니다 - message: {}", message);
+      return;
+    }
+
+    NotificationResponse notification = message.notification();
+    var emitters = emitterRegistry.findAllByUserId(message.userId());
+    if (emitters.isEmpty()) {
+      log.info("[NotificationRedisSubscriber] Redis 메시지 수신 - userId: {} (이 인스턴스엔 연결 없음, 무시)",
+          message.userId());
+      return;
+    }
+    log.info("[NotificationRedisSubscriber] Redis 메시지 수신 - userId: {}, 이 인스턴스의 연결 수: {} -> SSE 전송",
+        message.userId(), emitters.size());
+
+    for (SseEmitter emitter : emitters) {
+      try {
+        emitter.send(SseEmitter.event()
+            .id(notification.id().toString())
+            .name("notifications")
+            .data(notification));
+      } catch (IOException e) {
+        log.warn("[NotificationRedisSubscriber] SSE 알림 전송 실패 - userId: {}", message.userId(), e);
+        emitter.completeWithError(e);
+      }
+    }
+  }
+}

momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationSseMessage.java

@@ -0,0 +1,14 @@
+package com.momogo.api.notification.redis;
+
+import com.momogo.core.domain.notification.dto.response.NotificationResponse;
+import java.util.UUID;
+
+/*
+ * SSE 알림을 인스턴스 간에 중계하기 위해 Redis 채널에 실어보내는 메시지.
+ * userId로 "누구에게 보낼지"를, notification으로 "무엇을 보낼지"를 담는다.
+ */
+public record NotificationSseMessage(
+    UUID userId,
+    NotificationResponse notification
+) {
+}

momogo-api/src/main/java/com/momogo/api/notification/service/NotificationSseServiceImpl.java

@@ -1,17 +1,29 @@
 package com.momogo.api.notification.service;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.momogo.api.notification.redis.NotificationSseMessage;
 import com.momogo.api.notification.registry.NotificationEmitterRegistry;
+import com.momogo.core.common.exception.BusinessException;
+import com.momogo.core.common.exception.GlobalErrorCode;
 import com.momogo.core.domain.notification.dto.response.NotificationResponse;
 import com.momogo.core.domain.notification.sse.NotificationSseService;
 import java.io.IOException;
 import java.util.UUID;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.listener.ChannelTopic;
 import org.springframework.stereotype.Service;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 /*
  * NotificationSseService(publish) + NotificationSseConnector(connect) 구현체.
+ *
+ * publish()는 로컬 emitter에 직접 쓰지 않고 Redis 채널로 발행만 한다. 실제 로컬 전송은
+ * NotificationRedisSubscriber가 이 채널을 구독해서 담당한다 - 이 인스턴스에 연결된 유저가
+ * 아니면 조용히 무시되므로, 어느 인스턴스가 publish()를 호출하든 상관없이 그 유저가
+ * 실제로 연결된 인스턴스에서 SSE가 전송된다.
  */
 @Slf4j
 @Service
@@ -22,6 +34,9 @@ public class NotificationSseServiceImpl implements NotificationSseService, Notif
   private static final long TIMEOUT = 30L * 60 * 1000; // 30분
 
   private final NotificationEmitterRegistry emitterRegistry;
+  private final StringRedisTemplate stringRedisTemplate;
+  private final ChannelTopic notificationTopic;
+  private final ObjectMapper objectMapper;
 
   // 클라이언트의 최초 SSE 연결 요청을 처리
   @Override
@@ -41,20 +56,21 @@ public SseEmitter connect(UUID userId) {
     return emitter;
   }
 
-  // 실제 알림을 SSE로 전송
+  // 알림을 Redis 채널로 발행 (실제 SSE 전송은 NotificationRedisSubscriber가 수행)
+  // 발행 실패는 여기서 삼키지 않고 그대로 던진다. 호출자(NotificationSsePublishListener,
+  // NotificationEventListener)가 이미 유저 단위로 try/catch하고 있어, 한 명의 발행 실패가
+  // 다른 유저들의 알림 처리를 막지 않으면서도 실패를 호출자가 인지할 수 있다.
   @Override
   public void publish(UUID userId, NotificationResponse notification) {
-    for (SseEmitter emitter : emitterRegistry.findAllByUserId(userId)) {
-      try {
-        emitter.send(SseEmitter.event()
-            .id(notification.id().toString())
-            .name("notifications")
-            .data(notification));
-      } catch (IOException e) {
-        log.warn("SSE 알림 전송 실패 - userId: {}", userId, e);
-        // completeWithError()가 emitter의 onError 콜백을 트리거 -> registry 정리
-        emitter.completeWithError(e);
-      }
+    String json;
+    try {
+      json = objectMapper.writeValueAsString(new NotificationSseMessage(userId, notification));
+    } catch (JsonProcessingException e) {
+      throw new BusinessException(GlobalErrorCode.SSE_PUBLISH_FAILED, "SSE 메시지 직렬화 실패 - userId: " + userId, e);
     }
+
+    Long receivers = stringRedisTemplate.convertAndSend(notificationTopic.getTopic(), json);
+    log.info("[NotificationSseServiceImpl] Redis 채널({}) 발행 완료 - userId: {}, 구독 중인 인스턴스 수: {}",
+        notificationTopic.getTopic(), userId, receivers);
   }
 }

momogo-core/src/main/java/com/momogo/core/common/exception/GlobalErrorCode.java

@@ -24,6 +24,7 @@ public enum GlobalErrorCode implements ErrorCode {
     OAUTH2_ALREADY_LINKED_OTHER_SOCIAL(9013, "OAUTH2_ALREADY_LINKED_OTHER_SOCIAL", HttpStatus.CONFLICT, "이미 다른 소셜 계정으로 가입된 이메일입니다."),
     OAUTH2_SOCIAL_ID_MISMATCH(9014, "OAUTH2_SOCIAL_ID_MISMATCH", HttpStatus.CONFLICT, "기존 계정 정보와 일치하지 않습니다."),
     OAUTH2_MISSING_ATTRIBUTE(9015, "OAUTH2_MISSING_ATTRIBUTE", HttpStatus.BAD_REQUEST, "OAuth2 사용자 정보에 필수 값이 없습니다."),
+    SSE_PUBLISH_FAILED(9016, "SSE_PUBLISH_FAILED", HttpStatus.INTERNAL_SERVER_ERROR, "SSE 메시지 발행에 실패했습니다."),
 
     INTERNAL_SERVER_ERROR(9999, "SERVER_ERROR", HttpStatus.INTERNAL_SERVER_ERROR, "서버 내부 오류가 발생했습니다.");