From 8b04cb7a89929b875ad1e89151ba20963161f1d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=A1=B0=EC=9E=AC=EC=A4=91?= <126754298+m-a-king@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:37:53 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20SSE=20notification=20=EC=9D=B4=EB=B2=A4?= =?UTF-8?q?=ED=8A=B8=20id=20=EB=B6=80=EC=97=AC=20+=20Last-Event-ID=20?= =?UTF-8?q?=EC=9E=AC=EC=97=B0=EA=B2=B0=20replay?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - notification 이벤트에만 SSE id(알림 PK)를 싣는다. 별도 카운터 대신 DB 영속 값을 써 재시작·blue-green 과 무관하게 단조 증가하고, payload 의 id 와 같은 값이라 클라 dedup 키를 겸한다 - id 없는 이벤트(connect·silent-sync·하트비트)는 SSE 프로토콜상 클라 lastEventId 를 갱신하지 않으므로, 비영속이라 replay 불가능한 이벤트가 복구 기준점에서 자연 제외된다 - subscribe 가 Last-Event-ID 헤더를 받아 초과분을 발생 순서(id asc)로 그 연결에만 replay 한다. register 후 replay 순서로 "유실 대신 중복"을 택했고, 클라가 id 로 dedup 하는 계약을 스펙에 명시 - 상한(100건) 초과 공백은 일부만 보내면 replay 구간 뒤에 조용한 구멍이 남으므로 통째로 생략하고 기존 목록/배지 재조회 계약에 맡긴다 - 숫자가 아닌 Last-Event-ID 는 400 으로 끊지 않고 첫 연결로 취급한다 — EventSource 자동 재연결 루프를 깨지 않기 위함 - replay 조회는 기존 idx_notifications_user_id_id (user_id, id) 인덱스가 그대로 받쳐 마이그레이션이 없다 --- notification-sse-spec.md | 40 ++++- .../controller/NotificationSseApi.kt | 17 ++- .../controller/NotificationSseController.kt | 15 ++ .../repository/NotificationJpaRepository.kt | 8 + .../repository/NotificationRepository.kt | 7 + .../repository/NotificationRepositoryImpl.kt | 7 + .../piki/notification/sse/LocalSseDelivery.kt | 35 ++++- .../notification/sse/SseReconnectReplayer.kt | 48 ++++++ .../sse/NotificationSseIntegrationTest.kt | 141 ++++++++++++++++++ 9 files changed, 302 insertions(+), 16 deletions(-) create mode 100644 src/main/kotlin/com/depromeet/piki/notification/sse/SseReconnectReplayer.kt diff --git a/notification-sse-spec.md b/notification-sse-spec.md index 04c77cf0..63c229e5 100644 --- a/notification-sse-spec.md +++ b/notification-sse-spec.md @@ -64,10 +64,14 @@ data: connected 알림이 발생할 때마다 전송. `data` 는 아래 [5. payload](#5-notification-payload-스키마) JSON. ```text +id: 123 event: notification data: {"id":123,"type":"TOURNAMENT_JOINED","category":"ACTIVITY","title":"홍길동님이 참가했어요","body":"","imageUrl":"https://.../profiles/{uid}.png","refId":45,"isRead":false,"createdAt":"2026-06-06T14:32:10"} ``` +- SSE **`id` 필드에 알림 id** 가 실린다 (`data` 의 `id` 와 같은 값). 재연결 시 놓친 알림을 다시 받는 키다 — [7. 끊김 중 발생한 알림](#끊김-중-발생한-알림--last-event-id-replay) 참조. +- `notification` **에만** `id` 가 실린다. SSE 프로토콜상 `id` 없는 이벤트(`connect`·`silent-sync`·하트비트)는 클라이언트의 lastEventId 를 갱신하지 않으므로, 재연결 복구 기준점은 항상 "마지막으로 받은 알림"이 된다. + ### (3) `silent-sync` — 조용한 화면 갱신 신호 (알림 아님) **이건 "알림"이 아니라 화면 갱신 신호다.** 알림센터에 쌓이지 않고 FCM 표시 푸시도 가지 않으며, 오직 열려 있는 SSE 연결로만 흐른다(`notification` 과 구분 — `silent` 은 토스트 없이 조용히 화면만 갱신함을 뜻한다). 한 채널로 여러 갱신 사건이 흐르고, 클라는 payload 의 **`type`** 으로 사건을 분기한다. `data` 는 아래 [5-1. payload](#5-1-silent-sync-payload-스키마) JSON. @@ -210,12 +214,22 @@ data: {"type":"UNREAD_COUNT_CHANGED","unreadCount":1,"unreadCountByCategory":{"A - WEB(쿠키 인증): `access_token` 쿠키가 만료되면 `/api/v1/auth/token/refresh` 로 갱신(쿠키 재발급) 후 재연결. - APP(헤더 인증): refresh 로 새 access token 을 받아 **재연결 요청의 `Authorization` 헤더에 새 토큰**을 실어 연다. -### 끊김 중 발생한 알림 (현재 한계) +### 끊김 중 발생한 알림 — Last-Event-ID replay + +재연결 요청에 **`Last-Event-ID` 헤더**(마지막으로 받은 `notification` 의 SSE `id`)를 실으면, 서버가 끊김 동안 쌓인 `notification` 을 **발생 순서대로(오래된 것부터)** 그 연결에 다시 흘려보낸다. 브라우저 `EventSource` 는 이 헤더를 **자동으로** 싣는다(별도 구현 불필요). APP 직접 구현은 마지막으로 받은 이벤트의 `id` 를 저장했다가 재연결 요청 헤더에 싣는다. + +```text +GET /api/v1/notifications/subscribe +Last-Event-ID: 123 +``` + +- **replay 대상은 `notification` 만.** `silent-sync` 는 영속되지 않아 재전송되지 않는다 — 재연결·앱 진입 시 토너먼트 아이템 목록을 재조회하면 그 시점의 최신 `status`(READY/FAILED)가 그대로 내려오므로 자연히 따라잡힌다. +- **replay 상한 100건.** 끊김 동안 쌓인 알림이 100건을 넘으면(장기 미접속) replay 없이 라이브 스트림만 시작된다 — 그 공백은 목록/배지 조회 API 재조회로 복구한다. +- **중복 도착 가능 — `id` 로 dedup 할 것.** 서버는 유실을 막는 쪽을 택해(연결 등록 후 replay 조회) 재연결 직후 발생한 알림이 라이브·replay 양쪽으로 겹칠 수 있다. 같은 `id` 는 같은 알림이므로 클라이언트가 버리면 된다. +- **드문 한계**: 알림 id 발급 순서와 커밋(전송) 순서가 어긋나는 희귀한 경합에서는 replay 가 그 사이 알림을 놓칠 수 있다. 아래 재조회 fallback 이 그대로 커버한다. +- `Last-Event-ID` 가 없거나 숫자가 아니면 첫 연결로 취급한다(replay 없음, 연결은 정상 성립). -- 현재 SSE 는 **연결 중에 발생한 알림만** 실시간 전달한다. **연결이 끊겨 있던 동안 쌓인 알림은 재연결해도 스트림으로 다시 오지 않는다.** -- 단, 모든 알림은 DB(`notifications`)에 영속되므로, **목록/배지 조회 API** 로 놓친 알림을 따라잡는 설계가 정석이다. (해당 조회 API 는 후속 작업) -- `silent-sync`(조용한 화면 갱신)는 **영속되지 않는다** — 끊겨 있던 동안의 파싱 완료/실패는 재전송되지 않는다. 하지만 재연결·앱 진입 시 토너먼트 아이템 목록을 재조회하면 그 시점의 최신 `status`(READY/FAILED)가 그대로 내려오므로 자연히 따라잡힌다. -- 따라서 권장 클라이언트 패턴: **앱 진입/재연결 시 목록 API 로 동기화 + SSE 로 실시간 갱신.** +**권장 클라이언트 패턴은 그대로다: 앱 진입/재연결 시 목록/배지 API 로 동기화 + SSE 로 실시간 갱신.** replay 는 짧은 끊김(30분 타임아웃 재연결 등)의 유실을 스트림 차원에서 메워주는 보강이지, 재조회 동기화를 대체하지 않는다. --- @@ -274,10 +288,13 @@ es.addEventListener("silent-sync", (e) => { }); es.onerror = () => { - // EventSource 가 자동 재연결을 시도한다. 필요 시 추가 처리. + // EventSource 가 자동 재연결을 시도한다. 이때 Last-Event-ID 헤더(마지막 notification 의 id)도 + // 자동으로 실리므로, 끊김 동안 놓친 알림이 재연결 직후 replay 로 도착한다. 필요 시 추가 처리. }; ``` +> `notification` 리스너는 replay 로 같은 알림이 중복 도착할 수 있으므로 `n.id` 기준으로 dedup 한다 (예: 최근 처리한 id Set 보관). + ### APP (헤더 인증 + 재연결 시 토큰 갱신) 브라우저 `EventSource` 와 달리 앱은 네이티브 HTTP 스트림(Android `OkHttp-sse`, iOS `URLSession` 등)을 직접 다루므로 `Authorization` 헤더를 자유롭게 실을 수 있다. 핵심은 **재연결할 때마다 유효한 토큰으로 다시 여는 것**. @@ -289,14 +306,22 @@ fun openSse() { val request = Request.Builder() .url("$BASE_URL/api/v1/notifications/subscribe") .header("Authorization", "Bearer $token") + .apply { + // 재연결이면 마지막으로 받은 notification 의 id 를 실어 끊김 구간을 replay 받는다. + // (브라우저 EventSource 는 자동이지만 직접 구현은 저장·전송을 직접 한다) + lastEventIdStore.get()?.let { header("Last-Event-ID", it) } + } .build() EventSources.createFactory(client).newEventSource(request, object : EventSourceListener() { override fun onEvent(es: EventSource, id: String?, type: String?, data: String) { + // notification 이벤트에만 id 가 실린다 — 올 때마다 저장해 두면 그게 곧 재연결 기준점이다. + id?.let { lastEventIdStore.save(it) } when (type) { "connect" -> { /* 연결됨 */ } "notification" -> { val n = json.decode(data) + if (!seenNotificationIds.add(n.id)) return // replay 중복 dedup (id 기준) // type 으로 분기. 파싱 알림(ITEM_PARSING_*)은 kind 로 출처를 가른다: // kind == "TOURNAMENT" -> tournamentId 로 입장 후 tournamentItemId 로 그 아이템 지목 // kind == "WISH" -> /archive @@ -338,6 +363,7 @@ fun openSse() { - [ ] 파싱 알림(`ITEM_PARSING_*`)은 `kind` 로 출처 분기 (WISH → `/archive`, TOURNAMENT → `tournamentId`·`tournamentItemId`) - [ ] `silent-sync` 는 알림이 아닌 **화면 갱신 신호** — payload 의 `type` 으로 분기 (`TOURNAMENT_ITEM_PARSED` 면 `(tournamentId, tournamentItemId)` 로 카드를 찾아 `status`(READY/FAILED) 반영). 토스트·알림센터 아님 - [ ] 주석 `: ping` 은 무시 (data 이벤트 아님) -- [ ] 재연결 시 목록 API 로 놓친 알림 동기화 +- [ ] 재연결 시 `Last-Event-ID` 로 놓친 `notification` replay 받기 (WEB 은 자동, APP 은 마지막 id 저장·전송 직접) + 목록 API 동기화 병행 +- [ ] `notification` 은 `id` 기준 dedup (replay 와 라이브가 겹쳐 중복 도착 가능) - [ ] WEB 은 쿠키 인증(`withCredentials`), APP 은 `Authorization` 헤더 - [ ] **재연결 시 토큰이 만료됐으면 refresh 후 새 토큰으로 연결** (연결 도중 만료는 무관, 재연결 시점이 관건) diff --git a/src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseApi.kt b/src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseApi.kt index 74bdd64a..19f90160 100644 --- a/src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseApi.kt +++ b/src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseApi.kt @@ -3,6 +3,7 @@ package com.depromeet.piki.notification.controller import com.depromeet.piki.common.response.ApiResponseBody import io.swagger.v3.oas.annotations.Operation import io.swagger.v3.oas.annotations.Parameter +import io.swagger.v3.oas.annotations.enums.ParameterIn import io.swagger.v3.oas.annotations.media.Content import io.swagger.v3.oas.annotations.media.Schema import io.swagger.v3.oas.annotations.responses.ApiResponse @@ -23,13 +24,18 @@ interface NotificationSseApi { "|---|---|---|\n" + "| `connect` | 구독 직후 1회 | `data=\"connected\"`. 연결 성립 신호 |\n" + "| `notification` | 알림 1건마다 | `type` 으로 화면을, 파싱 알림은 `kind` 로 출처(위시/토너먼트)를 분기. " + + "SSE `id` 필드에 알림 id 가 실린다(재연결 복구 키). " + "출처별 payload 셰입과 라우팅 필드(`kind`·`tournamentId`·`tournamentItemId`)는 `notification-sse-spec.md` 참조 |\n" + "| `silent-sync` | 화면 갱신 사건마다 | 조용한 화면 갱신 신호(알림 아님). payload 의 `type` 으로 사건을 분기한다: " + "`TOURNAMENT_ITEM_PARSED`(`{type, tournamentId, tournamentItemId, status}`, status=`READY`\\|`FAILED`) · " + "`UNREAD_COUNT_CHANGED`(`{type, unreadCount, unreadCountByCategory}`, 읽음 후 멀티 디바이스 인앱 배지 동기화). 알림센터·푸시 없이 SSE 로만 흐른다. `notification-sse-spec.md` 참조 |\n" + "| `(주석 ping)` | 약 30초 간격 | 하트비트. 연결 유지용이며 data 이벤트가 아니다 |\n\n" + "- 토너먼트 알림은 해당 토너먼트 참여자에게만 fan-out 되므로, **자기 스트림 1개만 구독**하면 토너먼트·개인 알림이 모두 도착한다.\n" + - "- 연결은 **30분 후 타임아웃**되며, 클라이언트는 끊기면 재연결한다.", + "- 연결은 **30분 후 타임아웃**되며, 클라이언트는 끊기면 재연결한다.\n" + + "- **재연결 시 `Last-Event-ID` 헤더**(마지막으로 받은 `notification` 의 SSE `id`)를 보내면 끊김 동안 쌓인 " + + "`notification` 을 발생 순서대로 다시 흘려보낸다(최대 100건 — 초과 공백은 replay 없이 목록/배지 API 재조회로 복구). " + + "라이브 전송과 겹쳐 같은 알림이 중복 도착할 수 있으므로 클라이언트는 `id` 로 dedup 한다. " + + "`silent-sync` 는 비영속이라 replay 되지 않는다. 상세 계약은 `notification-sse-spec.md` 참조.", ) @ApiResponses( value = [ @@ -54,5 +60,14 @@ interface NotificationSseApi { ) fun subscribe( @Parameter(hidden = true) userId: UUID, + @Parameter( + name = "Last-Event-ID", + `in` = ParameterIn.HEADER, + required = false, + description = + "재연결 시 마지막으로 받은 `notification` 이벤트의 SSE `id`(알림 id). " + + "브라우저 `EventSource` 는 자동으로 싣고, APP 직접 구현은 마지막 id 를 저장했다가 싣는다. " + + "없거나 숫자가 아니면 첫 연결로 취급한다(replay 없음).", + ) lastEventIdHeader: String?, ): SseEmitter } diff --git a/src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseController.kt b/src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseController.kt index 0eee28ad..0463508c 100644 --- a/src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseController.kt +++ b/src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseController.kt @@ -1,10 +1,12 @@ package com.depromeet.piki.notification.controller import com.depromeet.piki.notification.sse.SseEmitterRegistry +import com.depromeet.piki.notification.sse.SseReconnectReplayer import org.slf4j.LoggerFactory import org.springframework.http.MediaType import org.springframework.security.core.annotation.AuthenticationPrincipal import org.springframework.web.bind.annotation.GetMapping +import org.springframework.web.bind.annotation.RequestHeader import org.springframework.web.bind.annotation.RequestMapping import org.springframework.web.bind.annotation.RestController import org.springframework.web.servlet.mvc.method.annotation.SseEmitter @@ -14,12 +16,14 @@ import java.util.UUID @RequestMapping("/api/v1/notifications") class NotificationSseController( private val registry: SseEmitterRegistry, + private val replayer: SseReconnectReplayer, ) : NotificationSseApi { private val log = LoggerFactory.getLogger(javaClass) @GetMapping("/subscribe", produces = [MediaType.TEXT_EVENT_STREAM_VALUE]) override fun subscribe( @AuthenticationPrincipal userId: UUID, + @RequestHeader(HEADER_LAST_EVENT_ID, required = false) lastEventIdHeader: String?, ): SseEmitter { val emitter = SseEmitter(SSE_TIMEOUT_MS) // 연결 종료(정상 종료·에러·타임아웃) 시 레지스트리에서 제거해 죽은 emitter 누적을 막는다. unregister 는 멱등. @@ -29,6 +33,9 @@ class NotificationSseController( registry.unregister(userId, emitter) emitter.complete() } + // register 를 replay 조회보다 먼저 둔다 — 반대 순서(조회 후 register)면 그 틈에 발행된 라이브 알림이 + // 유실된다. 이 순서로는 같은 알림이 라이브·replay 양쪽으로 겹칠 수 있으나(중복), 클라이언트가 id 로 + // dedup 하는 계약이다(notification-sse-spec.md). 유실보다 중복이 옳다. registry.register(userId, emitter) // 최초 connect 이벤트로 응답 헤더를 즉시 flush 해 클라이언트가 "연결됨" 을 곧장 인지하게 한다. runCatching { @@ -37,7 +44,12 @@ class NotificationSseController( log.warn("SSE 최초 connect 전송 실패 userId={}", userId, e) registry.unregister(userId, emitter) emitter.completeWithError(e) + return emitter } + // 재연결이면(Last-Event-ID) 끊김 동안 놓친 알림을 이 연결에만 replay 한다. + // 헤더 파싱은 보수적으로 — SSE 프로토콜상 서버는 id 재개를 지원하지 않을 수도 있는 optional 계약이라, + // 숫자가 아니거나 양수가 아닌 값은 400 으로 끊지 않고 "첫 연결" 로 취급한다(재연결 루프를 깨지 않는다). + lastEventIdHeader?.toLongOrNull()?.takeIf { it > 0 }?.let { replayer.replayMissed(userId, emitter, it) } return emitter } @@ -47,5 +59,8 @@ class NotificationSseController( // emitter 자체 타임아웃(30분). 만료되면 onTimeout 으로 정리되고 클라이언트가 재연결한다. const val SSE_TIMEOUT_MS = 30 * 60 * 1000L + + // SSE 표준 재연결 헤더 — EventSource 가 마지막 수신 이벤트 id 를 자동으로 싣는다. 값은 notification 의 알림 id. + const val HEADER_LAST_EVENT_ID = "Last-Event-ID" } } diff --git a/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationJpaRepository.kt b/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationJpaRepository.kt index 18b20406..5233163a 100644 --- a/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationJpaRepository.kt +++ b/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationJpaRepository.kt @@ -45,6 +45,14 @@ interface NotificationJpaRepository : JpaRepository { limit: Limit, ): List + // SSE 재연결 replay — Last-Event-ID(마지막 수신 알림 id) 초과분을 오래된 것부터(id asc) limit 건. + // 발생 순서대로 다시 흘려보내야 하므로 목록 페이지(desc)와 정렬이 반대다. (idx_notifications_user_id_id 커버) + fun findByUserIdAndIdGreaterThanOrderByIdAsc( + userId: UUID, + id: Long, + limit: Limit, + ): List + // type 별 안읽음 수 — 전체 badge + 탭별(활동/시스템) badge 를 한 쿼리(group by)로 집계한다. // closed projection(type·count alias)으로 받아 RepositoryImpl 이 카테고리로 접는다. (idx (user_id, is_read) 커버) @Query( diff --git a/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationRepository.kt b/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationRepository.kt index 38581b3b..d1213de7 100644 --- a/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationRepository.kt +++ b/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationRepository.kt @@ -21,6 +21,13 @@ interface NotificationRepository { types: List?, ): List + // SSE 재연결 replay — afterId(클라이언트가 마지막으로 받은 알림 id) 초과분을 발생 순서(id asc)로 최대 limit 건. + fun findAfterId( + userId: UUID, + afterId: Long, + limit: Int, + ): List + // 카테고리별 안읽음 수(탭별 badge). 모든 카테고리를 키로 포함하며 해당 없는 카테고리는 0 이다. // 전체 안읽음 수(앱 badge)는 이 맵의 값 합으로 도출한다 — 두 수치가 어긋날 여지를 없앤다. fun countUnreadByCategory(userId: UUID): Map diff --git a/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationRepositoryImpl.kt b/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationRepositoryImpl.kt index a5a66b3e..d657c98a 100644 --- a/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationRepositoryImpl.kt +++ b/src/main/kotlin/com/depromeet/piki/notification/repository/NotificationRepositoryImpl.kt @@ -47,6 +47,13 @@ class NotificationRepositoryImpl( return notificationJpaRepository.findByUserIdAndIdLessThanAndTypeInOrderByIdDesc(userId, cursor.lastNotificationId, types, limited) } + override fun findAfterId( + userId: UUID, + afterId: Long, + limit: Int, + ): List = + notificationJpaRepository.findByUserIdAndIdGreaterThanOrderByIdAsc(userId, afterId, Limit.of(limit)) + // type 별 group-by 결과를 카테고리로 접는다. 모든 카테고리를 0 으로 깔고 해당 type 의 수를 카테고리에 누적해, // 안읽음이 없는 카테고리도 키로 0 을 보장한다(FE 가 탭 badge 를 항상 읽을 수 있게). override fun countUnreadByCategory(userId: UUID): Map { diff --git a/src/main/kotlin/com/depromeet/piki/notification/sse/LocalSseDelivery.kt b/src/main/kotlin/com/depromeet/piki/notification/sse/LocalSseDelivery.kt index 103d2125..4ebe4ca1 100644 --- a/src/main/kotlin/com/depromeet/piki/notification/sse/LocalSseDelivery.kt +++ b/src/main/kotlin/com/depromeet/piki/notification/sse/LocalSseDelivery.kt @@ -28,14 +28,33 @@ class LocalSseDelivery( userId: UUID, notification: Notification, ) { - val event = - SseEmitter - .event() - .name(EVENT_NOTIFICATION) - .data(NotificationSsePayload.from(notification, defaultPushImage.url)) + val event = notificationEvent(notification) registry.emittersOf(userId).forEach { sendOrEvict(userId, it, event) } } + // 재연결한 특정 연결 하나에만 놓친 알림들을 발생 순서대로 다시 흘려보낸다(라이브 deliver 와 같은 이벤트 셰입). + // write 실패 = 그 연결이 죽은 것이므로 sendOrEvict 가 정리하고, 나머지 replay 는 의미가 없어 중단한다. + fun replayTo( + userId: UUID, + emitter: SseEmitter, + notifications: List, + ) { + for (notification in notifications) { + if (!sendOrEvict(userId, emitter, notificationEvent(notification))) return + } + } + + // notification 이벤트 한 건 — 라이브(deliver)와 replay 가 같은 셰입을 쓰도록 빌드를 한 곳에 모은다. + // SSE id 필드에 알림 PK 를 싣는다: 클라이언트(EventSource)가 마지막 수신 id 를 기억했다가 재연결 시 + // Last-Event-ID 로 보내 놓친 구간을 replay 받는 키다. id 없는 이벤트(connect·silent-sync·하트비트)는 + // SSE 프로토콜상 클라의 lastEventId 를 갱신하지 않으므로, 비영속이라 replay 불가능한 이벤트가 복구 대상에서 자연 제외된다. + private fun notificationEvent(notification: Notification): SseEmitter.SseEventBuilder = + SseEmitter + .event() + .id(notification.getId().toString()) + .name(EVENT_NOTIFICATION) + .data(NotificationSsePayload.from(notification, defaultPushImage.url)) + // 조용한(silent) 화면 갱신 신호를 대상 유저들 연결에 실시간 흘려보낸다(알림 아님 — 토스트·알림센터·FCM 표시 푸시 없이 SSE 로만). // notification 전달과 같은 emitter write 경로(sendOrEvict)를 공유하되 이벤트 name 만 다르다(클라가 name 으로 구분). // 한 payload 이벤트를 만들어 대상 유저 전원의 emitter 에 재사용한다 — 같은 갱신을 보는 화면이 모두 동일하게 반영된다. @@ -72,11 +91,12 @@ class LocalSseDelivery( // write 실패 = 죽은 연결(클라이언트가 끊겼는데 onError/onCompletion 콜백이 아직 안 탄 경우 등). // 레지스트리에서 빼 더 흘려보내지 않게 하고 completeWithError 로 정리를 마무리한다. completeWithError 가 // onError 콜백(컨트롤러의 unregister)을 다시 깨울 수 있으나 unregister 는 멱등이라 중복 호출이 무해하다. + // 성공 여부를 돌려줘 한 연결에 연속 write 하는 replay 가 죽은 연결에 헛 write 를 반복하지 않게 한다. private fun sendOrEvict( userId: UUID, emitter: SseEmitter, event: SseEmitter.SseEventBuilder, - ) { + ): Boolean = runCatching { emitter.send(event) } .onFailure { e -> // 클라이언트가 연결을 끊은 정상 종료는 SSE 의 일상적 라이프사이클이라 DEBUG 로 둔다. @@ -89,8 +109,7 @@ class LocalSseDelivery( } registry.unregister(userId, emitter) runCatching { emitter.completeWithError(e) } - } - } + }.isSuccess companion object { // SSE 이벤트 name. 클라이언트는 이 이름으로 알림 이벤트와 connect/하트비트를 구분한다. diff --git a/src/main/kotlin/com/depromeet/piki/notification/sse/SseReconnectReplayer.kt b/src/main/kotlin/com/depromeet/piki/notification/sse/SseReconnectReplayer.kt new file mode 100644 index 00000000..2aca3d76 --- /dev/null +++ b/src/main/kotlin/com/depromeet/piki/notification/sse/SseReconnectReplayer.kt @@ -0,0 +1,48 @@ +package com.depromeet.piki.notification.sse + +import com.depromeet.piki.notification.repository.NotificationRepository +import org.slf4j.LoggerFactory +import org.springframework.stereotype.Component +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter +import java.util.UUID + +// 재연결한 SSE 연결에 끊김 동안 쌓인 notification 을 다시 흘려보낸다(Last-Event-ID 기반 catch-up). +// notification 은 DB 영속이라 어느 인스턴스가 재연결을 받아도 같은 결과를 replay 할 수 있다 — +// 인메모리 버퍼가 아니라 DB 를 원천으로 삼아 재시작·blue-green·스케일아웃과 무관하다. +// +// 조회는 repository 안의 짧은 트랜잭션으로 끝나고 emitter write(외부 I/O)는 트랜잭션 밖이다 +// (CLAUDE.md "외부 호출은 트랜잭션 밖에서" 와 같은 결 — 느린 클라이언트 write 가 DB 커넥션을 잡지 않는다). +@Component +class SseReconnectReplayer( + private val notificationRepository: NotificationRepository, + private val localDelivery: LocalSseDelivery, +) { + private val log = LoggerFactory.getLogger(javaClass) + + // lastEventId(클라이언트가 마지막으로 받은 알림 id) 초과분을 발생 순서대로 그 연결에만 replay 한다. + // + // 상한 초과(끊김이 아주 길었던 경우)면 replay 를 통째로 생략한다 — 일부만 보내면 replay 구간 뒤에 + // 조용한 구멍이 남아 "받은 만큼은 연속" 이라는 계약이 깨진다. 생략 시 복구는 기존 계약(재연결 시 + // 목록/배지 API 재조회)이 그대로 책임진다. 상한은 초과 감지를 위해 limit+1 건을 조회해 판정한다. + fun replayMissed( + userId: UUID, + emitter: SseEmitter, + lastEventId: Long, + ) { + val missed = notificationRepository.findAfterId(userId, lastEventId, REPLAY_LIMIT + 1) + if (missed.isEmpty()) return + if (missed.size > REPLAY_LIMIT) { + // 유실 복구가 replay 대신 목록 재조회로 넘어가는 지점 — 빈도를 봐야 상한 적정성을 판단할 수 있어 info. + log.info("SSE replay 생략(상한 {} 초과) userId={} lastEventId={}", REPLAY_LIMIT, userId, lastEventId) + return + } + localDelivery.replayTo(userId, emitter, missed) + log.info("SSE replay {}건 userId={} lastEventId={}", missed.size, userId, lastEventId) + } + + companion object { + // 한 번의 재연결에 replay 하는 최대 알림 수. 유저당 알림 발생량(토너먼트·파싱 사건)에 비해 넉넉한 값으로, + // 이 상한을 넘는 공백은 "짧은 끊김" 이 아니라 장기 미접속이라 목록 재조회가 맞는 복구 경로다. + const val REPLAY_LIMIT = 100 + } +} diff --git a/src/test/kotlin/com/depromeet/piki/notification/sse/NotificationSseIntegrationTest.kt b/src/test/kotlin/com/depromeet/piki/notification/sse/NotificationSseIntegrationTest.kt index afe79266..e462ec36 100644 --- a/src/test/kotlin/com/depromeet/piki/notification/sse/NotificationSseIntegrationTest.kt +++ b/src/test/kotlin/com/depromeet/piki/notification/sse/NotificationSseIntegrationTest.kt @@ -2,6 +2,7 @@ package com.depromeet.piki.notification.sse import com.depromeet.piki.auth.infrastructure.jwt.JwtProvider import com.depromeet.piki.notification.controller.dto.NotificationSsePayload +import com.depromeet.piki.notification.controller.dto.UnreadCountChanged import com.depromeet.piki.notification.domain.Notification import com.depromeet.piki.notification.domain.NotificationCategory import com.depromeet.piki.notification.domain.NotificationKind @@ -34,6 +35,7 @@ import tools.jackson.databind.ObjectMapper import java.util.UUID import java.util.concurrent.CopyOnWriteArrayList import kotlin.test.assertEquals +import kotlin.test.assertFalse import kotlin.test.assertIs import kotlin.test.assertTrue @@ -54,6 +56,8 @@ class NotificationSseIntegrationTest : IntegrationTestSupport() { @Autowired private lateinit var sseNotificationChannel: SseNotificationChannel + @Autowired private lateinit var localSseDelivery: LocalSseDelivery + @Autowired private lateinit var channels: List @Autowired private lateinit var notificationRepository: NotificationRepository @@ -158,6 +162,8 @@ class NotificationSseIntegrationTest : IntegrationTestSupport() { assertEquals("나이키 에어맥스가 추가됐어요", payload.body) // SSE 이벤트 name 이 notification 으로 실린다. assertTrue(emitter.sentData.any { it is String && it.contains("event:notification") }) + // SSE id 필드에 알림 id 가 실린다 — 클라이언트 재연결(Last-Event-ID) 복구의 기준점. + assertTrue(emitter.sentData.any { it is String && it.contains("id:${notification.getId()}\n") }) } finally { registry.unregister(userId, emitter) } @@ -290,6 +296,141 @@ class NotificationSseIntegrationTest : IntegrationTestSupport() { // Jackson3+kotlin module 은 isRead 필드명으로 직렬화한다(모듈 없는 Jackson2 였으면 "read" 라 이 키가 없다). assertTrue(node.has("isRead")) } + + // --- Last-Event-ID 재연결 replay (#750) --- + // replay 는 구독 시점에 컨트롤러 스레드에서 동기로 일어나므로, asyncStarted 상태의 응답 버퍼(contentAsString)로 + // 실제 와이어(text/event-stream)에 흐른 이벤트를 그대로 단언한다. SSE id 필드 라인은 "id:{알림id}\n" 형식이라 + // data JSON 의 "id":{알림id} 와 substring 이 겹치지 않는다. + + @Test + fun `Last-Event-ID 를 실어 재연결하면 놓친 알림만 발생 순서대로 replay 된다`() { + val userId = UUID.randomUUID() + val received = + notificationRepository.save(Notification(userId, NotificationType.TOURNAMENT_JOINED, "이미 받은 알림", "", 1L)) + val missedFirst = + notificationRepository.save(Notification(userId, NotificationType.TOURNAMENT_ITEM_ADDED, "놓친 알림 1", "", 2L)) + val missedSecond = + notificationRepository.save(Notification(userId, NotificationType.TOURNAMENT_STARTED, "놓친 알림 2", "", 3L)) + + try { + val content = + buildMockMvc() + .perform( + get("/api/v1/notifications/subscribe") + .header(HttpHeaders.AUTHORIZATION, authHeader(userId)) + .header("Last-Event-ID", received.getId().toString()), + ).andExpect(request().asyncStarted()) + .andReturn() + .response.contentAsString + + // 놓친 두 건만 replay 된다 — 이미 받은 알림(Last-Event-ID 그 자체)은 다시 오지 않는다. + assertEquals(2, "event:notification".toRegex().findAll(content).count()) + assertFalse(content.contains("id:${received.getId()}\n")) + // 발생 순서(오래된 것부터, id asc)대로 흐른다. + val firstIndex = content.indexOf("id:${missedFirst.getId()}\n") + val secondIndex = content.indexOf("id:${missedSecond.getId()}\n") + assertTrue(firstIndex >= 0 && secondIndex >= 0 && firstIndex < secondIndex) + } finally { + registry.emittersOf(userId).toList().forEach { registry.unregister(userId, it) } + } + } + + @Test + fun `Last-Event-ID 이후 알림이 없으면 replay 없이 connect 만 온다`() { + val userId = UUID.randomUUID() + val latest = + notificationRepository.save(Notification(userId, NotificationType.TOURNAMENT_JOINED, "마지막 알림", "", 1L)) + + try { + val content = + buildMockMvc() + .perform( + get("/api/v1/notifications/subscribe") + .header(HttpHeaders.AUTHORIZATION, authHeader(userId)) + .header("Last-Event-ID", latest.getId().toString()), + ).andExpect(request().asyncStarted()) + .andReturn() + .response.contentAsString + + assertTrue(content.contains("event:connect")) + assertFalse(content.contains("event:notification")) + } finally { + registry.emittersOf(userId).toList().forEach { registry.unregister(userId, it) } + } + } + + @Test + fun `숫자가 아닌 Last-Event-ID 는 첫 연결로 취급돼 replay 없이 정상 구독된다`() { + val userId = UUID.randomUUID() + notificationRepository.save(Notification(userId, NotificationType.TOURNAMENT_JOINED, "기존 알림", "", 1L)) + + try { + val content = + buildMockMvc() + .perform( + get("/api/v1/notifications/subscribe") + .header(HttpHeaders.AUTHORIZATION, authHeader(userId)) + .header("Last-Event-ID", "not-a-number"), + ).andExpect(request().asyncStarted()) + .andReturn() + .response.contentAsString + + // 재연결 루프를 400 으로 깨지 않고 무시한다 — 연결은 성립하고 replay 만 없다. + assertTrue(content.contains("event:connect")) + assertFalse(content.contains("event:notification")) + } finally { + registry.emittersOf(userId).toList().forEach { registry.unregister(userId, it) } + } + } + + @Test + fun `놓친 알림이 replay 상한을 초과하면 replay 를 통째로 생략한다`() { + val userId = UUID.randomUUID() + val baseline = + notificationRepository.save(Notification(userId, NotificationType.TOURNAMENT_JOINED, "기준점", "", 1L)) + repeat(SseReconnectReplayer.REPLAY_LIMIT + 1) { + notificationRepository.save(Notification(userId, NotificationType.TOURNAMENT_ITEM_ADDED, "밀린 알림", "", 1L)) + } + + try { + val content = + buildMockMvc() + .perform( + get("/api/v1/notifications/subscribe") + .header(HttpHeaders.AUTHORIZATION, authHeader(userId)) + .header("Last-Event-ID", baseline.getId().toString()), + ).andExpect(request().asyncStarted()) + .andReturn() + .response.contentAsString + + // 일부만 보내면 replay 구간 뒤에 조용한 구멍이 남으므로 통째로 생략하고 목록 재조회 계약에 맡긴다. + assertTrue(content.contains("event:connect")) + assertFalse(content.contains("event:notification")) + } finally { + registry.emittersOf(userId).toList().forEach { registry.unregister(userId, it) } + } + } + + @Test + fun `silent-sync 이벤트에는 SSE id 가 실리지 않는다`() { + val userId = UUID.randomUUID() + val emitter = RecordingSseEmitter() + registry.register(userId, emitter) + + try { + localSseDelivery.deliverSilentSync( + listOf(userId), + UnreadCountChanged.of(mapOf(NotificationCategory.ACTIVITY to 1L, NotificationCategory.SYSTEM to 0L)), + ) + + // id 없는 이벤트는 클라이언트 lastEventId 를 갱신하지 않는다 — 비영속이라 replay 불가능한 silent-sync 가 + // 재연결 복구 기준점을 오염시키지 않는 프로토콜 계약. + assertTrue(emitter.sentData.any { it is String && it.contains("event:silent-sync") }) + assertTrue(emitter.sentData.filterIsInstance().none { it.contains("id:") }) + } finally { + registry.unregister(userId, emitter) + } + } } // send(SseEventBuilder) 를 가로채 실제 IO 없이 전송 내용을 기록한다. build() 가 내놓는 data 항목