feat: 알림 실시간 채널 (SSE)

docs/features/realtime-notifications.md

@@ -0,0 +1,61 @@
+# 실시간 알림 (SSE)
+
+> 관련 이슈: #45
+
+## 결정 — WebSocket 이 아니라 SSE
+
+| | SSE | WebSocket |
+|---|---|---|
+| 방향 | 서버 → 클라이언트 | 양방향 |
+| 인증 | 기존 JWT 필터 그대로 | 핸드셰이크 인증을 따로 구현 |
+| 인가·CORS·레이트 리밋 | 기존 규칙 그대로 (`/notifications/**` 인증) | 별도 |
+| 프록시 | 평범한 HTTP. 버퍼링만 끄면 된다 | Upgrade 설정 필요 |
+
+알림은 한 방향이다. 양방향 채널이 주는 이점이 없고, 인증·인가를 두 벌 만드는 비용만 생긴다.
+
+## 흐름
+
+```
+알림 저장 (어느 경로든: 서비스, 관리자 발송, 빈자리 알림 …)
+  → @PostPersist (NotificationPersistListener) → NotificationCreatedEvent
+  → 커밋 후 (@TransactionalEventListener AFTER_COMMIT)
+  → NotificationStreamService → 받는 사람의 열린 연결마다 전송
+```
+
+- 알림을 만드는 곳마다 발송 코드를 넣지 않는다. 엔티티 저장에 걸어서 새 생성 경로가 생겨도 빠지지 않는다.
+- **커밋 뒤에** 보낸다. 롤백된 알림이 화면에 떴다가 목록에서 사라지는 일이 없다.
+
+## 계약
+
+`GET /notifications/stream` — `Authorization: Bearer <accessToken>`, 응답 `text/event-stream`
+
+| 이벤트 | 언제 | data |
+|--------|------|------|
+| `connected` | 연결 직후 | `ok` |
+| `notification` | 새 알림 | 알림 JSON (목록 API 의 항목과 같은 모양). `id:` 는 알림 id |
+| `: ping` (주석) | 25초마다 | — |
+
+브라우저 `EventSource` 는 헤더를 못 붙이므로 `fetch` 스트림으로 읽는다 (CareCode_FE `useNotificationStream`).
+
+## 끊김·재연결·중복
+
+- 연결은 30분 뒤 서버가 닫는다. 클라이언트는 다시 연결한다(지수 백오프).
+- 끊긴 사이의 알림은 다시 보내 주지 않는다. 대신 **`connected` 를 받으면 목록을 새로 불러온다.**
+  목록이 진실의 원천이고 SSE 는 "새로 불러올 때가 됐다" 는 신호다.
+- 같은 알림을 두 번 받아도 목록을 다시 불러올 뿐이라 화면에 두 번 나오지 않는다.
+- 사용자당 연결은 5개까지. 넘치면 가장 오래된 것을 닫는다 (재연결을 반복하는 클라이언트가 연결을 쌓지 못하게).
+
+## 운영
+
+| 항목 | 내용 |
+|------|------|
+| 지표 | `carecode.notification.stream.connections` — 열린 연결 수 (`/actuator/prometheus`) |
+| 프록시 | 응답에 `X-Accel-Buffering: no`. Nginx 를 앞에 두면 `proxy_read_timeout` 을 heartbeat(25초)보다 길게 |
+| 설정 | `app.notification.stream.timeout-millis`(30분), `heartbeat-millis`(25초), `max-connections-per-user`(5) |
+| 보안 | 연결 종료 시 컨테이너의 ASYNC 디스패치는 인가를 다시 하지 않는다 (원 요청에서 이미 통과) |
+
+## 한계
+
+연결을 인스턴스 메모리에 둔다. 지금 운영은 단일 인스턴스라 충분하다.
+**여러 대로 늘리면** 알림이 저장된 인스턴스와 사용자가 연결된 인스턴스가 다를 수 있으므로,
+`NotificationCreatedEvent` 를 Redis pub/sub 로 모든 인스턴스에 퍼뜨려야 한다.

src/main/java/com/carecode/core/security/SecurityConfig.java

@@ -78,6 +78,10 @@ public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
                 })
             )
             .authorizeHttpRequests(authz -> {
+                // 비동기 응답(알림 SSE)이 끝날 때 서블릿 컨테이너가 ASYNC 디스패치를 한 번 더 한다.
+                // JWT 필터는 그 디스패치에서 돌지 않아 인증 정보가 비어 있고, 여기서 막으면 이미 커밋된
+                // 응답에 401 을 쓰려다 오류 로그만 남는다. 원래 요청에서 이미 인가를 통과했으므로 연다.
+                authz.dispatcherTypeMatchers(jakarta.servlet.DispatcherType.ASYNC).permitAll();
                 if (!environment.matchesProfiles("prod")) {
                     authz.requestMatchers("/swagger-ui/**", "/swagger-ui.html", "/api-docs/**", "/v3/api-docs/**").permitAll();
                 }

src/main/java/com/carecode/domain/notification/controller/NotificationController.java

@@ -17,6 +17,10 @@
 import com.carecode.domain.notification.dto.response.NotificationDeliveryStatusResponse;
 import com.carecode.domain.notification.app.NotificationFacade;
 import com.carecode.domain.notification.entity.Notification;
+import com.carecode.domain.notification.realtime.NotificationStreamService;
+import jakarta.servlet.http.HttpServletResponse;
+import org.springframework.http.MediaType;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.tags.Tag;
@@ -43,6 +47,19 @@ public class NotificationController extends BaseController {
 
     private final NotificationFacade notificationFacade;
     private final CurrentUserFacade currentUserFacade;
+    private final NotificationStreamService notificationStreamService;
+
+    // 실시간 수신 (SSE)
+    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    @Operation(summary = "알림 실시간 수신 (SSE)",
+            description = "연결 직후 connected 이벤트, 새 알림마다 notification 이벤트(JSON, id=알림 id)가 온다. "
+                    + "25초마다 heartbeat 주석. 끊기면 다시 연결하고 목록을 새로 불러온다.")
+    public SseEmitter stream(HttpServletResponse response) {
+        // Nginx 가 이벤트를 모아 두지 않고 바로 흘려보내게 한다.
+        response.setHeader("X-Accel-Buffering", "no");
+        response.setHeader("Cache-Control", "no-cache");
+        return notificationStreamService.connect(currentUserFacade.requireCurrentUserDbId());
+    }
 
     // 알림 목록 조회
     @GetMapping

src/main/java/com/carecode/domain/notification/entity/Notification.java

@@ -13,6 +13,7 @@
 /** 알림 엔티티 사용자에게 전송되는 알림을 관리합니다. 단순한 구조로 필수 기능만 포함합니다. */
 @Entity
 @Table(name = "TBL_NOTIFICATION")
+@EntityListeners(com.carecode.domain.notification.realtime.NotificationPersistListener.class)
 @Getter
 @Setter
 @NoArgsConstructor

src/main/java/com/carecode/domain/notification/realtime/NotificationCreatedEvent.java

@@ -0,0 +1,7 @@
+package com.carecode.domain.notification.realtime;
+
+import com.carecode.domain.notification.dto.response.NotificationInfoResponse;
+
+/** 알림 한 건이 저장됐다. 커밋 후 실시간 채널로 내보낸다. */
+public record NotificationCreatedEvent(Long userDbId, NotificationInfoResponse payload) {
+}

src/main/java/com/carecode/domain/notification/realtime/NotificationPersistListener.java

@@ -0,0 +1,44 @@
+package com.carecode.domain.notification.realtime;
+
+import com.carecode.domain.notification.dto.response.NotificationInfoResponse;
+import com.carecode.domain.notification.entity.Notification;
+import jakarta.persistence.PostPersist;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.stereotype.Component;
+
+/**
+ * 알림이 저장되는 순간을 잡는다.
+ *
+ * <p>알림을 만드는 곳이 여럿이다 (서비스, 관리자 발송, 빈자리 알림, 초기 데이터 …). 각각에 발송 코드를
+ * 넣으면 새 경로가 생길 때 빠뜨린다. 엔티티 저장에 걸면 어느 경로로 만들어도 실시간으로 나간다.
+ *
+ * <p>여기서는 이벤트만 올린다. 실제 전송은 커밋 뒤({@link NotificationStreamService})에 한다 —
+ * 롤백된 알림이 화면에 떴다가 목록에서 사라지는 일을 막는다.
+ * 사용자 식별은 DB id 만 쓴다. 지연 로딩 프록시라도 id 는 초기화 없이 읽힌다.
+ */
+@Component
+@RequiredArgsConstructor
+public class NotificationPersistListener {
+
+    private final ApplicationEventPublisher publisher;
+
+    @PostPersist
+    public void onPersist(Notification notification) {
+        if (notification.getUser() == null || notification.getUser().getId() == null) {
+            return;
+        }
+        publisher.publishEvent(new NotificationCreatedEvent(
+                notification.getUser().getId(),
+                NotificationInfoResponse.builder()
+                        .id(notification.getId())
+                        .notificationType(notification.getNotificationType() != null
+                                ? notification.getNotificationType().name() : null)
+                        .title(notification.getTitle())
+                        .message(notification.getMessage())
+                        .isRead(Boolean.TRUE.equals(notification.getIsRead()))
+                        .createdAt(notification.getCreatedAt())
+                        .sentAt(notification.getCreatedAt())
+                        .build()));
+    }
+}

src/main/java/com/carecode/domain/notification/realtime/NotificationStreamService.java

@@ -0,0 +1,121 @@
+package com.carecode.domain.notification.realtime;
+
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.MediaType;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+import java.util.Deque;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+
+/**
+ * 알림 실시간 채널 (SSE).
+ *
+ * <p>WebSocket 이 아니라 SSE 를 쓰는 이유: 알림은 서버에서 클라이언트로 가는 한 방향이다. SSE 는 평범한 HTTP 라
+ * 기존 JWT 필터·CORS·접근제어가 그대로 적용되고, 프록시·로드밸런서 설정도 따로 필요 없다.
+ *
+ * <p>한계: 연결을 이 인스턴스 메모리에 둔다. 지금 운영은 단일 인스턴스라 충분하다. 여러 대로 늘리면
+ * Redis pub/sub 로 이벤트를 모든 인스턴스에 퍼뜨려야 한다 ({@code docs/features/realtime-notifications.md}).
+ */
+@Slf4j
+@Service
+public class NotificationStreamService {
+
+    private final Map<Long, Deque<SseEmitter>> emitters = new ConcurrentHashMap<>();
+    private final long timeoutMillis;
+    private final int maxConnectionsPerUser;
+
+    public NotificationStreamService(
+            @Value("${app.notification.stream.timeout-millis:1800000}") long timeoutMillis,
+            @Value("${app.notification.stream.max-connections-per-user:5}") int maxConnectionsPerUser,
+            MeterRegistry meterRegistry) {
+        this.timeoutMillis = timeoutMillis;
+        this.maxConnectionsPerUser = maxConnectionsPerUser;
+        Gauge.builder("carecode.notification.stream.connections", this, NotificationStreamService::connectionCount)
+                .description("열려 있는 알림 실시간(SSE) 연결 수")
+                .register(meterRegistry);
+    }
+
+    /**
+     * 연결을 연다. 탭을 여러 개 열 수 있으므로 사용자당 여러 연결을 허용하되 상한을 둔다
+     * (끊긴 줄 모르고 재연결을 반복하는 클라이언트가 연결을 무한히 쌓지 못하게). 넘치면 가장 오래된 것을 닫는다.
+     */
+    public SseEmitter connect(Long userDbId) {
+        SseEmitter emitter = new SseEmitter(timeoutMillis);
+        Deque<SseEmitter> userEmitters = emitters.computeIfAbsent(userDbId, id -> new ConcurrentLinkedDeque<>());
+        userEmitters.addLast(emitter);
+        while (userEmitters.size() > maxConnectionsPerUser) {
+            SseEmitter oldest = userEmitters.pollFirst();
+            if (oldest != null) {
+                oldest.complete();
+            }
+        }
+
+        Runnable remove = () -> remove(userDbId, emitter);
+        emitter.onCompletion(remove);
+        emitter.onTimeout(remove);
+        emitter.onError(e -> remove.run());
+
+        // 첫 이벤트를 바로 보내야 프록시가 응답 헤더를 흘려보내고, 클라이언트도 연결됐음을 안다.
+        // 재연결 뒤에는 이 이벤트를 신호로 목록을 다시 불러 끊긴 사이의 알림을 채운다.
+        send(userDbId, emitter, SseEmitter.event().name("connected").data("ok"));
+        return emitter;
+    }
+
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)
+    public void onNotificationCreated(NotificationCreatedEvent event) {
+        Deque<SseEmitter> userEmitters = emitters.get(event.userDbId());
+        if (userEmitters == null || userEmitters.isEmpty()) {
+            return;
+        }
+        for (SseEmitter emitter : List.copyOf(userEmitters)) {
+            // id 는 알림 id. 클라이언트는 같은 id 를 두 번 받아도 한 번만 반영한다.
+            send(event.userDbId(), emitter, SseEmitter.event()
+                    .id(String.valueOf(event.payload().getId()))
+                    .name("notification")
+                    .data(event.payload(), MediaType.APPLICATION_JSON));
+        }
+    }
+
+    /** 유휴 연결을 프록시가 끊지 않게 주석 한 줄을 보낸다. 끊긴 연결도 여기서 정리된다. */
+    @Scheduled(fixedDelayString = "${app.notification.stream.heartbeat-millis:25000}")
+    public void heartbeat() {
+        emitters.forEach((userDbId, userEmitters) -> {
+            for (SseEmitter emitter : List.copyOf(userEmitters)) {
+                send(userDbId, emitter, SseEmitter.event().comment("ping"));
+            }
+        });
+    }
+
+    public int connectionCount() {
+        return emitters.values().stream().mapToInt(Deque::size).sum();
+    }
+
+    private void send(Long userDbId, SseEmitter emitter, SseEmitter.SseEventBuilder event) {
+        try {
+            emitter.send(event);
+        } catch (IOException | IllegalStateException e) {
+            // 클라이언트가 이미 떠났다. 정상 상황이라 경고로 남기지 않는다.
+            log.debug("SSE 전송 실패로 연결을 정리합니다 - userDbId={}", userDbId);
+            remove(userDbId, emitter);
+            emitter.completeWithError(e);
+        }
+    }
+
+    private void remove(Long userDbId, SseEmitter emitter) {
+        emitters.computeIfPresent(userDbId, (id, userEmitters) -> {
+            userEmitters.remove(emitter);
+            return userEmitters.isEmpty() ? null : userEmitters;
+        });
+    }
+}