조리 대기 → 조리 중 → 조리 완료 → 서빙 완료

SSE 연결
→ full baseline snapshot load
→ snapshot watermark 모음
→ 가장 작은 watermark까지 covered 처리
→ 이후 SSE event patch

EventSource onerror
또는
45초 watchdog timeout
또는
online / focus / visibilitychange

기존 EventSource cleanup
→ backoff + jitter
→ 새 ticket 발급
→ lastAppliedEventId 포함해 reconnect
→ Redis Stream에서 (lastEventId, +) replay
→ dedupe
→ Redux patch

sinceEventId 전달
→ Redis Stream scan
→ changed order/table/menu/call ID 계산

REST command transaction commit
→ outbox row remains unpublished
→ relay publish 실패
→ attempts 증가
→ nextAttemptAt 설정
→ 최대 60초 backoff
→ 이후 retry

POST /api/pos/{storeId}/hall/orders/{orderId}/confirm
  → HallOrderController.confirmOrder()
  → OrderService.confirmOrder()
  → OrderCommandService.confirmOrder()

// PosNotificationPublisher
@Component
class PosNotificationPublisher(... 생략 ...) {
	@TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
	fun handleCookingRequest(event: CookingRequestedEvent) {
	    if (!isCurrentSlotOrderEvent(event.storeId, event.orderId)) return
	    enqueueOutbox(event.storeId, "KITCHEN_COOK", event)
	}
 
	private fun enqueueOutbox(storeId: UUID, type: String, payload: Any) {
	    outboxRepository.save(
	        PosEventOutbox(
	            storeId = storeId,
	            eventType = type,
	            payload = objectMapper.writeValueAsString(payload),
	        )
	    )
	}
}

@Component
class PosOutboxRelay(...) {
	@Scheduled(fixedDelay = 500)
    @Transactional
    fun publishPendingEvents() {
        val events = outboxRepository.lockPublishable(batchSize)
        if (events.isEmpty()) return
 
        events.forEach { event ->
            try {
                val streamId = publishToRedis(event)
                event.markPublished(streamId)
            } catch (e: Exception) {
                event.markFailed(e)
                // log ...
            }
        }
    }
}
 
interface PosEventOutboxRepository : JpaRepository<PosEventOutbox, UUID> {
    @Query(
        value = """
            select *
            from pos_event_outbox
            where published_at is null
              and next_attempt_at <= now()
            order by created_at
            limit :limit
            for update skip locked
        """,
        nativeQuery = true,
    )
    fun lockPublishable(@Param("limit") limit: Int): List<PosEventOutbox>
}

GET /api/pos/{storeId}/notifications/subscribe
  → SseNotificationService.subscribe()
  → registerEmitter()
  → startRedisReader()
  → readStreamLoop()

    fun subscribe(
        expectedStoreId: UUID,
        ticket: String,
        connectionId: String?,
        lastEventId: String?
    ): SseEmitter {
        val storeIdStr = redisTemplate.opsForValue().getAndDelete("ticket:$ticket")
            ?: throw BusinessException(ErrorCode.INVALID_TICKET)
 
        if (storeIdStr != expectedStoreId.toString()) {
            throw BusinessException(ErrorCode.INVALID_TICKET)
        }
 
        val emitter = SseEmitter(60 * 60 * 1000L)
        val chanKey = channelKey(storeIdStr)
        val sKey = streamKey(storeIdStr)
        val safeConnectionId = connectionId?.takeIf { it.isNotBlank() }?.take(128) ?: UUID.randomUUID().toString()
 
        registerEmitter(chanKey, emitter, sKey, safeConnectionId)
        markActive(chanKey)
        logger.info("[SSE-CONNECT] storeId={} connectionId={} lastEventId={}", storeIdStr, safeConnectionId, lastEventId)
 
        enqueueOutbound(chanKey, emitter, SseMsg("connect", "connected", "0-0"))
 
        if (!lastEventId.isNullOrBlank()) {
            replayMissingEvents(sKey, lastEventId, emitter)
            markActive(chanKey)
        }
 
        return emitter
    }
 
    private fun registerEmitter(chanKey: String, emitter: SseEmitter, streamKey: String, connectionId: String) {
        connectionIds[emitter] = connectionId
        outboundQueueOf(emitter)
        startEmitterSender(chanKey, emitter, connectionId)
        emitter.onCompletion { unregisterEmitter(chanKey, emitter, "completion") }
        emitter.onTimeout { emitter.complete(); unregisterEmitter(chanKey, emitter, "timeout") }
        emitter.onError {
            logger.warn("[SSE-ERROR] storeId={} connectionId={}", chanKey, connectionId)
            unregisterEmitter(chanKey, emitter, "error")
        }
 
        emitters.compute(chanKey) { _, list ->
            val newList = list ?: CopyOnWriteArrayList()
            newList.add(emitter)
            newList
        }
 
        startRedisReader(chanKey, streamKey)
    }
 
    private fun startRedisReader(chanKey: String, streamKey: String) {
        streamReaders.compute(chanKey) { _, existing ->
            if (existing != null) return@compute existing
 
            try {
                val startId = currentStreamTailId(streamKey)
                val task = streamReaderExec.submit { readStreamLoop(chanKey, streamKey, startId) }
                logger.info("Redis Stream Reader Started: chanKey={} streamKey={} startId={}", chanKey, streamKey, startId)
                task
            } catch (e: Exception) {
                logger.error("Failed to start redis stream reader", e)
                null
            }
        }
    }
 
    private fun readStreamLoop(chanKey: String, streamKey: String, initialLastSeenId: String) {
        var lastSeenId = initialLastSeenId
        val options = StreamReadOptions.empty()
            .count(100)
            .block(Duration.ofSeconds(2))
 
        while (!Thread.currentThread().isInterrupted && emitters.containsKey(chanKey)) {
            try {
                val records = redisTemplate.opsForStream<String, String>().read(
                    options,
                    StreamOffset.create(streamKey, ReadOffset.from(lastSeenId))
                ) ?: emptyList()
 
                for (record in records) {
                    lastSeenId = record.id.toString()
                    enqueueRecord(chanKey, streamKey, record)
                }
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
            } catch (e: Exception) {
                logger.warn("Redis stream read failed: chanKey={} streamKey={} lastSeenId={}", chanKey, streamKey, lastSeenId)
                enqueue(chanKey, SseMsg("RELOAD_REQUIRED", "{}", "0-0"))
                Thread.sleep(1000)
            }
        }
 
        logger.info("Redis Stream Reader Stopped: chanKey={} streamKey={} lastSeenId={}", chanKey, streamKey, lastSeenId)
    }

매장 queue
  → broadcastNow()
  → emitter별 outbound queue
  → startEmitterSender()
  → sendToEmitter()
  → emitter.send()

    private fun broadcastNow(chanKey: String, msg: SseMsg) {
        emitters[chanKey]?.forEach { emitter ->
            enqueueOutbound(chanKey, emitter, msg)
        }
    }
    
    private fun startEmitterSender(chanKey: String, emitter: SseEmitter, connectionId: String) {
        val task = emitterSenderExec.submit {
            while (!Thread.currentThread().isInterrupted && outboundQueues.containsKey(emitter)) {
                val msg = try {
                    outboundQueues[emitter]?.poll(30, TimeUnit.SECONDS)
                } catch (e: InterruptedException) {
                    Thread.currentThread().interrupt()
                    null
                } ?: continue
 
                if (!sendToEmitter(emitter, msg.name, msg.data, msg.id)) {
                    unregisterEmitter(chanKey, emitter, "send-failed")
                    break
                }
            }
            logger.info("[SSE-SENDER-STOP] storeId={} connectionId={}", chanKey, connectionId)
        }
        emitterSenderTasks[emitter] = task
    }
 
    private fun sendToEmitter(emitter: SseEmitter, name: String, data: String, id: String): Boolean {
        try {
            emitter.send(
                SseEmitter.event()
                    .id(id)
                    .name(name)
                    .data(data)
                    .reconnectTime(2000)
            )
            return true
        } catch (e: Exception) {
            logger.debug("Failed to send SSE event: ${e.message}")
            emitter.completeWithError(e)
            return false
        }
    }

PosRoot
  → PosRealtimeBridge
  → ticket 발급
  → EventSource 생성
  → MessageEvent 수신
  → eventAdapter
  → Redux realtime action dispatch
  → listener middleware에서 entity patch
  → markStreamEventApplied
  → lastAppliedEventId 갱신

// PosRoot.tsx
export default function PosRoot() {
    const { storeId = "" } = useParams();
 
    return (
        <StoreProvider>
            <PosRealtimeBridge storeId={storeId} />
            <div className="contents font-sans">
                <Outlet />
            </div>
        </StoreProvider>
    );
}

// sseClient.ts
export function subscribeSse(params: {
    url: string;
    withCredentials?: boolean;
    handlers: Record<string, (raw: string, meta: SseEventMeta) => void>; // eventName -> handler(rawJson)
    onError?: (ev?: Event) => void;
    onOpen?: () => void;
}): SseSubscription {
    const es = new EventSource(params.url, { withCredentials: params.withCredentials ?? true });
 
    for (const [eventName, handler] of Object.entries(params.handlers)) {
        es.addEventListener(eventName, (ev) => {
            const message = ev as MessageEvent;
            handler(message.data, { lastEventId: message.lastEventId });
        });
    }
 
    es.onopen = () => params.onOpen?.();
    es.onerror = (ev) => params.onError?.(ev);
 
    return {
        close: () => es.close(),
    };
}

// PosRealtimeBridge.tsx
export function PosRealtimeBridge({ storeId }: Props) {
    const dispatch = useAppDispatch();
    const lastAppliedEventId = useAppSelector((state) => state.pos.stream.lastAppliedEventId);
    // ....

// PosRealtimeBridge.tsx
export function PosRealtimeBridge({ storeId }: Props) {
    const dispatch = useAppDispatch();
    const lastAppliedEventId = useAppSelector((state) => state.pos.stream.lastAppliedEventId);
    // ....
	// ...
	useEffect(() => {
	//. ...
		const reconcileAfterResume = () => {
		//. ...
			const connectionId = createRequestId();
			const url = getSseUrl(storeId, ticket, connectionId, lastAppliedEventIdRef.current);

// listener.ts
export const dispatchPosRealtimeEventWithId = (
    event: PosRealtimeEvent,
    streamEventId: string,
    dedupeEventId?: string,
): PosRealtimeAction & { meta: { streamEventId: string; dedupeEventId?: string } } =>
    ({
        type: `pos/realtime/${event.type}`,
        payload: event.payload,
        meta: { streamEventId, dedupeEventId },
    }) as PosRealtimeAction & { meta: { streamEventId: string; dedupeEventId?: string } };

// listener.ts
posListenerMiddleware.startListening({
	effect: async (action, api) => {
	// ...
		switch (action.type) {
		// ...
            case "pos/realtime/payment/resolved": {
                const payload = action.payload;
                const pos = getPosState(api);
                const orderExists = !!pos.entities.ordersById[payload.orderId];
 
                if (orderExists) {
                    api.dispatch(
                        posActions.patchOrderStatus({
                            orderId: payload.orderId,
                            status: toPosOrderStatus(payload.currentOrderStatus),
                        }),
                    );
                } else {
                    invalidatePayments(api);
                }
 
                api.dispatch(posActions.removePaymentQueue(payload.orderId));
                invalidateHall(api);
                invalidateKitchen(api);
                invalidateTables(api);
                break;
            }

//listener.ts
posListenerMiddleware.startListening({
	// .....
	effect: async (action, api) => {
	//......
        if (isReplayableStreamEvent) {
            api.dispatch(
                posActions.markStreamEventApplied({
                    eventId: streamEventId,
                    dedupeId,
                    snapshotScopes: getSnapshotScopesForRealtimeAction(action.type),
                }),
            );
        }

// slice.ts 
export const posSlice = createSlice({
    name: "pos",
    initialState: initialPosDomainState,
    reducers: {
        markStreamEventApplied(
            state,
            action: PayloadAction<{ eventId: string; dedupeId?: string; snapshotScopes: StreamSnapshotScope[] }>
        ) {
            const eventId = normalizeRedisStreamId(action.payload.eventId);
            rememberAppliedEventId(state, eventId);
            rememberAppliedDedupeId(state, action.payload.dedupeId);
 
            if (isRedisStreamIdAfter(eventId, state.stream.lastAppliedEventId)) {
                state.stream.lastAppliedEventId = eventId;
            }
 
            for (const scope of action.payload.snapshotScopes) {
                const prev = state.stream.lastAppliedEventIdBySnapshot[scope];
                if (isRedisStreamIdAfter(eventId, prev)) {
                    state.stream.lastAppliedEventIdBySnapshot[scope] = eventId;
                }
            }
        },

XREAD 결과 있음
  → 각 record 처리
  → lastSeenId 갱신
  → 다시 XREAD
 
XREAD 결과 없음
  → 최대 2초 blocking
  → 결과가 없으면 빈 목록
  → while loop를 돌며 다시 XREAD
 
새 event publish
  → blocking read 반환
  → record 처리

// SseNotificationService.kt 
    private fun readStreamLoop(chanKey: String, streamKey: String, initialLastSeenId: String) {
        var lastSeenId = initialLastSeenId
        val options = StreamReadOptions.empty()
            .count(100)
            .block(Duration.ofSeconds(2))
 
        while (!Thread.currentThread().isInterrupted && emitters.containsKey(chanKey)) {
            try {
                val records = redisTemplate.opsForStream<String, String>().read(
                    options,
                    StreamOffset.create(streamKey, ReadOffset.from(lastSeenId))
                ) ?: emptyList()
 
                for (record in records) {
                    lastSeenId = record.id.toString()
                    enqueueRecord(chanKey, streamKey, record)
                }
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
            } catch (e: Exception) {
                logger.warn("Redis stream read failed: chanKey={} streamKey={} lastSeenId={}", chanKey, streamKey, lastSeenId)
                enqueue(chanKey, SseMsg("RELOAD_REQUIRED", "{}", "0-0"))
                Thread.sleep(1000)
            }
        }
 
        logger.info("Redis Stream Reader Stopped: chanKey={} streamKey={} lastSeenId={}", chanKey, streamKey, lastSeenId)
    }
 

1. SSE 연결
2. 연결 성공 시 baseline 복구 실행
3. 최초 연결(`lastAppliedEventId = 0-0`)이면 full baseline 실행
4. 각 snapshot 응답의 watermark를 scope별로 저장
5. 모든 snapshot 중 가장 작은 watermark까지 stream cursor를 전진
6. SSE 이벤트를 Redux 상태에 patch

//SseNotificationService.kt
val startId = currentStreamTailId(streamKey)
val task = streamReaderExec.submit {
    readStreamLoop(chanKey, streamKey, startId)
}

//SseNotificationService.kt
private fun currentStreamTailId(streamKey: String): String =
    try {
        val info = redisTemplate.opsForStream<String, String>().info(streamKey)
        info.lastGeneratedId() ?: info.lastEntryId() ?: "0-0"
    } catch (_: Exception) {
        "0-0"
    }

// SseNotificationService.kt -> readStreamLoop
StreamOffset.create(streamKey, ReadOffset.from(lastSeenId))

// PosRealtimeBridge.tsx
const sub = subscribeSse({
    url,
    handlers,
    onOpen: () => {
        touchSseActivity();
        reconnectDelayMsRef.current = INITIAL_RECONNECT_DELAY_MS;
        startWatchdog();
        reloadBaseline();
    },
});

EventSource 연결 성공
→ baseline 복구 시작

initialState.ts
stream: {
    lastAppliedEventId: "0-0",
}

//reloadChangedPosBaseline.ts
if (!sinceEventId || sinceEventId === "0-0") {
    await dispatch(reloadPosBaseline(storeId)).unwrap();
    return { storeId, mode: "full" as const };
}

SSE open
→ reloadChangedPosBaseline(sinceEventId = "0-0")
→ reloadPosBaseline()
→ full snapshot load

//reloadPosBaseline.ts
await Promise.all([
    dispatch(loadPaymentQueue({ storeId, force: true })).unwrap(),
    dispatch(loadServingQueue({ storeId, force: true })).unwrap(),
    dispatch(loadCallQueue({ storeId, force: true })).unwrap(),
    dispatch(loadHallTableZones(storeId)).unwrap(),
    dispatch(loadHallTables({ storeId, force: true })).unwrap(),
    dispatch(loadMenuCatalog(storeId)).unwrap(),
    dispatch(loadHallOrders(...)).unwrap(),
    dispatch(loadKitchenOrders(...)).unwrap(),
]);

// PosStreamWatermarkService.kt
fun currentWatermark(storeId: UUID): String {
    val key = PosStreamKey.store(storeId)
    val info = redisTemplate.opsForStream<String, String>().info(key)
    return info.lastGeneratedId() ?: info.lastEntryId() ?: "0-0"
}

ResponseEntity.ok()
    .header(POS_STREAM_WATERMARK_HEADER, watermark)
    .body(body)

dispatch(posActions.markStreamSnapshotLoaded({
    eventId: snapshot.streamWatermark,
    snapshotScopes: ["payments"],
}));

// slice.ts
for (const scope of action.payload.snapshotScopes) {
    const prev = state.stream.lastAppliedEventIdBySnapshot[scope];
 
    if (isRedisStreamIdAfter(eventId, prev)) {
        state.stream.lastAppliedEventIdBySnapshot[scope] = eventId;
    }
}

//reloadPosBaseline.ts
const coveredWatermark = BASELINE_SNAPSHOT_SCOPES
    .map((scope) => stream.lastAppliedEventIdBySnapshot[scope])
    .reduce((min, eventId) =>
        compareRedisStreamIds(eventId, min) < 0 ? eventId : min
    );
 
dispatch(posActions.markStreamWatermarkCovered({
    eventId: coveredWatermark,
}));

// reducer.ts
if (isRedisStreamIdAfter(eventId, state.stream.lastAppliedEventId)) {
    state.stream.lastAppliedEventId = eventId;
}

PosRealtimeBridge.tsx
const event = adaptRealtimeEvent(eventName, parsed);
const reduxAction = dispatchPosRealtimeEventWithId(
    event,
    meta.lastEventId,
    dedupeEventId,
);
 
dispatch(reduxAction);

private fun sendToEmitter(emitter: SseEmitter, name: String, data: String, id: String): Boolean {
    emitter.send(
        SseEmitter.event()
            .id(id)
            .name(name)
            .data(data)
            .reconnectTime(2000)
    )
    return true
}

es.addEventListener(eventName, (ev) => {
    const message = ev as MessageEvent;
    handler(message.data, { lastEventId: message.lastEventId });
});

const url = getSseUrl(storeId, ticket, connectionId, lastAppliedEventIdRef.current);

SSE replay 시도
→ replay 불신 조건이면 RELOAD_REQUIRED
→ 클라이언트 reloadChangedPosBaseline 실행
→ 유효한 lastAppliedEventId가 있으면 /changes 호출
→ delta 계산 가능 + entity 100개 이하이면 partial reload
→ 아니면 full baseline reload
→ watermark까지 cursor 전진
→ SSE event 적용 재개

// SseNotificationService.kt
if (!lastEventId.isNullOrBlank()) {
    replayMissingEvents(sKey, lastEventId, emitter)
    markActive(chanKey)
}

//SseNotificationService.kt
private fun replayMissingEvents(
    streamKey: String,
    lastEventId: String,
    emitter: SseEmitter,
) {
    try {
        if (isReplayGapExpired(streamKey, lastEventId)) {
            sendReloadSignal(emitter)
            return
        }
 
        val range = Range.rightOpen(lastEventId, "+")
        val events = redisTemplate.opsForStream<String, String>()
            .range(streamKey, range)
            ?: return
 
        var count = 0
        for (record in events) {
            if (count++ >= 300) {
                enqueueReplay(
                    emitter,
                    SseMsg("RELOAD_REQUIRED", "{}", "0-0"),
                )
                return
            }
 
            val data = record.value
            val type = data["type"] ?: "UNKNOWN"
 
            if (type !in knownEventNames) {
                enqueueReplay(
                    emitter,
                    SseMsg(
                        "RELOAD_REQUIRED",
                        "{}",
                        record.id.toString(),
                    ),
                )
                return
            }
 
            enqueueReplay(/* event */)
        }
    } catch (e: Exception) {
        sendReloadSignal(emitter)
    }
}

// SseNotificationService.kt
private fun isReplayGapExpired(
    streamKey: String,
    lastEventId: String,
): Boolean {
    val lastId = parseStreamId(lastEventId) ?: return true
 
    val info = try {
        redisTemplate.opsForStream<String, String>().info(streamKey)
    } catch (_: Exception) {
        return false
    }
 
    val firstId = info.firstEntryId() ?: return false
 
    return compareStreamIds(
        parseStreamId(firstId) ?: return false,
        lastId,
    ) > 0
}

// SseNotificationService.kt
private val knownEventNames = setOf(
    "RELOAD_REQUIRED",
    "HALL_PAYMENT",
    "HALL_PAYMENT_RESOLVED",
    // ...
    "KITCHEN_MENU_CHANGED",
)

//sseClient.ts
export function safeJson<T>(raw: string): T | null {
    try {
        return JSON.parse(raw) as T;
    } catch {
        return null;
    }
}

// PosRealtimeBridge.tsx 
const parsed = safeJson<unknown>(raw);
 
if (!parsed) {
    console.error("[SSE PARSE ERROR]", {
        storeId,
        eventName,
        raw,
    });
 
    reloadBaseline();
    return;
}

const event = adaptRealtimeEvent(eventName, parsed);
 
if (!event) {
    reloadBaseline();
    return;
}

// 매장 broadcast queue
ArrayBlockingQueue<SseMsg>(2000)
 
// emitter별 replay queue
ArrayBlockingQueue<SseMsg>(500)
 
// emitter별 outbound queue
ArrayBlockingQueue<SseMsg>(500)

if (q.offer(msg)) {
    markActive(chanKey)
    return
}
 
logger.warn("queue overflow: chanKey=$chanKey")
q.clear()
q.offer(SseMsg("RELOAD_REQUIRED", "{}", "0-0"))
markActive(chanKey)

if (!q.offer(msg)) {
    q.clear()
    q.offer(SseMsg("RELOAD_REQUIRED", "{}", "0-0"))
}

logger.warn(
    "[SSE-SLOW-CONSUMER] outbound queue overflow"
)
 
q.clear()
q.offer(SseMsg("RELOAD_REQUIRED", "{}", "0-0"))
emitter.complete()
unregisterEmitter(chanKey, emitter, "slow-consumer")

// posStreamChangesApi.ts 
 
const sp = new URLSearchParams();
sp.set("since", sinceEventId);
sp.set("limit", "500");
 
const res = await apiRequest(
    `/api/pos/${storeId}/notifications/changes?${sp.toString()}`,
);

// PosNotificationController.kt 
@GetMapping("/changes")
fun getChangesSince(
    @RequestParam since: String?,
    @RequestParam(
        required = false,
        defaultValue = "500",
    ) limit: Int,
) = ResponseEntity.ok(
    posStreamChangeService.collectChanges(
        storeId = storeId,
        sinceEventId = since,
        maxEvents = limit.coerceIn(1, 1000),
    ),
)

// PosStreamChangeService.kt
val records = redisTemplate.opsForStream<String, String>()
    .range(
        streamKey,
        Range.rightOpen(normalizedSince, "+"),
    )
    ?: emptyList()

val orderIds = linkedSetOf<UUID>()
val tableIds = linkedSetOf<UUID>()
val menuIds = linkedSetOf<UUID>()
val callRequestIds = linkedSetOf<UUID>()

when (type) {
    "HALL_PAYMENT",
    "HALL_SERVE",
    "KITCHEN_COOK" -> {
        payload.uuid("orderId")?.let(orderIds::add)
        payload.uuid("tableId")?.let(tableIds::add)
        payload.uuid("menuId")?.let(menuIds::add)
    }
 
    "CUSTOMER_CALL" -> {
        payload.uuid("requestId")?.let(callRequestIds::add)
        payload.uuid("tableId")?.let(tableIds::add)
    }
 
    "KITCHEN_MENU_CHANGED" -> {
        payload.uuid("menuId")?.let(menuIds::add)
    }
}

if (records.size > maxEvents) {
    return reloadRequired(
        normalizedSince,
        currentWatermark,
        "TOO_MANY_EVENTS",
    )
}

val payload = parsePayload(record.value["payload"])
    ?: return reloadRequired(
        normalizedSince,
        currentWatermark,
        "INVALID_EVENT_PAYLOAD",
    )

// reloadChangedPosBaseline.ts
if (changes.requiresReload) {
    await dispatch(reloadPosBaseline(storeId)).unwrap();
 
    return {
        storeId,
        mode: "full",
        reason: changes.reason,
    };
}

const PARTIAL_RELOAD_ENTITY_LIMIT = 100;
 
if (
    changes.changedOrderIds.length > PARTIAL_RELOAD_ENTITY_LIMIT ||
    changes.changedTableIds.length > PARTIAL_RELOAD_ENTITY_LIMIT ||
    changes.changedMenuIds.length > PARTIAL_RELOAD_ENTITY_LIMIT ||
    changes.changedCallRequestIds.length > PARTIAL_RELOAD_ENTITY_LIMIT
) {
    await dispatch(reloadPosBaseline(storeId)).unwrap();
 
    return {
        storeId,
        mode: "full",
        reason: "TOO_MANY_CHANGED_ENTITIES",
    };
}

// reloadChangedPosBaseline.ts
await Promise.all([
    orderIds.length > 0
        ? fetchHallOrdersByIds(storeId, orderIds)
        : Promise.resolve(),
 
    orderIds.length > 0
        ? fetchKitchenOrdersByIds(storeId, orderIds)
        : Promise.resolve(),
 
    orderIds.length > 0
        ? dispatch(loadPaymentQueue({ storeId, force: true })).unwrap()
        : Promise.resolve(),
 
    orderIds.length > 0
        ? dispatch(loadServingQueue({ storeId, force: true })).unwrap()
        : Promise.resolve(),
 
    changes.changedCallRequestIds.length > 0
        ? dispatch(loadCallQueue({ storeId, force: true })).unwrap()
        : Promise.resolve(),
 
    changes.changedMenuIds.length > 0
        ? dispatch(loadMenuCatalog(storeId)).unwrap()
        : Promise.resolve(),
 
    tableIds.length > 0
        ? reloadChangedTables(storeId, tableIds, dispatch, getState)
        : Promise.resolve(),
]);

dispatch(
    posActions.markStreamWatermarkCovered({
        eventId: changes.currentWatermark,
    }),
);

//reloadChangedPosBaseline.ts
if (!sinceEventId || sinceEventId === "0-0") {
    await dispatch(reloadPosBaseline(storeId)).unwrap();
}

// reloadPosBaseline.ts
await Promise.all([
    dispatch(loadPaymentQueue({ storeId, force: true })).unwrap(),
    dispatch(loadServingQueue({ storeId, force: true })).unwrap(),
    dispatch(loadCallQueue({ storeId, force: true })).unwrap(),
    dispatch(loadHallTableZones(storeId)).unwrap(),
    dispatch(loadHallTables({ storeId, force: true })).unwrap(),
    dispatch(loadMenuCatalog(storeId)).unwrap(),
    dispatch(loadHallOrders(/* ... */)).unwrap(),
    dispatch(loadKitchenOrders(/* ... */)).unwrap(),
]);

const coveredWatermark = BASELINE_SNAPSHOT_SCOPES
    .map(
        (scope) =>
            stream.lastAppliedEventIdBySnapshot[scope],
    )
    .reduce(
        (min, eventId) =>
            compareRedisStreamIds(eventId, min) < 0
                ? eventId
                : min,
    );
 
dispatch(
    posActions.markStreamWatermarkCovered({
        eventId: coveredWatermark,
    }),
);

payments
servings
calls
hallOrders
kitchenOrders
tables

if (!parsed) {
    reloadBaseline();
    return;
}
 
if (!event) {
    reloadBaseline();
    return;
}
 
if (eventName === "RELOAD_REQUIRED") {
    reloadBaseline();
}

// PosRealtimeBridge.tsx
 
cleanupSubscription();
scheduleReconnect();

onOpen: () => {
    startWatchdog();
    reloadBaseline();
}

Event traffic 증가
→ Redis Stream batch read
→ bounded queue에 임시 저장
→ 16개 shard worker가 매장별 queue 처리
→ emitter별 outbound queue로 전달
→ 한계 초과 시 RELOAD_REQUIRED 또는 연결 종료
→ 클라이언트 partial/full baseline 복구

val options = StreamReadOptions.empty()
    .count(100)
    .block(Duration.ofSeconds(2))

val records = redisTemplate.opsForStream<String, String>().read(
    options,
    StreamOffset.create(streamKey, ReadOffset.from(lastSeenId)),
) ?: emptyList()
 
for (record in records) {
    lastSeenId = record.id.toString()
    enqueueRecord(chanKey, streamKey, record)
}

//PosOutboxRelay.kt
private val streamMaxLen = 10_000L
private val batchSize = 100
 
@Scheduled(fixedDelay = 500)
@Transactional
fun publishPendingEvents() {
    val events = outboxRepository.lockPublishable(batchSize)
}

// 매장별 broadcast queue
private fun queueOf(chanKey: String) =
    queues.computeIfAbsent(chanKey) {
        ArrayBlockingQueue(2000)
    }
 
// SSE 재연결 replay용 emitter별 queue
private fun replayQueueOf(emitter: SseEmitter) =
    replayQueues.computeIfAbsent(emitter) {
        ArrayBlockingQueue(500)
    }
 
// 실제 브라우저 송신용 emitter별 queue
private fun outboundQueueOf(emitter: SseEmitter) =
    outboundQueues.computeIfAbsent(emitter) {
        ArrayBlockingQueue(500)
    }

private val senderWorkers = 16
private val senderExec = Executors.newFixedThreadPool(senderWorkers)
 
private fun shard(chanKey: String) =
    (chanKey.hashCode() and Int.MAX_VALUE) % senderWorkers
 
private val activeKeys =
    Array(senderWorkers) {
        ConcurrentLinkedQueue<String>()
    }

private fun markActive(chanKey: String) {
    if (activeFlag.putIfAbsent(chanKey, true) == null) {
        activeKeys[shard(chanKey)].offer(chanKey)
    }
}

private fun startSenders() {
    repeat(senderWorkers) { idx ->
        senderExec.submit {
            val myKeys = mutableListOf<String>()
 
            while (!Thread.currentThread().isInterrupted) {
                while (true) {
                    val key = activeKeys[idx].poll() ?: break
                    myKeys.add(key)
                }
 
                // 매장별 queue 처리
            }
        }
    }
}

var processed = 0
 
while (processed < 50) {
    val msg = q.poll() ?: break
    broadcastNow(chanKey, msg)
    processed++
}

drainReplay(
    chanKey,
    limitPerEmitter = 20,
)

private fun broadcastNow(
    chanKey: String,
    msg: SseMsg,
) {
    emitters[chanKey]?.forEach { emitter ->
        enqueueOutbound(chanKey, emitter, msg)
    }
}

while (
    !Thread.currentThread().isInterrupted &&
    outboundQueues.containsKey(emitter)
) {
    val msg = outboundQueues[emitter]
        ?.poll(30, TimeUnit.SECONDS)
        ?: continue
 
    if (!sendToEmitter(
            emitter,
            msg.name,
            msg.data,
            msg.id,
        )
    ) {
        unregisterEmitter(
            chanKey,
            emitter,
            "send-failed",
        )
        break
    }
}

private fun enqueue(
    chanKey: String,
    msg: SseMsg,
) {
    val q = queueOf(chanKey)
 
    if (q.offer(msg)) {
        markActive(chanKey)
        return
    }
 
    logger.warn("queue overflow: chanKey=$chanKey")
 
    q.clear()
    q.offer(
        SseMsg(
            "RELOAD_REQUIRED",
            "{}",
            "0-0",
        ),
    )
    markActive(chanKey)
}

중간 event를 모두 전달할 수 없음
→ event cursor 기반 incremental apply를 더 이상 신뢰하지 않음
→ snapshot 기반 복구 요청

private fun enqueueReplay(
    emitter: SseEmitter,
    msg: SseMsg,
) {
    val q = replayQueueOf(emitter)
 
    if (!q.offer(msg)) {
        q.clear()
        q.offer(
            SseMsg(
                "RELOAD_REQUIRED",
                "{}",
                "0-0",
            ),
        )
    }
}

if (q.offer(msg)) return
 
logger.warn(
    "[SSE-SLOW-CONSUMER] " +
        "storeId={} connectionId={} " +
        "outbound queue overflow",
    chanKey,
    connectionIds[emitter],
)
 
q.clear()
q.offer(
    SseMsg(
        "RELOAD_REQUIRED",
        "{}",
        "0-0",
    ),
)
 
emitter.complete()
unregisterEmitter(
    chanKey,
    emitter,
    "slow-consumer",
)

dispatch(reduxAction);
 
if (eventName === "RELOAD_REQUIRED") {
    reloadBaseline();
}

await dispatch(
    reloadChangedPosBaseline({
        storeId,
        sinceEventId:
            lastAppliedEventIdRef.current,
    }),
).unwrap();

lastAppliedEventId
→ /notifications/changes
→ 변경된 order/table/menu/call ID 계산
→ 해당 entity만 재조회

if (changes.requiresReload) {
    await dispatch(
        reloadPosBaseline(storeId),
    ).unwrap();
}

if (
    changes.changedOrderIds.length > 100 ||
    changes.changedTableIds.length > 100 ||
    changes.changedMenuIds.length > 100 ||
    changes.changedCallRequestIds.length > 100
) {
    await dispatch(
        reloadPosBaseline(storeId),
    ).unwrap();
}

특정 client 송신 지연
→ emitter별 outbound queue에 이벤트 적재
→ queue limit 500 안쪽이면 계속 송신
→ queue가 가득 차면 해당 emitter만 종료
→ FE EventSource error 감지
→ backoff + jitter 재연결
→ 연결 성공 후 partial/full baseline 복구

// SseNotificationService.kt
private val replayQueues =
    ConcurrentHashMap<SseEmitter, ArrayBlockingQueue<SseMsg>>()
 
private fun replayQueueOf(emitter: SseEmitter) =
    replayQueues.computeIfAbsent(emitter) {
        ArrayBlockingQueue(500)
    }
 
private val outboundQueues =
    ConcurrentHashMap<SseEmitter, ArrayBlockingQueue<SseMsg>>()
 
private fun outboundQueueOf(emitter: SseEmitter) =
    outboundQueues.computeIfAbsent(emitter) {
        ArrayBlockingQueue(500)
    }

private fun broadcastNow(
    chanKey: String,
    msg: SseMsg,
) {
    emitters[chanKey]?.forEach { emitter ->
        enqueueOutbound(chanKey, emitter, msg)
    }
}

private fun registerEmitter(
    chanKey: String,
    emitter: SseEmitter,
    streamKey: String,
    connectionId: String,
) {
    connectionIds[emitter] = connectionId
    outboundQueueOf(emitter)
    startEmitterSender(chanKey, emitter, connectionId)
 
    // ...
}

private fun startEmitterSender(
    chanKey: String,
    emitter: SseEmitter,
    connectionId: String,
) {
    val task = emitterSenderExec.submit {
        while (
            !Thread.currentThread().isInterrupted &&
            outboundQueues.containsKey(emitter)
        ) {
            val msg = try {
                outboundQueues[emitter]
                    ?.poll(30, TimeUnit.SECONDS)
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                null
            } ?: continue
 
            if (!sendToEmitter(
                    emitter,
                    msg.name,
                    msg.data,
                    msg.id,
                )
            ) {
                unregisterEmitter(
                    chanKey,
                    emitter,
                    "send-failed",
                )
                break
            }
        }
    }
 
    emitterSenderTasks[emitter] = task
}

private fun enqueueOutbound(
    chanKey: String,
    emitter: SseEmitter,
    msg: SseMsg,
) {
    val q = outboundQueues[emitter] ?: return
 
    if (q.offer(msg)) return
 
    // overflow 처리
}

private fun enqueueOutbound(
    chanKey: String,
    emitter: SseEmitter,
    msg: SseMsg,
) {
    val q = outboundQueues[emitter] ?: return
    if (q.offer(msg)) return
 
    logger.warn(
        "[SSE-SLOW-CONSUMER] " +
            "storeId={} connectionId={} " +
            "outbound queue overflow",
        chanKey,
        connectionIds[emitter],
    )
 
    q.clear()
    q.offer(
        SseMsg(
            "RELOAD_REQUIRED",
            "{}",
            "0-0",
        ),
    )
 
    emitter.complete()
    unregisterEmitter(
        chanKey,
        emitter,
        "slow-consumer",
    )
}

outbound queue overflow
→ 기존 적체 이벤트 제거
→ RELOAD_REQUIRED 삽입
→ emitter 종료
→ 해당 client 연결 제거

private fun unregisterEmitter(
    chanKey: String,
    emitter: SseEmitter,
    reason: String,
) {
    val connectionId = connectionIds.remove(emitter)
 
    replayQueues.remove(emitter)?.clear()
    outboundQueues.remove(emitter)?.clear()
    emitterSenderTasks.remove(emitter)?.cancel(true)
 
    emitters.computeIfPresent(chanKey) { _, list ->
        list.remove(emitter)
 
        if (list.isEmpty()) {
            cancelStreamReader(chanKey)
            queues.remove(chanKey)?.clear()
            activeFlag.remove(chanKey)
            null
        } else {
            list
        }
    }
}

const sub = subscribeSse({
    url,
    withCredentials: true,
    handlers,
 
    onError: (error) => {
        console.error("[SSE ERROR]", {
            storeId,
            connectionId,
            error,
            mySeq,
        });
 
        if (
            closedRef.current ||
            connectSeqRef.current !== mySeq
        ) {
            return;
        }
 
        if (subRef.current === sub) {
            cleanupSubscription();
        }
 
        scheduleReconnect();
    },
});

const INITIAL_RECONNECT_DELAY_MS = 2_000;
const MAX_RECONNECT_DELAY_MS = 30_000;

const scheduleReconnect = () => {
    if (closedRef.current) return;
    if (connectingRef.current) return;
    if (reconnectTimerRef.current != null) return;
 
    const baseDelayMs =
        reconnectDelayMsRef.current;
 
    const jitterMs = Math.floor(
        Math.random() *
        Math.min(2_000, baseDelayMs * 0.3),
    );
 
    const delayMs = baseDelayMs + jitterMs;
 
    reconnectDelayMsRef.current = Math.min(
        baseDelayMs * 2,
        MAX_RECONNECT_DELAY_MS,
    );
 
    reconnectTimerRef.current =
        window.setTimeout(() => {
            reconnectTimerRef.current = null;
            void connect();
        }, delayMs);
};

onOpen: () => {
    touchSseActivity();
 
    reconnectDelayMsRef.current =
        INITIAL_RECONNECT_DELAY_MS;
 
    startWatchdog();
    reloadBaseline();
},

const reloadBaseline = () => {
    if (baselineReloadRef.current) return;
 
    const p = (async () => {
        await dispatch(
            reloadChangedPosBaseline({
                storeId,
                sinceEventId:
                    lastAppliedEventIdRef.current,
            }),
        ).unwrap();
    })();
 
    baselineReloadRef.current = p;
};

if (!sinceEventId || sinceEventId === "0-0") {
    await dispatch(
        reloadPosBaseline(storeId),
    ).unwrap();
 
    return {
        storeId,
        mode: "full" as const,
    };
}

await Promise.all([
    orderIds.length > 0
        ? fetchHallOrdersByIds(storeId, orderIds)
        : Promise.resolve(),
 
    orderIds.length > 0
        ? fetchKitchenOrdersByIds(storeId, orderIds)
        : Promise.resolve(),
 
    changes.changedCallRequestIds.length > 0
        ? dispatch(
            loadCallQueue({
                storeId,
                force: true,
            }),
        ).unwrap()
        : Promise.resolve(),
 
    changes.changedMenuIds.length > 0
        ? dispatch(loadMenuCatalog(storeId)).unwrap()
        : Promise.resolve(),
]);

if (changes.requiresReload) {
    await dispatch(
        reloadPosBaseline(storeId),
    ).unwrap();
 
    return {
        storeId,
        mode: "full" as const,
        reason: changes.reason,
    };
}

Outbox pending
→ Relay가 조회
→ Redis Stream publish
→ stream ID 저장
→ published 처리

Redis publish 실패
→ attempts 증가
→ nextAttemptAt 설정
→ 이후 relay에서 재조회

@Entity
@Table(name = "pos_event_outbox")
class PosEventOutbox(
    // 생략 ...
 
    @Column(name = "published_at")
    var publishedAt: Instant? = null,
 
    @Column(name = "redis_stream_id", length = 80)
    var redisStreamId: String? = null,
 
    @Column(name = "attempts", nullable = false)
    var attempts: Int = 0,
 
    @Column(name = "next_attempt_at", nullable = false)
    var nextAttemptAt: Instant = Instant.now(),
 
    @Column(name = "last_error", columnDefinition = "TEXT")
    var lastError: String? = null,
) {
    fun markPublished(streamId: String, now: Instant = Instant.now()) {
        redisStreamId = streamId
        publishedAt = now
        lastError = null
    }
 
    fun markFailed(error: Throwable) {
        attempts += 1
        lastError = (error.message ?: error::class.java.simpleName).take(1000)
        val delaySeconds = minOf(60L, 1L shl attempts.coerceAtMost(6))
        nextAttemptAt = Instant.now().plus(Duration.ofSeconds(delaySeconds))
    }
}

@Component
class PosOutboxRelay(...) {
	@Scheduled(fixedDelay = 500)
    @Transactional
    fun publishPendingEvents() {
        val events = outboxRepository.lockPublishable(batchSize)
        if (events.isEmpty()) return
 
        events.forEach { event ->
            try {
                val streamId = publishToRedis(event)
                event.markPublished(streamId)
            } catch (e: Exception) {
                event.markFailed(e)
                // log ...
            }
        }
    }
}
 
interface PosEventOutboxRepository : JpaRepository<PosEventOutbox, UUID> {
    @Query(
        value = """
            select *
            from pos_event_outbox
            where published_at is null
              and next_attempt_at <= now()
            order by created_at
            limit :limit
            for update skip locked
        """,
        nativeQuery = true,
    )
    fun lockPublishable(@Param("limit") limit: Int): List<PosEventOutbox>
}
 

Redis XADD 성공
→ 애플리케이션 종료 또는 DB commit 실패
→ Outbox에는 published_at이 여전히 null
→ 재시작 후 같은 Outbox 이벤트를 다시 XADD

    private fun enqueueRecord(chanKey: String, streamKey: String, message: MapRecord<String, String, String>) {
        try {
            val type = message.value["type"] ?: "UNKNOWN"
            val payload = message.value["payload"] ?: "{}"
            val id = message.id.toString()
            if (type !in knownEventNames) {
                logger.warn("Unknown SSE event type: type={} id={} streamKey={}", type, id, streamKey)
                enqueue(chanKey, SseMsg("RELOAD_REQUIRED", "{}", id))
                return
            }
 
            enqueue(chanKey, SseMsg(type, payloadWithOutboxEventId(payload, message.value["eventId"]), id))
        } catch (e: Exception) {
            logger.error("Error processing redis message", e)
        }
    }
 
    private fun payloadWithOutboxEventId(payload: String, outboxEventId: String?): String {
        if (outboxEventId.isNullOrBlank()) return payload
 
        return try {
            val node = objectMapper.readTree(payload)
            if (!node.isObject) return payload
 
            (node as com.fasterxml.jackson.databind.node.ObjectNode)
                .put("__outboxEventId", outboxEventId)
            objectMapper.writeValueAsString(node)
        } catch (_: Exception) {
            payload
        }
    }

function readOutboxDedupeEventId(value: unknown): string | undefined {
    if (!value || typeof value !== "object") return undefined;
    const outboxEventId = (value as { __outboxEventId?: unknown }).__outboxEventId;
    return typeof outboxEventId === "string" && outboxEventId.trim()
        ? `outbox:${outboxEventId}`
        : undefined;
}

posListenerMiddleware.startListening({
    predicate: (action): action is PosRealtimeAction =>
        typeof action.type === "string" && action.type.startsWith("pos/realtime/"),
 
    effect: async (action, api) => {
        const meta = (action as { meta?: { streamEventId?: string; dedupeEventId?: string } }).meta;
        const streamEventId = normalizeRedisStreamId(meta?.streamEventId);
        const dedupeEventId = meta?.dedupeEventId;
        const dedupeId = dedupeEventId ?? (streamEventId !== "0-0" ? `stream:${streamEventId}` : undefined);
        const isReplayableStreamEvent = streamEventId !== "0-0";
 
        if (dedupeId) {
            const alreadyApplied = getPosState(api).stream.recentAppliedDedupeIds.includes(dedupeId);
            if (alreadyApplied) {
                if (isReplayableStreamEvent) {
                    api.dispatch(
                        posActions.markStreamEventApplied({
                            eventId: streamEventId,
                            dedupeId,
                            snapshotScopes: getSnapshotScopesForRealtimeAction(action.type),
                        }),
                    );
                }
                return;
            }
        }
 
        switch (action.type) {
            case "pos/realtime/payment/open": {
        // ... 이하 생략 ...

fun createTicket(storeId: UUID): String {
    val ticket = UUID.randomUUID().toString().replace("-", "")
    redisTemplate.opsForValue().set("ticket:$ticket", storeId.toString(), Duration.ofSeconds(30))
    return ticket
}
 
fun subscribe(
    expectedStoreId: UUID,
    ticket: String,
    connectionId: String?,
    lastEventId: String?
): SseEmitter {
    val storeIdStr = redisTemplate.opsForValue().getAndDelete("ticket:$ticket")
        ?: throw BusinessException(ErrorCode.INVALID_TICKET)
 
    if (storeIdStr != expectedStoreId.toString()) {
        throw BusinessException(ErrorCode.INVALID_TICKET)
    }
 
    val emitter = SseEmitter(60 * 60 * 1000L)
    ...
}

async function getPosTicket(storeId: string): Promise<string> {
    const res = await apiRequest(`/api/pos/${storeId}/notifications/ticket`, {
        method: "GET",
        credentials: "include",
    });
    ...
}
 
function getSseUrl(
    storeId: string,
    ticket: string,
    connectionId: string,
    lastEventId: string,
): string {
    const params = new URLSearchParams({ ticket, connectionId });
    if (lastEventId !== "0-0") params.set("lastEventId", lastEventId);
    return `/api/pos/${storeId}/notifications/subscribe?${params.toString()}`;
}