실시간 통신: WebSocket vs SSE vs Webhook

"실시간 통신이 필요하다"
 ├─ 양방향? (채팅/게임/협업)
 │   └─ YES → WebSocket
 │       └─ HTTP 호환 필요? → Socket.IO (폴백 포함)
 ├─ 서버→클라이언트만? (알림/피드/대시보드)
 │   └─ YES → SSE
 │       └─ IE 지원 필요? → Long Polling (폴백)
 └─ 서버→서버? (외부 연동/결제/배송)
     └─ YES → Webhook
         └─ 순서 보장 필요? → 메시지 큐 + Webhook 조합

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        // 외부 브로커(Redis)로 스케일 아웃 지원
        config.enableStompBrokerRelay("/topic", "/queue")
            .setRelayHost("localhost")
            .setRelayPort(61613)       // RabbitMQ STOMP 포트
            .setClientLogin("guest")
            .setClientPasscode("guest")
            .setSystemHeartbeatSendInterval(10000)
            .setSystemHeartbeatReceiveInterval(10000);

        config.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
            .setAllowedOriginPatterns("*")
            .withSockJS();  // SockJS 폴백
    }
}

@Controller
@RequiredArgsConstructor
@Slf4j
public class ChatController {
    private final SimpMessagingTemplate messagingTemplate;
    private final ChatMessageRepository messageRepo;

    @MessageMapping("/chat.send")
    public void sendMessage(@Payload ChatMessage message,
                            SimpMessageHeaderAccessor headerAccessor) {
        String sessionId = headerAccessor.getSessionId();
        message.setTimestamp(Instant.now());
        message.setMessageId(UUID.randomUUID().toString());

        // 영속화 (메시지 유실 방지)
        messageRepo.save(message);

        // 룸 구독자에게 브로드캐스트
        messagingTemplate.convertAndSend(
            "/topic/chat.room." + message.getRoomId(), message);

        log.debug("메시지 전송: room={}, from={}, id={}",
            message.getRoomId(), message.getSender(), message.getMessageId());
    }

    @MessageMapping("/chat.join")
    public void joinRoom(@Payload JoinRequest request,
                         SimpMessageHeaderAccessor headerAccessor) {
        String sessionId = headerAccessor.getSessionId();
        headerAccessor.getSessionAttributes().put("username", request.getUsername());
        headerAccessor.getSessionAttributes().put("roomId", request.getRoomId());

        ChatMessage notification = ChatMessage.builder()
            .type(MessageType.JOIN)
            .sender(request.getUsername())
            .roomId(request.getRoomId())
            .content(request.getUsername() + "님이 입장했습니다.")
            .timestamp(Instant.now())
            .build();

        messagingTemplate.convertAndSend(
            "/topic/chat.room." + request.getRoomId(), notification);
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class WebSocketEventListener {
    private final SimpMessagingTemplate messagingTemplate;
    private final ConnectionRegistry connectionRegistry;

    @EventListener
    public void handleConnect(SessionConnectedEvent event) {
        String sessionId = event.getMessage().getHeaders()
            .get(SimpMessageHeaderAccessor.SESSION_ID_HEADER, String.class);
        connectionRegistry.register(sessionId);
        log.info("WebSocket 연결: sessionId={}", sessionId);
    }

    @EventListener
    public void handleDisconnect(SessionDisconnectEvent event) {
        String sessionId = event.getSessionId();
        ConnectionInfo info = connectionRegistry.remove(sessionId);

        if (info != null && info.getRoomId() != null) {
            ChatMessage leave = ChatMessage.builder()
                .type(MessageType.LEAVE)
                .sender(info.getUsername())
                .roomId(info.getRoomId())
                .content(info.getUsername() + "님이 퇴장했습니다.")
                .timestamp(Instant.now())
                .build();
            messagingTemplate.convertAndSend(
                "/topic/chat.room." + info.getRoomId(), leave);
        }
        log.info("WebSocket 해제: sessionId={}, reason={}",
            sessionId, event.getCloseStatus());
    }
}

@Configuration
public class WebSocketSecurityConfig {

    /**
     * CONNECT 시점에 JWT 검증
     */
    @Bean
    public ChannelInterceptor authInterceptor(JwtTokenProvider tokenProvider) {
        return new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                StompHeaderAccessor accessor =
                    StompHeaderAccessor.wrap(message);

                if (StompCommand.CONNECT.equals(accessor.getCommand())) {
                    String token = accessor.getFirstNativeHeader("Authorization");
                    if (token == null || !token.startsWith("Bearer ")) {
                        throw new MessagingException("인증 토큰 필요");
                    }
                    String jwt = token.substring(7);
                    if (!tokenProvider.validate(jwt)) {
                        throw new MessagingException("토큰 만료/무효");
                    }
                    String userId = tokenProvider.getUserId(jwt);
                    accessor.getSessionAttributes().put("userId", userId);
                }
                return message;
            }
        };
    }
}

@RestController
@RequestMapping("/api/notifications")
@RequiredArgsConstructor
public class NotificationSseController {
    private final NotificationService notificationService;

    /**
     * SSE 스트림 — 무한 Flux + heartbeat
     */
    @GetMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<NotificationDto>> stream(
            @RequestParam String userId,
            @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId) {

        // 1) 놓친 이벤트 복구 (Last-Event-ID 기반)
        Flux<ServerSentEvent<NotificationDto>> missed = Flux.empty();
        if (lastEventId != null) {
            missed = notificationService.getAfter(userId, lastEventId)
                .map(this::toSse);
        }

        // 2) 실시간 이벤트 스트림
        Flux<ServerSentEvent<NotificationDto>> live =
            notificationService.subscribe(userId)
                .map(this::toSse);

        // 3) 하트비트 (30초마다 빈 이벤트 → 연결 유지)
        Flux<ServerSentEvent<NotificationDto>> heartbeat =
            Flux.interval(Duration.ofSeconds(30))
                .map(tick -> ServerSentEvent.<NotificationDto>builder()
                    .comment("heartbeat")
                    .build());

        return Flux.concat(missed, Flux.merge(live, heartbeat))
            .doOnCancel(() -> notificationService.unsubscribe(userId));
    }

    private ServerSentEvent<NotificationDto> toSse(NotificationDto dto) {
        return ServerSentEvent.<NotificationDto>builder()
            .id(dto.getId())                    // Last-Event-ID로 사용
            .event(dto.getType())               // 이벤트 타입
            .data(dto)                          // JSON 페이로드
            .retry(Duration.ofSeconds(5))        // 재연결 간격 힌트
            .build();
    }
}

@Service
@Slf4j
public class NotificationService {
    // userId → Sinks.Many (SSE 구독자)
    private final ConcurrentMap<String, Sinks.Many<NotificationDto>> sinks
        = new ConcurrentHashMap<>();
    private final NotificationRepository notificationRepo;

    public Flux<NotificationDto> subscribe(String userId) {
        Sinks.Many<NotificationDto> sink = sinks.computeIfAbsent(userId,
            k -> Sinks.many().multicast().onBackpressureBuffer(256));
        return sink.asFlux();
    }

    public void unsubscribe(String userId) {
        Sinks.Many<NotificationDto> sink = sinks.remove(userId);
        if (sink != null) sink.tryEmitComplete();
    }

    /**
     * 알림 발행 — DB 저장 + SSE 푸시
     */
    public void publish(String userId, NotificationDto notification) {
        // 영속화 (재연결 시 Last-Event-ID로 복구 가능)
        notificationRepo.save(notification);

        Sinks.Many<NotificationDto> sink = sinks.get(userId);
        if (sink != null) {
            Sinks.EmitResult result = sink.tryEmitNext(notification);
            if (result.isFailure()) {
                log.warn("SSE 발행 실패: userId={}, result={}", userId, result);
            }
        }
    }

    public Flux<NotificationDto> getAfter(String userId, String lastEventId) {
        return notificationRepo.findByUserIdAndIdGreaterThan(userId, lastEventId);
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class WebhookDispatcher {
    private final WebClient webClient;
    private final WebhookEndpointRepository endpointRepo;
    private final WebhookDeliveryLogRepository deliveryLog;

    private static final int MAX_RETRY = 5;
    private static final Duration[] BACKOFF = {
        Duration.ofSeconds(1),   // 1차
        Duration.ofSeconds(5),   // 2차
        Duration.ofSeconds(30),  // 3차
        Duration.ofMinutes(5),   // 4차
        Duration.ofMinutes(30),  // 5차
    };

    /**
     * Webhook 전송 — HMAC-SHA256 서명 + 지수 백오프 재시도
     */
    public Mono<Void> dispatch(String eventType, Object payload) {
        String payloadJson = toJson(payload);
        String deliveryId = UUID.randomUUID().toString();

        return endpointRepo.findByEventType(eventType)
            .flatMap(endpoint -> {
                String signature = hmacSha256(endpoint.getSecret(), payloadJson);

                return webClient.post()
                    .uri(endpoint.getUrl())
                    .header("Content-Type", "application/json")
                    .header("X-Webhook-Id", deliveryId)
                    .header("X-Webhook-Signature", "sha256=" + signature)
                    .header("X-Webhook-Timestamp", Instant.now().toString())
                    .bodyValue(payloadJson)
                    .retrieve()
                    .toBodilessEntity()
                    .timeout(Duration.ofSeconds(5))
                    .retryWhen(Retry.fixedDelay(MAX_RETRY, Duration.ofSeconds(1))
                        .filter(ex -> isRetryable(ex)))
                    .doOnSuccess(resp ->
                        deliveryLog.save(DeliveryLog.success(deliveryId, endpoint)))
                    .doOnError(ex ->
                        deliveryLog.save(DeliveryLog.failure(deliveryId, endpoint, ex)))
                    .then();
            })
            .then();
    }

    /**
     * HMAC-SHA256 서명 생성
     */
    private String hmacSha256(String secret, String payload) {
        try {
            Mac mac = Mac.getInstance("HmacSHA256");
            mac.init(new SecretKeySpec(secret.getBytes(UTF_8), "HmacSHA256"));
            byte[] hash = mac.doFinal(payload.getBytes(UTF_8));
            return Hex.encodeHexString(hash);
        } catch (Exception e) {
            throw new RuntimeException("HMAC 생성 실패", e);
        }
    }

    private boolean isRetryable(Throwable ex) {
        if (ex instanceof WebClientResponseException wcre) {
            int status = wcre.getStatusCode().value();
            return status >= 500 || status == 429;
        }
        return ex instanceof java.net.ConnectException
            || ex instanceof java.util.concurrent.TimeoutException;
    }
}

@RestController
@RequestMapping("/webhooks")
@RequiredArgsConstructor
@Slf4j
public class WebhookReceiverController {
    private final StringRedisTemplate redis;
    private final WebhookProcessingService processingService;

    @Value("${webhook.secret}")
    private String webhookSecret;

    @PostMapping("/payment")
    public ResponseEntity<String> receivePaymentWebhook(
            @RequestBody String payload,
            @RequestHeader("X-Webhook-Id") String webhookId,
            @RequestHeader("X-Webhook-Signature") String signature,
            @RequestHeader("X-Webhook-Timestamp") String timestamp) {

        // 1) 타임스탬프 검증 (5분 이내만 허용 → Replay 방어)
        Instant ts = Instant.parse(timestamp);
        if (Duration.between(ts, Instant.now()).abs().toMinutes() > 5) {
            return ResponseEntity.status(HttpStatus.FORBIDDEN)
                .body("TIMESTAMP_EXPIRED");
        }

        // 2) HMAC 서명 검증
        String expected = "sha256=" + hmacSha256(webhookSecret, payload);
        if (!MessageDigest.isEqual(expected.getBytes(), signature.getBytes())) {
            return ResponseEntity.status(HttpStatus.FORBIDDEN)
                .body("INVALID_SIGNATURE");
        }

        // 3) 멱등성 체크 (같은 webhook-id는 한 번만 처리)
        Boolean isNew = redis.opsForValue()
            .setIfAbsent("webhook:processed:" + webhookId, "1",
                Duration.ofHours(48));
        if (Boolean.FALSE.equals(isNew)) {
            // 이미 처리됨 → 2xx 반환 (발행자가 재시도 안 하도록)
            return ResponseEntity.ok("ALREADY_PROCESSED");
        }

        // 4) 빠르게 2xx 반환 후 비동기 처리
        processingService.processAsync(payload);
        return ResponseEntity.ok("ACCEPTED");
    }
}

# 권장 모니터링 항목
- 전송 성공률: success_count / total_count (목표 > 99%)
- 평균 지연: 전송 시작 → 2xx 수신 시간
- 재시도율: retry_count / total_count (높으면 수신자 장애)
- DLQ 건수: 최대 재시도 초과 건수 (0이 목표)
- 엔드포인트별 실패율: 특정 수신자만 실패하는지

[클라이언트] ←→ [Server A] ←→ [Redis Pub/Sub] ←→ [Server B] ←→ [클라이언트]
                              (메시지 브릿지)

// Redis Pub/Sub 기반 크로스 서버 브로드캐스트
@Component
@RequiredArgsConstructor
public class RedisCrossServerRelay {
    private final RedisTemplate<String, String> redis;
    private final SimpMessagingTemplate messagingTemplate;

    // 메시지 수신 시 → Redis 채널에 발행
    public void broadcastToCluster(String roomId, ChatMessage message) {
        redis.convertAndSend("chat:room:" + roomId, toJson(message));
    }

    // Redis 채널 구독 → 로컬 WebSocket 클라이언트에게 전달
    @Bean
    public RedisMessageListenerContainer messageListenerContainer(
            RedisConnectionFactory factory) {
        var container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.addMessageListener((message, pattern) -> {
            String channel = new String(message.getChannel());
            String roomId = channel.replace("chat:room:", "");
            ChatMessage chatMessage = fromJson(new String(message.getBody()));
            messagingTemplate.convertAndSend(
                "/topic/chat.room." + roomId, chatMessage);
        }, new PatternTopic("chat:room:*"));
        return container;
    }
}

[Producer] → [Redis Streams] → [Server A: SSE] → [클라이언트]
                              → [Server B: SSE] → [클라이언트]

필요 서버 수 = 최대 동시 접속 × 연결당 메모리 / 서버당 가용 메모리

예시:
- 동시 접속 10만
- WebSocket 연결당 ~50KB
- 서버 메모리 8GB, 50% 할당
→ 10만 × 50KB = 5GB → 최소 2대 (여유 포함 3~4대)

@Configuration
public class WebSocketTransportConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
        registration
            .setSendBufferSizeLimit(512 * 1024)      // 512KB
            .setSendTimeLimit(20_000)                  // 20초
            .setMessageSizeLimit(128 * 1024);          // 메시지 최대 128KB
    }
}

class ResilientWebSocket {
  constructor(url) {
    this.url = url;
    this.reconnectAttempt = 0;
    this.maxReconnect = 10;
    this.connect();
  }

  connect() {
    this.ws = new WebSocket(this.url);
    this.ws.onopen = () => {
      console.log('Connected');
      this.reconnectAttempt = 0;     // 성공 시 카운터 리셋
      this.startHeartbeat();
    };
    this.ws.onclose = (event) => {
      this.stopHeartbeat();
      if (this.reconnectAttempt < this.maxReconnect) {
        // 지수 백오프 + 지터
        const delay = Math.min(1000 * 2 ** this.reconnectAttempt, 30000)
                    + Math.random() * 1000;
        this.reconnectAttempt++;
        console.log(`Reconnecting in ${delay}ms (attempt ${this.reconnectAttempt})`);
        setTimeout(() => this.connect(), delay);
      }
    };
  }

  startHeartbeat() {
    this.heartbeatInterval = setInterval(() => {
      if (this.ws.readyState === WebSocket.OPEN) {
        this.ws.send(JSON.stringify({ type: 'ping' }));
      }
    }, 25000);   // 25초마다 ping
  }

  stopHeartbeat() {
    if (this.heartbeatInterval) clearInterval(this.heartbeatInterval);
  }
}