feat: 알림 실시간 수신 (SSE 클라이언트)

src/apis/__tests__/notificationStream.test.ts

@@ -0,0 +1,35 @@
+import { describe, expect, it } from 'vitest'
+import { SseParser } from '@/apis/notificationStream'
+
+// 서버(Spring SseEmitter)가 실제로 보낸 바이트를 그대로 옮겼다.
+const SERVER_OUTPUT =
+  'event:connected\ndata:ok\n\n' +
+  ':ping\n\n' +
+  'id:1\nevent:notification\ndata:{"id":1,"notificationType":"FACILITY","title":"빈자리 알림","isRead":false}\n\n'
+
+describe('SseParser', () => {
+  it('서버 출력에서 connected·notification 을 읽고 heartbeat 는 건너뛴다', () => {
+    const events = new SseParser().push(SERVER_OUTPUT)
+
+    expect(events).toHaveLength(2)
+    expect(events[0]).toEqual({ event: 'connected', data: 'ok', id: undefined })
+    expect(events[1].event).toBe('notification')
+    expect(events[1].id).toBe('1')
+    expect(JSON.parse(events[1].data).title).toBe('빈자리 알림')
+  })
+
+  it('이벤트가 여러 조각으로 끊겨 와도 완성된 뒤에만 넘긴다', () => {
+    const parser = new SseParser()
+    const pieces = SERVER_OUTPUT.match(/[\s\S]{1,7}/g) ?? []
+
+    const events = pieces.flatMap((piece) => parser.push(piece))
+
+    expect(events.map((e) => e.event)).toEqual(['connected', 'notification'])
+  })
+
+  it('CRLF 줄바꿈과 콜론 뒤 공백도 표준대로 처리한다', () => {
+    const events = new SseParser().push('event: notification\r\ndata: a\r\ndata: b\r\n\r\n')
+
+    expect(events).toEqual([{ event: 'notification', data: 'a\nb', id: undefined }])
+  })
+})

src/apis/notificationStream.ts

@@ -0,0 +1,91 @@
+/**
+ * 알림 실시간 채널 (서버 GET /notifications/stream, text/event-stream).
+ *
+ * 브라우저 EventSource 는 Authorization 헤더를 붙일 수 없다. 액세스 토큰은 메모리에만 두므로
+ * (쿠키가 아님) fetch 스트림으로 직접 읽는다. 서버 계약은 서버 저장소의
+ * docs/features/realtime-notifications.md 에 있다.
+ */
+import { getAccessToken } from '@/apis/auth'
+
+export type SseEvent = {
+  event: string
+  data: string
+  id?: string
+}
+
+/**
+ * 조각난 텍스트를 받아 완성된 이벤트만 돌려준다. 이벤트는 빈 줄로 끝나고,
+ * 네트워크 조각은 이벤트 경계와 무관하게 끊겨 들어오므로 남은 부분을 다음 호출까지 들고 있는다.
+ */
+export class SseParser {
+  private buffer = ''
+
+  push(chunk: string): SseEvent[] {
+    this.buffer += chunk.replace(/\r\n?/g, '\n')
+    const events: SseEvent[] = []
+    let boundary = this.buffer.indexOf('\n\n')
+    while (boundary !== -1) {
+      const block = this.buffer.slice(0, boundary)
+      this.buffer = this.buffer.slice(boundary + 2)
+      const parsed = parseBlock(block)
+      if (parsed) events.push(parsed)
+      boundary = this.buffer.indexOf('\n\n')
+    }
+    return events
+  }
+}
+
+function parseBlock(block: string): SseEvent | null {
+  let event = 'message'
+  let id: string | undefined
+  const data: string[] = []
+  for (const line of block.split('\n')) {
+    // ':' 로 시작하면 주석(서버 heartbeat)이다.
+    if (!line || line.startsWith(':')) continue
+    const colon = line.indexOf(':')
+    const field = colon === -1 ? line : line.slice(0, colon)
+    let value = colon === -1 ? '' : line.slice(colon + 1)
+    if (value.startsWith(' ')) value = value.slice(1)
+    if (field === 'event') event = value
+    else if (field === 'data') data.push(value)
+    else if (field === 'id') id = value
+  }
+  // heartbeat 처럼 필드가 하나도 없는 블록은 이벤트가 아니다.
+  if (data.length === 0 && event === 'message') return null
+  return { event, data: data.join('\n'), id }
+}
+
+export class StreamHttpError extends Error {
+  constructor(readonly status: number) {
+    super(`notification stream responded ${status}`)
+  }
+}
+
+/**
+ * 연결해서 끊길 때까지 이벤트를 넘긴다. 정상 종료(서버 타임아웃)면 resolve, 오류면 reject.
+ * 다시 연결할지는 호출하는 쪽이 정한다.
+ */
+export async function readNotificationStream(
+  onEvent: (event: SseEvent) => void,
+  signal: AbortSignal,
+): Promise<void> {
+  const token = getAccessToken()
+  if (!token) throw new StreamHttpError(401)
+
+  const res = await fetch(`${process.env.NEXT_PUBLIC_API_URL ?? ''}/notifications/stream`, {
+    headers: { Authorization: `Bearer ${token}`, Accept: 'text/event-stream' },
+    credentials: 'include',
+    cache: 'no-store',
+    signal,
+  })
+  if (!res.ok || !res.body) throw new StreamHttpError(res.status)
+
+  const reader = res.body.getReader()
+  const decoder = new TextDecoder()
+  const parser = new SseParser()
+  for (;;) {
+    const { value, done } = await reader.read()
+    if (done) return
+    parser.push(decoder.decode(value, { stream: true })).forEach(onEvent)
+  }
+}

src/app/layout.tsx

@@ -1,6 +1,7 @@
 import type { Metadata, Viewport } from 'next'
 import '@/styles/globals.css'
 import { ReactNode } from 'react'
+import NotificationStreamListener from '@/components/common/NotificationStreamListener'
 import PushListener from '@/components/common/PushListener'
 import SessionBootstrap from '@/components/common/SessionBootstrap'
 import PromotionPanel from '@/components/organism/PromotionPanel'
@@ -37,17 +38,19 @@ export default function RootLayout({
         <QueryProvider>
           <SessionBootstrap>
             <PushListener>
-              <div className="flex min-h-dvh">
-                {/* 데스크톱 프로모션 패널 */}
-                <aside className="hidden sm:block sm:flex-1/3">
-                  <PromotionPanel />
-                </aside>
+              <NotificationStreamListener>
+                <div className="flex min-h-dvh">
+                  {/* 데스크톱 프로모션 패널 */}
+                  <aside className="hidden sm:block sm:flex-1/3">
+                    <PromotionPanel />
+                  </aside>
 
-                {/* 앱 콘텐츠 영역 */}
-                <div className="flex-1 bg-amber-50 sm:flex-2/3">
-                  <div className="mx-auto h-dvh max-w-sm overflow-y-auto">{children}</div>
+                  {/* 앱 콘텐츠 영역 */}
+                  <div className="flex-1 bg-amber-50 sm:flex-2/3">
+                    <div className="mx-auto h-dvh max-w-sm overflow-y-auto">{children}</div>
+                  </div>
                 </div>
-              </div>
+              </NotificationStreamListener>
             </PushListener>
           </SessionBootstrap>
         </QueryProvider>

src/components/common/NotificationStreamListener.tsx

@@ -0,0 +1,54 @@
+'use client'
+import { useQueryClient } from '@tanstack/react-query'
+import { ReactNode, useEffect } from 'react'
+import { getAccessToken } from '@/apis/auth'
+import { readNotificationStream } from '@/apis/notificationStream'
+import { notificationQueries } from '@/queries/notification'
+
+const MIN_RETRY_MS = 1_000
+const MAX_RETRY_MS = 30_000
+
+/**
+ * 로그인해 있는 동안 알림 실시간 채널을 열어 둔다.
+ *
+ * 이벤트 내용을 캐시에 직접 끼워 넣지 않고 알림 목록·배지를 다시 불러온다. 목록이 진실의 원천이라,
+ * 끊긴 사이에 온 알림도(connected 를 받으면 다시 불러오므로) 빠지지 않고 같은 알림을 두 번 받아도
+ * 두 번 보이지 않는다.
+ *
+ * 끊기면 1초부터 두 배씩 최대 30초까지 기다렸다가 다시 연결한다. 로그인 전이면 연결하지 않고 기다린다.
+ */
+const NotificationStreamListener = ({ children }: { children: ReactNode }): ReactNode => {
+  const queryClient = useQueryClient()
+
+  useEffect(() => {
+    const controller = new AbortController()
+    let retryMs = MIN_RETRY_MS
+
+    const refresh = () => queryClient.invalidateQueries({ queryKey: notificationQueries._def })
+
+    const run = async () => {
+      while (!controller.signal.aborted) {
+        if (getAccessToken()) {
+          try {
+            await readNotificationStream((event) => {
+              if (event.event === 'connected') retryMs = MIN_RETRY_MS
+              if (event.event === 'connected' || event.event === 'notification') refresh()
+            }, controller.signal)
+          } catch {
+            // 네트워크 오류·401(토큰 만료 직후) 모두 잠시 뒤 다시 시도한다. 토큰은 자동 갱신된다.
+          }
+        }
+        if (controller.signal.aborted) return
+        await new Promise((resolve) => setTimeout(resolve, retryMs))
+        retryMs = Math.min(retryMs * 2, MAX_RETRY_MS)
+      }
+    }
+    run()
+
+    return () => controller.abort()
+  }, [queryClient])
+
+  return children
+}
+
+export default NotificationStreamListener