diff --git a/src/apis/__tests__/notificationStream.test.ts b/src/apis/__tests__/notificationStream.test.ts new file mode 100644 index 0000000..ad42c16 --- /dev/null +++ b/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 }]) + }) +}) diff --git a/src/apis/notificationStream.ts b/src/apis/notificationStream.ts new file mode 100644 index 0000000..ca6d0cd --- /dev/null +++ b/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 { + 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) + } +} diff --git a/src/app/layout.tsx b/src/app/layout.tsx index c66a809..57705be 100644 --- a/src/app/layout.tsx +++ b/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({ -
- {/* 데스크톱 프로모션 패널 */} - + +
+ {/* 데스크톱 프로모션 패널 */} + - {/* 앱 콘텐츠 영역 */} -
-
{children}
+ {/* 앱 콘텐츠 영역 */} +
+
{children}
+
-
+
diff --git a/src/components/common/NotificationStreamListener.tsx b/src/components/common/NotificationStreamListener.tsx new file mode 100644 index 0000000..a94581c --- /dev/null +++ b/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