Lets go show

@GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public ResponseEntity<SseEmitter> subscribe(
        @CurrentMember Member member,
        @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId) {
    return ResponseEntity.ok(sseEmitterService.subscribe(member, lastEventId));
}

public SseEmitter subscribe(Member member, String lastEventId) {
    SseEmitter emitter = new SseEmitter(DEFAULT_TIMEOUT);
    String uniqueEmitterId = generateUniqueClientId(member.getEmail(),LocalDateTime.now());
    sseEmitterRepository.save(uniqueEmitterId, emitter);

    emitter.onCompletion(() -> sseEmitterRepository.deleteById(uniqueEmitterId));
    emitter.onTimeout(() -> sseEmitterRepository.deleteById(uniqueEmitterId));
    emitter.onError((e) -> sseEmitterRepository.deleteById(uniqueEmitterId));

    sendToClient(emitter,uniqueEmitterId,"init","init");

    if(lastEventId != null){
        LocalDateTime dateTime = extractDateTime(lastEventId);
        List<Notification> notifications = notificationService.findAllAfterDateTime(dateTime, member);
        notifications.forEach(
                notification -> sendToClient(emitter,
                        uniqueEmitterId,
                        "notification",
                        NotificationResponse.from(notification))
        );
    }

    return emitter;
}

public void sendToClient(SseEmitter emitter, String id, String name, Object data) {
    try {
        emitter.send(
                SseEmitter.event()
                        .id(id)
                        .name(name)
                        .data(data)
        );
    } catch (IOException e) {
        log.error(e.getMessage(), e);
        sseEmitterRepository.deleteById(id);
        throw new RuntimeException(FAILED_SSE_NOTIFICATION_SEND.getMessage());
    }
}

@Repository
public class SseEmitterRepository {
    private Map<String, SseEmitter> emitterMap = new ConcurrentHashMap<>();

    public List<SseEmitter> findAllStartWithByMemberEmail(String email) {
        return emitterMap.entrySet().stream()
                .filter(entry-> entry.getKey().startsWith(email))
                .map(Entry::getValue)
                .toList();
    }

    public SseEmitter save(String id, SseEmitter emitter) {
        return emitterMap.put(id, emitter);
    }

    public void deleteById(String uniqueEmitterId) {
        emitterMap.remove(uniqueEmitterId);
    }

    public void deleteAll(){
        emitterMap.clear();
    }
}

public class NotificationEvent {
    String clientEmail;
    Long targetEntityId;
    Long relatedEntityId;
    String message;
    NotificationType notificationType;
}

@Getter
public enum NotificationMessage {
    REPLY_ADD("회원님의 게시글에 댓글이 달렸어요 :) ") ,
    NESTED_REPLY_ADD("회원님의 댓글에 댓글이 달렸어요 :)"),
    TODO_UPDATE("잊지 않으셨죠 ? 투두 리스트 제출 마감 시간은 오후 11시 입니다 :)"),
    ;
    private final String message;

    NotificationMessage(String message) {
        this.message = message;
    }
}

@Getter
public enum NotificationType {
    REPLY_ADD("댓글 추가", "/board",true),
    NESTED_REPLY_ADD("대댓글 추가", "/board", true),
    TODO_UPDATE("TODO 업데이트", null, true),
    ACHIEVEMENT_FEEDBACK("성취도", null, true),
    ;

    private final String value;
    private final String baseTargetUrl;
    private final Boolean isNeededSave;

    NotificationType(String value, String baseTargetUrl,Boolean isNeededSave) {
        this.value = value;
        this.baseTargetUrl = baseTargetUrl;
        this.isNeededSave = isNeededSave;
    }
}

@Async
public void publishReplyAddEvent(Board board, Reply reply) {
    Member boardWriter = board.getMember();
    Member replyWriter = reply.getMember();

    if(reply.getParent() != null){
        eventPublisher.publishEvent(new NotificationEvent(
                reply.getParent().getMember().getEmail(),
                board.getId(),
                reply.getId(),
                NotificationMessage.NESTED_REPLY_ADD.getMessage(),
                NotificationType.NESTED_REPLY_ADD));
    }

    if (!boardWriter.getEmail().equals(replyWriter.getEmail())) {
        eventPublisher.publishEvent(new NotificationEvent(
                boardWriter.getEmail(),
                board.getId(),
                reply.getId(),
                NotificationMessage.REPLY_ADD.getMessage(),
                NotificationType.REPLY_ADD));
    }

@Async
@EventListener
public void publishNotificationEvent(NotificationEvent event){
    try {
        String channelEvent = objectMapper.writeValueAsString(event);
        redisOperations.convertAndSend(notificationTopic.getTopic(), channelEvent);
    } catch (JsonProcessingException e) {
        log.error(e.getMessage(), e);
        throw new InvalidRedisMessageException(INVALID_REDIS_MESSAGE);
    }
}

public void handleNotificationEvent(String message) {
    NotificationEvent event = convertToObject(message, NotificationEvent.class);

    Member member = memberService.findByEmail(event.getClientEmail());

    Notification notification = Notification.from(member, event);
    if (notification.getNotificationType().getIsNeededSave()) {
        notificationService.save(notification);
    }

    List<SseEmitter> emitters = sseEmitterService.findAllStartWithByMemberEmail(member.getEmail());
    if (!emitters.isEmpty()) {
        String response = convertToJson(NotificationResponse.from(notification));
        emitters.forEach(emitter -> sseEmitterService.sendToClient(
                emitter,
                sseEmitterService.generateUniqueClientId(member.getEmail(), LocalDateTime.now()),
                NOTIFICATION_EVENT_NAME,
                response)
        );
    }
}