feat(realtime): 실시간 푸시를 Core 직접 SSE → RealTime 서버 경유로 전환

backend/src/main/java/com/stackup/stackup/auth/application/AuthService.java

@@ -94,7 +94,7 @@ public StreamTokenResult createStreamToken(Long userId) {
         if (userId == null) {
             throw new BadCredentialsException("Authentication is required to create stream token");
         }
-        return new StreamTokenResult(streamTokenProvider.createStreamToken(userId));
+        return new StreamTokenResult(streamTokenProvider.createStreamToken(userId, "USER", userId));
     }
 
 }

backend/src/main/java/com/stackup/stackup/common/config/properties/RabbitMqProperties.java

@@ -88,7 +88,9 @@ public record RoutingKeyProperties(
 		@NotBlank String callbackQuestions,
 		@NotBlank String callbackFeedback,
 		@NotBlank String callbackVoice,
-		@NotBlank String realtimeSessionNotify
+		@NotBlank String realtimeSessionNotify,
+		@NotBlank String realtimeUserNotify,
+		@NotBlank String realtimeDocumentNotify
 	) {
 	}
 

backend/src/main/java/com/stackup/stackup/common/messaging/RealtimeNotifyEvent.java

@@ -0,0 +1,21 @@
+package com.stackup.stackup.common.messaging;
+
+import com.stackup.stackup.common.sse.SseEventType;
+
+// Core 도메인 트랜잭션 안에서 발행하는 알림 의도. 실제 RabbitMQ publish 는
+// RealtimeNotifyEventListener 가 AFTER_COMMIT 에서 수행한다 (롤백 시 미발행 보장).
+public record RealtimeNotifyEvent(Channel channel, Long id, SseEventType type, Object payload) {
+    public enum Channel { SESSION, USER, DOCUMENT }
+
+    public static RealtimeNotifyEvent session(Long sessionId, SseEventType type, Object payload) {
+        return new RealtimeNotifyEvent(Channel.SESSION, sessionId, type, payload);
+    }
+
+    public static RealtimeNotifyEvent user(Long userId, SseEventType type, Object payload) {
+        return new RealtimeNotifyEvent(Channel.USER, userId, type, payload);
+    }
+
+    public static RealtimeNotifyEvent document(Long documentId, SseEventType type, Object payload) {
+        return new RealtimeNotifyEvent(Channel.DOCUMENT, documentId, type, payload);
+    }
+}

backend/src/main/java/com/stackup/stackup/common/messaging/RealtimeNotifyEventListener.java

@@ -0,0 +1,23 @@
+package com.stackup.stackup.common.messaging;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+// 도메인 트랜잭션 커밋 후 RealTime 알림을 발행한다. 롤백 시 미발행 → DB 와 알림의 일관성 보장.
+@Component
+@RequiredArgsConstructor
+public class RealtimeNotifyEventListener {
+
+    private final RealtimeNotifyPublisher publisher;
+
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void on(RealtimeNotifyEvent event) {
+        switch (event.channel()) {
+            case SESSION -> publisher.publishToSession(event.id(), event.type(), event.payload());
+            case USER -> publisher.publishToUser(event.id(), event.type(), event.payload());
+            case DOCUMENT -> publisher.publishToDocument(event.id(), event.type(), event.payload());
+        }
+    }
+}

backend/src/main/java/com/stackup/stackup/common/messaging/RealtimeNotifyPublisher.java

@@ -1,15 +1,19 @@
 package com.stackup.stackup.common.messaging;
 
 import com.stackup.stackup.common.config.properties.RabbitMqProperties;
+import com.stackup.stackup.common.sse.SseEventType;
 import com.stackup.stackup.common.trace.TraceContext;
 import java.time.Instant;
 import java.util.UUID;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.stereotype.Component;
 
-// Core -> 실시간 서버 알림 
-// 실시간 서버는 SSE로 구독한 컴포넌트에게 알려줌
+// Core -> RealTime 서버 알림 발행. RealTime 이 SSE/WS 구독자에게 fan-out 한다.
+// 채널별 messageType(realtime.{session|user|document}.notify)으로 발행하며,
+// RealTime 의 bridge.Envelope.Channel() 이 messageType + context 로 채널을 판별한다.
+// 호출은 콜백 서비스의 @Transactional 안에서 DB 영속 이후 일어나므로, SSE 페이로드의 ID 는
+// 프론트가 조회 가능한 상태가 보장된다.
 @Component
 public class RealtimeNotifyPublisher {
 
@@ -24,17 +28,27 @@ public RealtimeNotifyPublisher(RabbitMqProperties properties, RabbitTemplate rab
         this.rabbitTemplate = rabbitTemplate;
     }
 
-    public MessageEnvelope<RealtimeNotifyPayload> publishSessionNotify(
-        long sessionId,
-        Long userId,
-        String eventType,
-        Object data
-    ) {
-        var payload = new RealtimeNotifyPayload(eventType, data);
-        var context = new MessageContext(userId, sessionId, null, null);
+    public void publishToSession(Long sessionId, SseEventType type, Object data) {
+        publish(properties.routingKeys().realtimeSessionNotify(),
+            new MessageContext(null, sessionId, null, null), type, data);
+    }
+
+    public void publishToUser(Long userId, SseEventType type, Object data) {
+        publish(properties.routingKeys().realtimeUserNotify(),
+            new MessageContext(userId, null, null, null), type, data);
+    }
+
+    public void publishToDocument(Long documentId, SseEventType type, Object data) {
+        publish(properties.routingKeys().realtimeDocumentNotify(),
+            new MessageContext(null, null, documentId, null), type, data);
+    }
+
+    private void publish(String routingKey, MessageContext context, SseEventType type, Object data) {
+        var payload = new RealtimeNotifyPayload(type.name(), data);
+        // messageType == routing key (realtime.{kind}.notify) — RealTime 이 이 값으로 채널을 판별한다.
         var envelope = new MessageEnvelope<>(
             UUID.randomUUID().toString(),
-            "realtime.session.notify",
+            routingKey,
             properties.version(),
             TraceContext.getTraceId(),
             Instant.now(),
@@ -45,11 +59,10 @@ public MessageEnvelope<RealtimeNotifyPayload> publishSessionNotify(
 
         rabbitTemplate.convertAndSend(
             properties.exchanges().names().realtime(),
-            properties.routingKeys().realtimeSessionNotify(),
+            routingKey,
             envelope,
             withEnvelopeHeaders(envelope)
         );
-        return envelope;
     }
 
     private MessagePostProcessor withEnvelopeHeaders(MessageEnvelope<?> envelope) {

backend/src/main/java/com/stackup/stackup/common/security/SecurityConfig.java

@@ -30,18 +30,15 @@ public class SecurityConfig {
 
     private final JwtAuthenticationFilter jwtAuthenticationFilter;
     private final InternalApiKeyAuthenticationFilter internalApiKeyAuthenticationFilter;
-    private final StreamTokenAuthenticationFilter streamTokenAuthenticationFilter;
     private final CorsProperties corsProperties;
 
     public SecurityConfig(
         JwtAuthenticationFilter jwtAuthenticationFilter,
         InternalApiKeyAuthenticationFilter internalApiKeyAuthenticationFilter,
-        StreamTokenAuthenticationFilter streamTokenAuthenticationFilter,
         CorsProperties corsProperties
     ) {
         this.jwtAuthenticationFilter = jwtAuthenticationFilter;
         this.internalApiKeyAuthenticationFilter = internalApiKeyAuthenticationFilter;
-        this.streamTokenAuthenticationFilter = streamTokenAuthenticationFilter;
         this.corsProperties = corsProperties;
     }
 
@@ -74,7 +71,6 @@ public SecurityFilterChain securityFilterChain(HttpSecurity http) throws Excepti
                 .anyRequest().permitAll()
             )
             .addFilterBefore(internalApiKeyAuthenticationFilter, UsernamePasswordAuthenticationFilter.class)
-            .addFilterBefore(streamTokenAuthenticationFilter, UsernamePasswordAuthenticationFilter.class)
             .addFilterBefore(jwtAuthenticationFilter, UsernamePasswordAuthenticationFilter.class)
             .build();
     }

backend/src/main/java/com/stackup/stackup/common/security/StreamTokenProvider.java

@@ -26,6 +26,11 @@ public class StreamTokenProvider {
     private static final String SSE_CONNECT_SCOPE = "SSE_CONNECT";
     private static final String TOKEN_TYPE_CLAIM = "tokenType";
     private static final String STREAM_TOKEN_TYPE = "STREAM";
+    private static final String RESOURCE_TYPE_CLAIM = "resourceType";
+    private static final String RESOURCE_ID_CLAIM = "resourceId";
+
+    public record StreamTokenClaims(Long userId, String resourceType, Long resourceId) {
+    }
 
     private final SseProperties sseProperties;
     private final byte[] secretKey;
@@ -35,20 +40,33 @@ public StreamTokenProvider(SecurityProperties securityProperties, SseProperties
         this.secretKey = sha256(securityProperties.jwtSecret());
     }
 
-    public String createStreamToken(Long userId) {
+    public String createStreamToken(Long userId, String resourceType, Long resourceId) {
         Instant now = Instant.now();
         JWTClaimsSet claimsSet = new JWTClaimsSet.Builder()
             .subject(String.valueOf(userId))
             .claim(USER_ID_CLAIM, userId)
             .claim(TOKEN_TYPE_CLAIM, STREAM_TOKEN_TYPE)
             .claim(SCOPE_CLAIM, SSE_CONNECT_SCOPE)
+            .claim(RESOURCE_TYPE_CLAIM, resourceType)
+            .claim(RESOURCE_ID_CLAIM, resourceId)
             .issueTime(Date.from(now))
             .expirationTime(Date.from(now.plus(sseProperties.streamTokenTtl())))
             .build();
-
         return sign(claimsSet);
     }
 
+    public StreamTokenClaims parseClaims(String token) {
+        JWTClaimsSet claims = parseAndValidate(token);
+        try {
+            Long userId = claims.getLongClaim(USER_ID_CLAIM);
+            String resourceType = claims.getStringClaim(RESOURCE_TYPE_CLAIM);
+            Long resourceId = claims.getLongClaim(RESOURCE_ID_CLAIM);
+            return new StreamTokenClaims(userId, resourceType, resourceId);
+        } catch (java.text.ParseException exception) {
+            throw invalidToken();
+        }
+    }
+
     public boolean validateStreamToken(String token) {
         parseAndValidate(token);
         return true;