[FEAT] SSE 구현 #19
[FEAT] SSE 구현 #19
Conversation
|
Caution Review failedThe pull request is closed. 📝 WalkthroughSummary by CodeRabbit릴리스 노트
✏️ Tip: You can customize this high-level summary in your review settings. Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
📜 Recent review detailsConfiguration used: Repository UI Review profile: CHILL Plan: Pro 📒 Files selected for processing (1)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
Fix all issues with AI Agents 🤖
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventPublisher.kt:
- Around line 16-26: OutboxEventPublisher.publish currently calls
objectMapper.writeValueAsString(payload) without handling serialization
failures; wrap the serialization and subsequent
applicationEventPublisher.publishEvent(OutboxPublishEvent(...)) in a try/catch
that catches JsonProcessingException (or a broader Exception) so serialization
errors do not bubble up to callers, log the failure with contextual info
(userId, eventType) via the class logger, and skip or publish a safe fallback
(e.g., omit publishing) rather than letting the exception interrupt business
logic; keep references to objectMapper.writeValueAsString,
applicationEventPublisher.publishEvent, and OutboxPublishEvent when locating the
change.
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventRecordListener.kt:
- Around line 17-28: The listener OutboxEventRecordListener.record currently
uses @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT) which
will cause failures in OutboxSendEvent handlers to roll back the business
transaction; change the phase to TransactionPhase.AFTER_COMMIT so
OutboxSendEvent is published only after OutboxEvent.save succeeds and send
failures don't rollback the original transaction, keeping the existing save and
publishing logic (OutboxEvent.create, outboxEventRepository.save,
applicationEventPublisher.publishEvent(OutboxSendEvent(...))) intact.
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventSendListener.kt:
- Around line 16-19: OutboxEventSendListener의 send(OutboxSendEvent) 메서드가
outboxEventService.sendAsync(event.outboxEventId) 호출 중 발생하는 예외를 처리하지 않아 다른 리스너
실행을 방해할 수 있으므로, send 내부에서 try-catch로 예외를 잡아 처리하고 적절히 로깅하세요; 구체적으로
send(OutboxSendEvent) 메서드에서 outboxEventService.sendAsync 호출을 감싸고 Throwable을 캐치해
에러 로그를 남기되 예외를 무시하거나 재시도 로직을 구현할 필요가 있으면 그 정책을 적용하고, 트랜잭션
어노테이션(@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT))은 그대로
유지하세요.
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.kt:
- Around line 42-44: 현재 SSE 구독 핸들러(SseSubscribeController)의 isConnected 분기에서
호출되는 outboxEventService.enqueuePendingSends가 내부에서
outboxEventRepository.findPendingIds()를 동기적으로 실행해 HTTP 요청 스레드를 블로킹하고 있으므로,
enqueuePendingSends(또는 그 내부 DB 조회 부분)를 별도 스레드/비동기 실행으로 옮기거나 리액티브 리포지토리로 리팩토링하세요;
구체적으로는 SseSubscribeController의 분기에서는 enqueuePendingSends 호출을 유지하되 해당 메서드 구현을
변경하여 findPendingIds 호출을 Executor(예: CompletableFuture.runAsync 또는 지정된
TaskExecutor)로 오프로드하거나 R2DBC/ReactiveCrudRepository 등으로 대체해 논블로킹 방식으로 조회 후 결과를
기존 executor에 전달하도록 수정하세요.
In @src/main/kotlin/org/appjam/smashing/domain/outbox/dto/OutboxPublishEvent.kt:
- Around line 5-9: Add defensive bean-validation: annotate the
OutboxPublishEvent data class properties with Kotlin field-targeted validation
annotations (e.g., @field:NotBlank on userId and payloadJson, and @field:NotNull
on eventType) and mirror the same constraints on the OutboxEvent entity fields.
Then ensure OutboxEventPublisher.publish() receives a validated parameter by
annotating the method parameter with @Valid (or programmatically validate using
a javax.validation.Validator) so invalid DTOs are rejected before persistence.
Include necessary imports (javax.validation.constraints.* and
javax.validation.Valid) and adjust any mapping code to rely on validated source
objects.
In @src/main/kotlin/org/appjam/smashing/domain/outbox/entity/OutboxEvent.kt:
- Around line 71-99: The entity method markAsFailed() increments retryCount and
sets status, but OutboxEventRepository.markFailedIfProcessing() does not update
retryCount or derive status, causing mismatch and race conditions; update the
repository query markFailedIfProcessing to atomically increment retryCount, set
lastAttemptAt = NOW(), and conditionally set status = FAILED when retryCount+1
>= MAX_RETRY else PENDING (and populate sentAt/null as appropriate), and then
ensure service layer uses the repository method as the single source of truth
instead of calling markAsFailed() on the entity.
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/repository/OutboxEventRepository.kt:
- Around line 13-24: The method name findPendingIds in OutboxEventRepository is
misleading because the query returns IDs for any passed status value; rename the
method to a clearer name like findIdsByUserAndStatus or findEventIdsByStatus to
reflect its behavior, update all call sites to use the new name, and ensure the
function signature (userId: String, status: OutboxEventStatus, pageable:
Pageable): List<String> and @Query remain unchanged so only the identifier is
updated.
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventSender.kt:
- Around line 37-44: The current runCatching around sseEmitterRegistry.send only
checks success boolean and discards the exception and save() errors; change it
to capture the thrown exception (from the runCatching or try/catch around
sseEmitterRegistry.send) and log the failure with actionable details (include
event.id or event.aggregateId, event.userId, event.eventType.eventName and a
truncated payload) before calling event.markAsFailed(); after updating the event
status, persist with outboxEventRepository.save inside its own try/catch so save
failures are logged (include the same identifiers) and do not swallow exceptions
silently. Ensure you still call event.markAsSent() on success and log success at
debug/info level.
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventService.kt:
- Around line 31-37: sendAsync currently calls outboxEventExecutor.execute {
outboxEventSender.send(outboxEventId) } without any error handling; add handling
for both executor rejection and failures during send by wrapping the execute
call in a try/catch that catches RejectedExecutionException (log or handle
executor rejection) and by wrapping the runnable body (the call to
outboxEventSender.send(outboxEventId)) in its own try/catch to handle/log send
failures or report them for retry, mirroring the error-handling approach used in
enqueuePendingSends; reference sendAsync, outboxEventExecutor,
outboxEventSender.send for locating places to add these catches and logging.
- Around line 26-28: Wrap the submission and task execution with proper error
handling: when iterating ids.forEach and calling outboxEventExecutor.execute,
catch RejectedExecutionException around the execute call so rejected tasks are
logged and handled (e.g., enqueue to a fallback queue or re-submit later);
inside the Runnable submitted to outboxEventExecutor.execute wrap
outboxEventSender.send(id) in a try-catch that logs failures (include id and
exception) and implements a retry or failure handling strategy
(dead-letter/fallback) so exceptions during send are not lost and are traceable.
🧹 Nitpick comments (19)
src/main/kotlin/org/appjam/smashing/domain/outbox/components/CurrentUserProvider.kt (2)
11-11: TODO 주석을 이슈로 추적하세요.패키지 분리 작업을 추적하기 위해 TODO 주석을 별도 이슈로 등록하는 것을 권장합니다.
이슈 생성을 도와드릴까요?
13-27: anonymousUser 케이스 처리를 고려하세요.
AuditorAwareImpl과 달리 이 구현체는anonymousUser를 명시적으로 처리하지 않습니다. SSE 구독은 인증된 사용자만 가능하므로 현재 로직이 적절하지만, 일관성을 위해authentication.name == "anonymousUser"체크를 추가하거나 이를 의도적으로 제외한 이유를 문서화하는 것이 좋습니다.Based on learnings, 코드베이스의 기존
AuditorAwareImpl패턴과의 일관성을 고려했습니다.src/main/kotlin/org/appjam/smashing/global/config/AsyncConfig.kt (3)
15-18: 스레드 풀 설정을 외부화하는 것을 고려하세요.스레드 풀 크기와 큐 용량이 하드코딩되어 있습니다. 운영 환경에서 부하에 따라 조정할 수 있도록
application.yml을 통해 설정 가능하게 만드는 것을 권장합니다.🔎 제안하는 개선 방안
application.yml에 설정 추가:async: outbox: core-pool-size: 4 max-pool-size: 8 queue-capacity: 500 keep-alive-seconds: 30그리고
@ConfigurationProperties를 사용하여 바인딩합니다.
12-23: Rejection Policy를 명시적으로 설정하세요.
RejectedExecutionHandler가 설정되지 않아 기본값인AbortPolicy가 사용됩니다. 큐가 가득 찰 경우RejectedExecutionException이 발생합니다. SSE 이벤트 전송의 중요도에 따라CallerRunsPolicy또는 커스텀 정책을 고려해보세요.🔎 제안하는 수정
executor.setThreadNamePrefix("Outbox-Sse-") +executor.setRejectedExecutionHandler(ThreadPoolExecutor.CallerRunsPolicy()) executor.initialize()
18-18: keepAliveSeconds는 maxPoolSize 이상의 스레드에만 적용됩니다.현재 설정에서
keepAliveSeconds는 corePoolSize(4)를 초과하는 스레드에만 적용됩니다. 코어 스레드도 타임아웃시키려면executor.setAllowCoreThreadTimeOut(true)를 추가해야 합니다.src/main/kotlin/org/appjam/smashing/domain/outbox/components/SseEmitterRegistry.kt (4)
32-32:isNotEmpty()사용을 고려하세요.
peek() != null대신isNotEmpty()를 사용하면 의도가 더 명확합니다.🔎 제안하는 수정
-fun hasConnection(userId: String): Boolean = emitters[userId]?.peek() != null +fun hasConnection(userId: String): Boolean = emitters[userId]?.isNotEmpty() == true
37-55: SSE 전송 실패에 대한 로깅을 추가하세요.Line 52-53에서 전송 실패 시 emitter를 조용히 제거합니다. 디버깅과 모니터링을 위해 로깅을 추가하는 것을 권장합니다.
🔎 제안하는 수정
for (emitter in queue) { - val isSucceed = runCatching { emitter.send(event) }.isSuccess - if (!isSucceed) remove(userId, emitter) + runCatching { + emitter.send(event) + }.onFailure { e -> + logger.warn("Failed to send SSE event to user $userId: ${e.message}") + remove(userId, emitter) + } }클래스 상단에 logger 추가:
companion object { private val logger = LoggerFactory.getLogger(SseEmitterRegistry::class.java) }
17-27: 사용자당 emitter 개수 제한을 고려하세요.현재 구현은 단일 사용자가 무제한으로 SSE 연결을 생성할 수 있어 리소스 고갈 위험이 있습니다. 사용자당 최대 연결 수를 제한하는 것을 권장합니다(예: 5개).
최대 연결 수 제한 로직 구현을 도와드릴까요?
88-88:isNotEmpty()로 리팩토링하세요.
takeIf { it.peek() != null }은takeIf { it.isNotEmpty() }로 더 명확하게 표현할 수 있습니다.🔎 제안하는 수정
-queue.takeIf { it.peek() != null } +queue.takeIf { it.isNotEmpty() }src/main/kotlin/org/appjam/smashing/domain/outbox/enums/MatchingUpdateStatus.kt (1)
3-6: SSE 페이로드 전용 enum의 명확성을 개선하세요.코드베이스의 패턴을 검토한 결과,
MatchingUpdateStatus는 SSE 페이로드 전용이고MatchingStatus와의 값 집합이 다릅니다(ACCEPTED, REJECTED만 필요). 따라서 enum 재사용은 적절하지 않으며, 현재outbox패키지 위치가 용도를 암시합니다.그러나
SseMatchingUpdateStatus로 명명하면 의도가 더 명확해질 것 같습니다. 현재 이름도 합리적이지만, 특히 기능이 완전히 통합될 때 네이밍을 재검토할 가치가 있습니다.src/main/kotlin/org/appjam/smashing/domain/outbox/enums/SseEventType.kt (1)
3-14: LGTM! 명확한 이벤트 타입 정의입니다.이벤트 타입이 명확하게 정의되어 있고 dot notation 컨벤션을 일관되게 사용하고 있습니다.
NotificationType열거형과도 적절히 구분되어 있습니다.선택사항: SseEventType과 NotificationType 간 매핑 관계 문서화
SseEventType의NOTIFICATION_CREATED와NotificationType(예:MATCHING_REQUESTED,MATCHING_ACCEPTED) 간의 관계가 명확하지 않습니다. 이들 간의 매핑이나 변환 로직이 필요한 경우, 해당 관계를 문서화하거나 유틸리티 함수를 추가하는 것을 고려해보세요.src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventService.kt (1)
19-24: 하드코딩된 페이지 크기를 설정 가능하도록 변경하세요.200이라는 페이지 크기가 하드코딩되어 있습니다. 이는 환경이나 부하에 따라 조정이 필요할 수 있으므로, 설정 파일이나 생성자 파라미터를 통해 설정할 수 있도록 변경하는 것이 좋습니다.
🔎 설정 가능한 페이지 크기로 변경하는 제안
@Service class OutboxEventService( private val outboxEventRepository: OutboxEventRepository, private val outboxEventSender: OutboxEventSender, @Qualifier("outboxEventExecutor") private val outboxEventExecutor: Executor, @Value("\${outbox.pending.page-size:200}") private val pendingPageSize: Int, ) { fun enqueuePendingSends(userId: String) { val ids = outboxEventRepository.findPendingIds( userId = userId, status = OutboxEventStatus.PENDING, pageable = PageRequest.of(0, pendingPageSize), ) // ... } }src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventSender.kt (1)
30-34: TODO 항목 처리 지원이 가능합니다.
PROCESSING상태에서 고여있는 이벤트들을 처리하기 위한 스케줄러가 필요하다는 TODO가 있습니다. 이는 네트워크 문제나 예외 발생 시 이벤트가 영구적으로PROCESSING상태에 머무를 수 있는 중요한 시나리오입니다.스케줄러 구현 코드를 생성하거나 이를 추적하기 위한 이슈를 생성하시겠습니까?
src/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.kt (1)
31-39: 초기 연결 이벤트 전송 실패 시 로깅을 추가하세요.초기
system.connected이벤트 전송 실패 시 emitter가 제거되지만, 실패 원인이 로깅되지 않아 디버깅이 어렵습니다. 연결 실패 원인을 파악하기 위한 로깅을 추가하는 것을 권장합니다.🔎 로깅 추가 제안
+private val logger = LoggerFactory.getLogger(SseSubscribeController::class.java) + val isConnected = runCatching { emitter.send( SseEmitter.event() .name(SYSTEM_CONNECTED_EVENT) .data("""{"type":"$SYSTEM_CONNECTED_EVENT"}""", MediaType.APPLICATION_JSON) ) }.onFailure { + logger.warn("Failed to send initial SSE connect event: userId={}", userId, it) sseEmitterRegistry.removeEmitter(userId, emitter) }.isSuccesssrc/main/kotlin/org/appjam/smashing/domain/outbox/entity/OutboxEvent.kt (3)
14-14: 인덱스 최적화 고려사항.현재 인덱스
(user_id, status, id)는 사용자별 상태 조회에 적합하지만, 시간 기반 쿼리(예: 오래된 PROCESSING 레코드 찾기)를 고려하면lastAttemptAt또는createdAt컬럼 추가를 검토하세요. 특히 재처리나 모니터링 쿼리에서 타임아웃된 이벤트를 찾을 때 유용합니다.🔎 제안하는 인덱스 추가
@Table( indexes = [ Index(name = "idx_outbox_user_status_id", columnList = "user_id,status,id"), + Index(name = "idx_outbox_status_last_attempt", columnList = "status,lastAttemptAt"), ] )
35-37: JSON 페이로드 유효성 검증 누락.
payload필드가 JSON 문자열로 저장되지만, 유효성 검증이 없습니다. 잘못된 JSON이 저장되면 SSE 전송 시점에 파싱 오류가 발생할 수 있습니다. 엔티티 생성 시점에 JSON 유효성을 검증하거나, Jackson의@JsonRawValue와 같은 타입 안전 접근을 고려하세요.
55-55: Outbox 패턴에 BaseCreatedEntity 사용 검토.Outbox 이벤트는 일시적인(transient) 데이터로, 처리 후 정리되는 경우가 많습니다.
BaseCreatedEntity의 감사 필드(createdBy,updatedBy,updatedAt,deletedAt)가 모두 필요한지 확인하세요. 불필요한 컬럼은 스토리지와 인덱스 오버헤드를 증가시킵니다. 최소 필드만 포함하는BaseCreatedEntity대신createdAt만 직접 선언하는 것을 고려하세요.src/main/kotlin/org/appjam/smashing/domain/outbox/repository/OutboxEventRepository.kt (2)
34-39: Repository 메서드의 기본 파라미터 사용은 비관례적.
pending,processing파라미터에 기본값을 설정하는 것은 Spring Data JPA repository에서 일반적이지 않으며, 호출자에게 혼란을 줄 수 있습니다. 이러한 상태값은 비즈니스 로직의 핵심이므로 명시적으로 전달하거나, 쿼리 내부에 하드코딩하는 것이 더 명확합니다.권장사항:
- 기본 파라미터를 제거하고 호출자가 명시적으로 전달
- 또는 쿼리 내에서 enum을 직접 참조 (예:
:#{T(org.appjam.smashing.domain.outbox.enums.OutboxEventStatus).PENDING})🔎 제안하는 리팩토링 (옵션 1: 기본값 제거)
fun markProcessingIfPending( id: String, - pending: OutboxEventStatus = OutboxEventStatus.PENDING, - processing: OutboxEventStatus = OutboxEventStatus.PROCESSING, now: LocalDateTime, ): Int쿼리를 다음과 같이 수정:
@Query(""" update OutboxEvent e - set e.status = :processing, + set e.status = :#{T(org.appjam.smashing.domain.outbox.enums.OutboxEventStatus).PROCESSING}, e.lastAttemptAt = :now where e.id = :id - and e.status = :pending + and e.status = :#{T(org.appjam.smashing.domain.outbox.enums.OutboxEventStatus).PENDING} """)
49-54: 동일한 기본 파라미터 패턴 반복.
markPendingIfProcessing과markFailedIfProcessing메서드에도 동일한 기본 파라미터 패턴이 사용되고 있습니다. 앞서 지적한 것과 동일한 이유로 일관성을 위해 수정을 권장합니다.Also applies to: 64-69
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (17)
src/main/kotlin/org/appjam/smashing/domain/outbox/components/CurrentUserProvider.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventPublisher.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventRecordListener.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventSendListener.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/components/SseEmitterRegistry.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/dto/OutboxPublishEvent.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/dto/OutboxSendEvent.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/dto/SsePayload.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/entity/OutboxEvent.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/enums/MatchingUpdateStatus.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/enums/OutboxEventStatus.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/enums/SseEventType.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/repository/OutboxEventRepository.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventSender.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventService.ktsrc/main/kotlin/org/appjam/smashing/global/config/AsyncConfig.kt
🧰 Additional context used
🧬 Code graph analysis (7)
src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventPublisher.kt (1)
src/main/kotlin/org/appjam/smashing/SmashingApplication.kt (1)
@SpringBootApplication(7-9)
src/main/kotlin/org/appjam/smashing/domain/outbox/enums/MatchingUpdateStatus.kt (7)
src/main/kotlin/org/appjam/smashing/domain/matching/enums/MatchingStatus.kt (1)
REQUESTED(3-5)src/main/kotlin/org/appjam/smashing/domain/matching/enums/SubmissionStatus.kt (1)
SUBMITTED(3-5)src/main/kotlin/org/appjam/smashing/domain/notification/enums/NotificationType.kt (1)
MATCHING_REQUESTED(3-14)src/main/kotlin/org/appjam/smashing/global/common/enums/Status.kt (1)
SUCCESS(3-7)src/main/kotlin/org/appjam/smashing/domain/matching/enums/GameResultStatus.kt (1)
PENDING_RESULT(3-5)src/main/kotlin/org/appjam/smashing/domain/matching/enums/RejectReason.kt (1)
SCORE_MISMATCH(3-6)src/main/kotlin/org/appjam/smashing/domain/matching/entity/Matching.kt (1)
indexes(14-71)
src/main/kotlin/org/appjam/smashing/domain/outbox/components/CurrentUserProvider.kt (3)
src/main/kotlin/org/appjam/smashing/domain/auth/jwt/CustomUserDetails.kt (2)
userId(6-23)getUsername(14-14)src/main/kotlin/org/appjam/smashing/global/config/AuditorAwareImpl.kt (2)
getCurrentAuditor(7-18)getCurrentAuditor(9-17)src/main/kotlin/org/appjam/smashing/domain/auth/jwt/JwtProvider.kt (2)
getAuthentication(32-57)jwtGenerator(12-62)
src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventSender.kt (1)
src/main/kotlin/org/appjam/smashing/SmashingApplication.kt (1)
@SpringBootApplication(7-9)
src/main/kotlin/org/appjam/smashing/domain/outbox/entity/OutboxEvent.kt (3)
src/main/kotlin/org/appjam/smashing/domain/notification/entity/Notification.kt (1)
indexes(11-57)src/main/kotlin/org/appjam/smashing/domain/common/entity/BaseEntity.kt (1)
name(15-49)src/main/kotlin/org/appjam/smashing/domain/notification/entity/NotificationTemplate.kt (1)
strategy(7-27)
src/main/kotlin/org/appjam/smashing/domain/outbox/enums/SseEventType.kt (1)
src/main/kotlin/org/appjam/smashing/domain/notification/enums/NotificationType.kt (1)
MATCHING_REQUESTED(3-14)
src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventSendListener.kt (1)
src/main/kotlin/org/appjam/smashing/SmashingApplication.kt (1)
@SpringBootApplication(7-9)
🔇 Additional comments (4)
src/main/kotlin/org/appjam/smashing/domain/outbox/enums/OutboxEventStatus.kt (1)
3-8: LGTM!아웃박스 패턴의 상태 전이를 명확하게 표현하는 적절한 enum 정의입니다.
src/main/kotlin/org/appjam/smashing/domain/outbox/dto/OutboxSendEvent.kt (1)
3-5: LGTM! 단순하고 명확한 이벤트 DTO입니다.트랜잭션 커밋 후 비동기 전송을 트리거하기 위한 내부 이벤트로 적절합니다.
src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventRecordListener.kt (1)
21-23: 코드가 이미 null 체크로 보호되고 있으며, @tsid는 신뢰할 수 있는 ID 생성 전략입니다.OutboxEvent는 @GeneratedValue가 아닌 @tsid를 ID 생성 전략으로 사용 중이며, 이는 매우 신뢰할 수 있는 방식입니다. 게다가 OutboxEventRecordListener의 24번 라인에서
requireNotNull(savedOutboxEvent.id)로 이미 null 체크를 수행하고 있으므로, 저장 후 ID가 null인 경우 즉시 예외가 발생하게 됩니다. 따라서 추가적인 검증이 필요하지 않습니다.src/main/kotlin/org/appjam/smashing/domain/outbox/dto/SsePayload.kt (1)
7-40: 깔끔한 페이로드 설계입니다!Sealed interface를 활용한 타입 안전한 페이로드 구조가 잘 설계되었습니다. 각 페이로드 타입이
SseEventTypeenum의eventName을 기본값으로 사용하는 패턴도 일관성 있고 좋습니다.targetId의 nullable 처리도 적절합니다.
src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventPublisher.kt
Outdated
Show resolved
Hide resolved
| @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT) | ||
| fun record( | ||
| event: OutboxPublishEvent | ||
| ) { | ||
| val savedOutboxEvent = outboxEventRepository.save( | ||
| OutboxEvent.create(event.userId, event.eventType, event.payloadJson) | ||
| ) | ||
|
|
||
| applicationEventPublisher.publishEvent( | ||
| OutboxSendEvent(requireNotNull(savedOutboxEvent.id)) | ||
| ) | ||
| } |
There was a problem hiding this comment.
트랜잭션 페이즈 선택을 재검토하세요.
BEFORE_COMMIT 페이즈를 사용하면 OutboxSendEvent를 처리하는 리스너가 실패할 경우 전체 트랜잭션이 롤백됩니다. 일반적인 아웃박스 패턴에서는 데이터 영속성을 보장하기 위해 AFTER_COMMIT을 사용합니다.
현재 구현에서 BEFORE_COMMIT을 선택한 이유가 명확하지 않습니다. 만약 의도적인 설계가 아니라면, 이벤트 전송 실패가 비즈니스 트랜잭션 롤백으로 이어질 수 있어 문제가 될 수 있습니다.
🔎 AFTER_COMMIT 페이즈로 변경하는 제안
- @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
fun record(
event: OutboxPublishEvent
) {이렇게 변경하면 OutboxEvent가 성공적으로 저장된 후에만 OutboxSendEvent가 발행되어, 전송 실패가 비즈니스 로직에 영향을 주지 않습니다.
🤖 Prompt for AI Agents
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventRecordListener.kt
around lines 17-28, The listener OutboxEventRecordListener.record currently uses
@TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT) which will
cause failures in OutboxSendEvent handlers to roll back the business
transaction; change the phase to TransactionPhase.AFTER_COMMIT so
OutboxSendEvent is published only after OutboxEvent.save succeeds and send
failures don't rollback the original transaction, keeping the existing save and
publishing logic (OutboxEvent.create, outboxEventRepository.save,
applicationEventPublisher.publishEvent(OutboxSendEvent(...))) intact.
| @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) | ||
| fun send(event: OutboxSendEvent) { | ||
| outboxEventService.sendAsync(event.outboxEventId) | ||
| } |
There was a problem hiding this comment.
리스너 메서드에 예외 처리 추가를 권장합니다.
AFTER_COMMIT 단계의 리스너에서 발생하는 예외는 이미 커밋된 트랜잭션에는 영향을 주지 않지만, 다른 리스너들의 실행을 방해하거나 예상치 못한 동작을 유발할 수 있습니다. sendAsync 호출 시 발생할 수 있는 예외를 처리하고 로깅하는 것을 권장합니다.
🔎 예외 처리 추가 제안
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
fun send(event: OutboxSendEvent) {
- outboxEventService.sendAsync(event.outboxEventId)
+ runCatching {
+ outboxEventService.sendAsync(event.outboxEventId)
+ }.onFailure { e ->
+ // TODO: 로깅 추가 (logger.error("Failed to enqueue outbox send", e))
+ }
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) | |
| fun send(event: OutboxSendEvent) { | |
| outboxEventService.sendAsync(event.outboxEventId) | |
| } | |
| @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) | |
| fun send(event: OutboxSendEvent) { | |
| runCatching { | |
| outboxEventService.sendAsync(event.outboxEventId) | |
| }.onFailure { e -> | |
| // TODO: 로깅 추가 (logger.error("Failed to enqueue outbox send", e)) | |
| } | |
| } |
🤖 Prompt for AI Agents
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventSendListener.kt
around lines 16-19, OutboxEventSendListener의 send(OutboxSendEvent) 메서드가
outboxEventService.sendAsync(event.outboxEventId) 호출 중 발생하는 예외를 처리하지 않아 다른 리스너
실행을 방해할 수 있으므로, send 내부에서 try-catch로 예외를 잡아 처리하고 적절히 로깅하세요; 구체적으로
send(OutboxSendEvent) 메서드에서 outboxEventService.sendAsync 호출을 감싸고 Throwable을 캐치해
에러 로그를 남기되 예외를 무시하거나 재시도 로직을 구현할 필요가 있으면 그 정책을 적용하고, 트랜잭션
어노테이션(@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT))은 그대로
유지하세요.
src/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.kt
Outdated
Show resolved
Hide resolved
| data class OutboxPublishEvent( | ||
| val userId: String, | ||
| val eventType: SseEventType, | ||
| val payloadJson: String, | ||
| ) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Search for OutboxEventPublisher class definition
find . -name "*.kt" -type f | xargs grep -l "class OutboxEventPublisher" 2>/dev/null | head -5Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 159
🏁 Script executed:
#!/bin/bash
# Find OutboxPublishEvent instantiation points
rg -n "OutboxPublishEvent\s*\(" --type kotlin -C 3 2>/dev/null | head -100Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 1613
🏁 Script executed:
#!/bin/bash
# Check for validation annotations on OutboxPublishEvent fields
cat src/main/kotlin/org/appjam/smashing/domain/outbox/dto/OutboxPublishEvent.kt 2>/dev/nullRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 299
🏁 Script executed:
#!/bin/bash
# Search for OutboxEventPublisher implementation
rg -n "class OutboxEventPublisher" --type kotlin -A 30 2>/dev/null | head -80Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 2022
🏁 Script executed:
#!/bin/bash
# Search for all calls to OutboxEventPublisher.publish()
rg -n "\.publish\(" --type kotlin -B 5 | grep -A 5 -B 5 "OutboxEventPublisher\|publish(" | head -80Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 55
🏁 Script executed:
#!/bin/bash
# More specific search for OutboxEventPublisher usage
rg -n "outboxEventPublisher\|OutboxEventPublisher" --type kotlin | head -30Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 55
🏁 Script executed:
#!/bin/bash
# Check if userId validation happens in calling code
rg -n "publish\s*\(" src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventPublisher.kt -A 1 -B 10Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 473
🏁 Script executed:
#!/bin/bash
# Search for usages of OutboxEventPublisher in all kotlin files
rg -n "OutboxEventPublisher" --type kotlinRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 188
🏁 Script executed:
#!/bin/bash
# Search for the publish method call pattern more broadly
rg -n "publish\(" --type kotlin -B 3 -A 3 | grep -B 5 -A 5 "userId\|OutboxPublishEvent" | head -100Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 839
🏁 Script executed:
#!/bin/bash
# Check if there are any event listeners for OutboxPublishEvent
rg -n "OutboxPublishEvent\|@EventListener\|@TransactionalEventListener" --type kotlinRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 55
🏁 Script executed:
#!/bin/bash
# Search more broadly for calls to publish method with userId parameter
rg -n "\.publish\s*\(" --type kotlin -A 3 -B 5 2>/dev/null | head -150Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 55
🏁 Script executed:
#!/bin/bash
# Check the entire codebase for where OutboxEventPublisher is injected/used
rg -n "outboxEventPublisher\." --type kotlin -B 2 -A 2 2>/dev/nullRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 55
🏁 Script executed:
#!/bin/bash
# List all Kotlin files to understand project structure
find . -name "*.kt" -type f | grep -E "(outbox|publisher|event)" | head -20Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 1384
🏁 Script executed:
#!/bin/bash
# Check OutboxEventRecordListener and OutboxEventSendListener to see where publish is called
cat src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventRecordListener.ktRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 1221
🏁 Script executed:
#!/bin/bash
# Check OutboxEventSendListener
cat src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventSendListener.ktRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 789
🏁 Script executed:
#!/bin/bash
# Check the service files that might use OutboxEventPublisher
cat src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventService.ktRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 1259
🏁 Script executed:
#!/bin/bash
# Check the controller that subscribes to events
cat src/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.ktRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 1916
🏁 Script executed:
#!/bin/bash
# Search all files for calls to publisher.publish or .publish method
rg -n "\.publish\(" --type kotlin -B 3 -A 3 2>/dev/null | head -200Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 55
🏁 Script executed:
#!/bin/bash
# Check OutboxEvent entity to see if it has validation
cat src/main/kotlin/org/appjam/smashing/domain/outbox/entity/OutboxEvent.ktRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 2532
🏁 Script executed:
#!/bin/bash
# Search for where OutboxEventPublisher.publish is called
rg -n "publish\s*\(\s*" --type kotlin -B 5 -A 1 2>/dev/null | grep -v "applicationEventPublisher.publishEvent" | head -100Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 876
🏁 Script executed:
#!/bin/bash
# Check if there's any validation on userId in domain services
rg -n "userId|user.*validation|@NotBlank|@NotEmpty|require" --type kotlin src/main/kotlin/org/appjam/smashing/domain/ | head -50Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 5731
🏁 Script executed:
#!/bin/bash
# Broader search for any publish method calls in the entire codebase
find . -name "*.kt" -type f -exec grep -l "publish(" {} \; 2>/dev/null | head -20Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 159
🏁 Script executed:
#!/bin/bash
# Check all service files for OutboxEventPublisher usage
find . -path "*/service/*.kt" -type f -exec grep -l "OutboxEventPublisher\|\.publish(" {} \; 2>/dev/nullRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 55
🏁 Script executed:
#!/bin/bash
# Check the complete OutboxEventPublisher file to see if there's more context
wc -l src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventPublisher.kt && cat src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventPublisher.ktRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 1020
🏁 Script executed:
#!/bin/bash
# Search for any @Valid or validation annotations in request/controller DTOs
rg -n "@Valid|@NotBlank|@NotEmpty|@NotNull" --type kotlin -B 2 -A 2 | head -100Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 591
🏁 Script executed:
#!/bin/bash
# Check if there's validation at the API/controller level
find . -name "*Controller.kt" -o -name "*Request.kt" -o -name "*Dto.kt" | xargs grep -l "userId\|@Valid" 2>/dev/null | head -20Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 161
🏁 Script executed:
#!/bin/bash
# Look for where OutboxEventPublisher is actually used/injected
rg -n "OutboxEventPublisher" --type kotlin -A 5 -B 5Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 1381
검증 로직이 부재합니다. 방어적 검증을 추가하세요.
OutboxEventPublisher.publish() 메서드에서 검증이 수행되지 않으며, OutboxPublishEvent DTO와 OutboxEvent 엔티티 모두 검증 어노테이션이 없습니다. 현재 구현에서는 userId가 Spring Security 인증 컨텍스트에서 흘러오기 때문에 실제 위험이 낮지만, 다른 코드 경로에서 이 메서드가 호출될 경우 유효하지 않은 데이터가 저장될 수 있습니다.
OutboxPublishEvent와 OutboxEvent에 검증 어노테이션을 추가하세요 (userId에 @NotBlank, payloadJson에 @NotBlank 등).
🤖 Prompt for AI Agents
In @src/main/kotlin/org/appjam/smashing/domain/outbox/dto/OutboxPublishEvent.kt
around lines 5-9, Add defensive bean-validation: annotate the OutboxPublishEvent
data class properties with Kotlin field-targeted validation annotations (e.g.,
@field:NotBlank on userId and payloadJson, and @field:NotNull on eventType) and
mirror the same constraints on the OutboxEvent entity fields. Then ensure
OutboxEventPublisher.publish() receives a validated parameter by annotating the
method parameter with @Valid (or programmatically validate using a
javax.validation.Validator) so invalid DTOs are rejected before persistence.
Include necessary imports (javax.validation.constraints.* and
javax.validation.Valid) and adjust any mapping code to rely on validated source
objects.
| @Query(""" | ||
| select e.id | ||
| from OutboxEvent e | ||
| where e.userId = :userId | ||
| and e.status = :status | ||
| order by e.id asc | ||
| """) | ||
| fun findPendingIds( | ||
| userId: String, | ||
| status: OutboxEventStatus, | ||
| pageable: Pageable, | ||
| ): List<String> |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
메서드 이름이 실제 동작과 불일치.
메서드명이 findPendingIds이지만, 실제로는 파라미터로 전달받은 모든 status에 대해 동작합니다. status 파라미터가 PROCESSING, FAILED 등이 될 수 있으므로, 더 정확한 이름은 findIdsByUserAndStatus 또는 findEventIdsByStatus입니다.
🔎 제안하는 메서드명 변경
-fun findPendingIds(
+fun findEventIdsByUserAndStatus(
userId: String,
status: OutboxEventStatus,
pageable: Pageable,
): List<String>🤖 Prompt for AI Agents
In
@src/main/kotlin/org/appjam/smashing/domain/outbox/repository/OutboxEventRepository.kt
around lines 13-24, The method name findPendingIds in OutboxEventRepository is
misleading because the query returns IDs for any passed status value; rename the
method to a clearer name like findIdsByUserAndStatus or findEventIdsByStatus to
reflect its behavior, update all call sites to use the new name, and ensure the
function signature (userId: String, status: OutboxEventStatus, pageable:
Pageable): List<String> and @Query remain unchanged so only the identifier is
updated.
| @Modifying(clearAutomatically = true, flushAutomatically = true) | ||
| @Query(""" | ||
| update OutboxEvent e | ||
| set e.status = :processing, | ||
| e.lastAttemptAt = :now | ||
| where e.id = :id | ||
| and e.status = :pending | ||
| """) | ||
| fun markProcessingIfPending( | ||
| id: String, | ||
| pending: OutboxEventStatus = OutboxEventStatus.PENDING, | ||
| processing: OutboxEventStatus = OutboxEventStatus.PROCESSING, | ||
| now: LocalDateTime, | ||
| ): Int | ||
|
|
||
| @Modifying(clearAutomatically = true, flushAutomatically = true) | ||
| @Query(""" | ||
| update OutboxEvent e | ||
| set e.status = :pending, | ||
| e.lastAttemptAt = :now | ||
| where e.id = :id | ||
| and e.status = :processing | ||
| """) | ||
| fun markPendingIfProcessing( | ||
| id: String, | ||
| pending: OutboxEventStatus = OutboxEventStatus.PENDING, | ||
| processing: OutboxEventStatus = OutboxEventStatus.PROCESSING, | ||
| now: LocalDateTime, | ||
| ): Int | ||
|
|
||
| @Modifying(clearAutomatically = true, flushAutomatically = true) | ||
| @Query(""" | ||
| update OutboxEvent e | ||
| set e.status = :failed, | ||
| e.lastAttemptAt = :now | ||
| where e.id = :id | ||
| and e.status = :processing | ||
| """) | ||
| fun markFailedIfProcessing( | ||
| id: String, | ||
| failed: OutboxEventStatus = OutboxEventStatus.FAILED, | ||
| processing: OutboxEventStatus = OutboxEventStatus.PROCESSING, | ||
| now: LocalDateTime, | ||
| ): Int |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: OutboxEventRepository의 원자적 업데이트 메서드 호출 시 반환값 검증 여부 확인
# markProcessingIfPending, markPendingIfProcessing, markFailedIfProcessing 호출을 찾고
# 반환값이 체크되는지 확인
rg -n -A5 -B2 'mark(Processing|Pending|Failed)If(Pending|Processing)' \
--type kotlin \
-g '!**/repository/**' \
-g '!**/entity/**'Repository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 2273
🏁 Script executed:
rg -n 'markProcessingIfPending|markPendingIfProcessing|markFailedIfProcessing' \
--type kotlinRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 947
markFailedIfProcessing와 markPendingIfProcessing의 반환값 검증 필요.
26줄과 32줄에서 호출되는 markFailedIfProcessing과 markPendingIfProcessing이 22줄의 markProcessingIfPending과 달리 반환값을 검증하지 않습니다. 동시성 충돌로 인한 상태 변경 실패를 감지할 수 없으므로, 다음과 같이 모든 원자적 업데이트 메서드의 반환값을 일관되게 검증해야 합니다:
// 26줄
if (outboxEventRepository.markFailedIfProcessing(outboxEventId, now = now) == 0) {
log.warn("Failed to mark as failed: $outboxEventId")
}
// 32줄
if (outboxEventRepository.markPendingIfProcessing(outboxEventId, now = now) == 0) {
log.warn("Failed to recover to pending: $outboxEventId")
}
src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventSender.kt
Outdated
Show resolved
Hide resolved
src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventService.kt
Outdated
Show resolved
Hide resolved
src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventService.kt
Show resolved
Hide resolved
leeeyubin
left a comment
leeeyubin
left a comment
There was a problem hiding this comment.
SSE 구현 방식을 처음 봐서 덕분에 공부하면서 코드 봤던 것 같아요!!
제가 감히 이 피알을 봐도 되는 걸까요.....
어려웠을 텐데 구현하시느라 수고하셨어요!!!👍
| for (emitter in queue) { | ||
| val isSucceed = runCatching { emitter.send(event) }.isSuccess | ||
| if (!isSucceed) remove(userId, emitter) | ||
| } | ||
| } |
There was a problem hiding this comment.
| for (emitter in queue) { | |
| val isSucceed = runCatching { emitter.send(event) }.isSuccess | |
| if (!isSucceed) remove(userId, emitter) | |
| } | |
| } | |
| for (emitter in queue) { | |
| runCatching { emitter.send(event) } | |
| .onFailure { remove(userId, emitter) } | |
| } |
별도로 isSucceed의 변수를 만들지 않고 onFailure를 사용하면 runCatching의 특징을 더 살릴 수 있을 것 같아요!
There was a problem hiding this comment.
오호 이런 방법이 있었군요! 적용 완료했습니다 감사합니다
| val cleanup = { remove(userId, emitter) } | ||
|
|
||
| emitter.onCompletion(cleanup) | ||
| emitter.onTimeout(cleanup) | ||
| emitter.onError { cleanup() } |
There was a problem hiding this comment.
| val cleanup = { remove(userId, emitter) } | |
| emitter.onCompletion(cleanup) | |
| emitter.onTimeout(cleanup) | |
| emitter.onError { cleanup() } | |
| emitter.apply { | |
| val cleanup = { remove(userId, this) } | |
| onCompletion(cleanup) | |
| onTimeout(cleanup) | |
| onError { cleanup() } | |
| } |
apply 스코프 함수를 사용하면 emitter 부분의 중복되는 부분을 제거할 수 있을 것 같습니당
There was a problem hiding this comment.
오호 적용 완료했습니다! 좋은 인사이트 감사합니다
| // 연결이 정상인 경우 backlog flush | ||
| if (isConnected) { | ||
| outboxEventService.enqueuePendingSends(userId) | ||
| } |
There was a problem hiding this comment.
이 부분도 isConnected 변수를 만들기 보다는 runCatching의 onSuccess 블록 안에 넣는 다면 코드가 깔끔해질 것 같습니다!
There was a problem hiding this comment.
오호 적용 완료했습니다! 감사합니다
| companion object { | ||
| const val OUTBOX_EVENT_EXECUTOR = "outboxEventExecutor" | ||
| } |
There was a problem hiding this comment.
매우 사소.. ) 다른 데서 안 쓰이면 private 붙여도 될 것 가타용
| val success = runCatching { | ||
| sseEmitterRegistry.send(event.userId, event.eventType.eventName, event.payload) | ||
| }.isSuccess | ||
|
|
||
| if (success) event.markAsSent() else event.markAsFailed() |
There was a problem hiding this comment.
여기두 onSuccess, onFailure 사용하면 좋을 것 같습니다!
| fun currentUserId(): String { | ||
| val authentication = SecurityContextHolder.getContext().authentication | ||
| ?: throw AuthenticationCredentialsNotFoundException("Unauthenticated") | ||
|
|
||
| if (!authentication.isAuthenticated) { | ||
| throw InsufficientAuthenticationException("Unauthenticated") | ||
| } |
There was a problem hiding this comment.
이 부분은 SpringSecurity에 저장되어있는 값을 가져오는 거 맞죠?
CustomException으로 예외처리를 통일해주는 건 어떤가요..?!
There was a problem hiding this comment.
해당 부분이 보안 관련 예외라 custom 예외 대신 보안 관련 예외로 내려 Security 설정에서 EntryPoint에서 통일 응답 내는 방식이 더 보안 컨벤션에 적합하지 않을까, 라는 고민이 들어 별도의 예외로 내리도록 구현해두었습니다..!
| @Component | ||
| @Profile("!test") |
There was a problem hiding this comment.
와우 테스트코드에서는 동작하지 않도록 해서 실제 알림을 날리지 않기 위함으로 이해했는데 맞을까요? 꼼꼼하네유..
There was a problem hiding this comment.
맞습니다! 꼼꼼히 봐주셔서 정말 감사합니다
| val executor = ThreadPoolTaskExecutor() | ||
|
|
||
| executor.corePoolSize = 4 | ||
| executor.maxPoolSize = 8 |
There was a problem hiding this comment.
ThreadPoolTaskExecutor에 대해 덕분에 알아가는 것 같아요!
corePoolSize가 최소 스레드 수라고 이해했는데 혹시 4개로 지정한 이유가 있으신가용?
There was a problem hiding this comment.
API 스레드랑 분리된 풀에서 최소 동시 처리량 확보를 위해 core를 4로 두었습니다! 추후에 모니터링을 통해 확장을 하는 방향으로 바꾸게 될 것같습니다
|
|
||
| private val emitters = ConcurrentHashMap<String, ConcurrentLinkedQueue<SseEmitter>>() | ||
|
|
There was a problem hiding this comment.
ConcurrentHashMap이랑 ConcurrentLinkedQueue 사용하신 거 너무 좋네요.. 👍👍
| @Transactional(propagation = Propagation.REQUIRES_NEW) | ||
| fun send(outboxEventId: String) { | ||
| val now = LocalDateTime.now() | ||
|
|
||
| // 선점: PENDING -> PROCESSING | ||
| if (outboxEventRepository.markProcessingIfPending(outboxEventId, now = now) != 1) return | ||
|
|
There was a problem hiding this comment.
구현해주신 부분 정말 흥미롭게 읽은 것 같아요!!
낙관적 락의 흐름이라고 보았는데 맞을까요....?
Version을 사용한 관리 방식도 있을 것 같은데 직접 Status를 통해 구현하신 이유가 궁금해요!
There was a problem hiding this comment.
좋은 질문 감사해요! 😊
말씀 주신 것처럼 개념적으로는 낙관적 락의 흐름과 유사한 접근이 맞습니다.
다만 @Version 기반 낙관적 락을 사용하지 않고, 상태 전이(PENDING → PROCESSING)를 조건으로 둔 update 쿼리를 사용한 몇가지 이유가 있습니다.
우선 Outbox 특성상 선점 성공/실패만 명확하면 충분한 케이스라서 엔티티 전체의 버전 충돌까지 감지할 필요는 없다고 판단했습니다.
→ update ... where status = PENDING 자체로 원자적 선점 역할을 할 수 있도록 하였습니다.
@Version을 사용할 경우, 엔티티를 반드시 조회 → 수정 → flush 해야 하고 충돌 시 예외 기반 흐름(OptimisticLockException)을 처리해야 해서 재시도/무시가 빈번한 Outbox 처리에는 다소 무거운 방식이라고 느꼈습니다.
반면 현재 방식은 DB 레벨에서 한 줄 update로 선점 여부가 결정되고 결과 row count(1 / 0)로 분기할 수 있고, 동시성 처리 흐름이 명시적이고 제어하기 쉽다는 장점이 있습니다.
그래서 해당 부분을 Outbox에 맞춰 상태 기반 선점에 가까운 구현으로 진행하였습니다
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (4)
src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventSender.kt (1)
45-45: 스케줄러 도입 계획을 추적하세요.TODO 코멘트에서 언급된
PROCESSING상태로 고착된 이벤트 처리는 PR 목표에서 향후 작업으로 명시되어 있습니다.이 작업을 위한 별도 이슈를 생성하거나 스케줄러 구현 코드를 제안해 드릴까요?
src/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.kt (1)
23-47: 운영 환경 고려사항을 검토하세요.현재 구현은 기능적으로 올바르나, 프로덕션 배포 시 다음 사항들을 고려해보세요:
- 속도 제한: 사용자당 SSE 구독 요청 빈도 제한 미설정
- 연결 수 제한: 사용자당 동시 연결 수 제한 없음 (멀티탭 지원 의도이나, 남용 방지 필요)
- 타임아웃: 30분 타임아웃이 의도적이라면 적절하나, 재연결 로직과 함께 문서화 권장
이러한 보호 장치를 추가하는 구현을 제안해드릴까요?
src/main/kotlin/org/appjam/smashing/domain/outbox/components/SseEmitterRegistry.kt (1)
91-96:peek() != null대신isNotEmpty()를 사용하면 의도가 더 명확합니다.
peek()은 큐의 첫 번째 요소를 확인하는 메서드이지만, 여기서는 큐가 비어있는지 확인하는 것이 목적입니다.isNotEmpty()를 사용하면 코드의 의도가 더 명확하게 드러납니다.🔎 제안하는 리팩토링
private fun remove(userId: String, emitter: SseEmitter) { emitters.computeIfPresent(userId) { _, queue -> queue.remove(emitter) - queue.takeIf { it.peek() != null } + queue.takeIf { it.isNotEmpty() } } }src/main/kotlin/org/appjam/smashing/domain/outbox/repository/OutboxEventRepository.kt (1)
34-39: 기본 매개변수로 enum 값을 지정하는 것은 불필요해 보입니다.
pending,processing,failed매개변수에 기본값을 지정하는 것은 다소 불필요합니다. 호출자가 명시적으로 enum 값을 전달하는 것이 의도를 더 명확하게 드러내며, 실수로 잘못된 상태 전환을 하는 것을 방지할 수 있습니다. 특히 원자적 상태 전환에서는 명시성이 중요합니다.🔎 제안하는 리팩토링
fun markProcessingIfPending( id: String, - pending: OutboxEventStatus = OutboxEventStatus.PENDING, - processing: OutboxEventStatus = OutboxEventStatus.PROCESSING, + pending: OutboxEventStatus, + processing: OutboxEventStatus, now: LocalDateTime, ): Int fun markPendingIfProcessing( id: String, - pending: OutboxEventStatus = OutboxEventStatus.PENDING, - processing: OutboxEventStatus = OutboxEventStatus.PROCESSING, + pending: OutboxEventStatus, + processing: OutboxEventStatus, now: LocalDateTime, ): Int fun markFailedIfProcessing( id: String, - failed: OutboxEventStatus = OutboxEventStatus.FAILED, - processing: OutboxEventStatus = OutboxEventStatus.PROCESSING, + failed: OutboxEventStatus, + processing: OutboxEventStatus, now: LocalDateTime, ): Int호출 시점에 명시적으로 상태 값을 전달하도록 변경하면 코드의 의도가 더 명확해집니다.
Also applies to: 49-54, 64-69
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventPublisher.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventRecordListener.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/components/SseEmitterRegistry.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/repository/OutboxEventRepository.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventSender.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventService.kt
🔇 Additional comments (8)
src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventRecordListener.kt (1)
19-41: 트랜잭션 생명주기 처리가 올바르게 구현되었습니다.과거 리뷰에서
BEFORE_COMMIT페이즈 사용에 대한 우려가 있었으나, 현재 구현은 다음과 같이 올바르게 처리하고 있습니다:
BEFORE_COMMIT페이즈에서OutboxEvent를 저장하여 이벤트 영속성 보장TransactionSynchronization.afterCommit()콜백을 통해 실제 전송은 커밋 후로 지연- 이를 통해 전송 실패가 비즈니스 트랜잭션을 롤백시키지 않음
이 패턴은 이벤트의 원자적 저장과 안전한 비동기 전송을 모두 보장하는 유효한 접근 방식입니다.
src/main/kotlin/org/appjam/smashing/domain/outbox/components/OutboxEventPublisher.kt (1)
17-32: 에러 핸들링이 포괄적으로 구현되었습니다.과거 리뷰에서 요청된 JSON 직렬화 예외 처리가 완벽하게 적용되었습니다:
runCatching을 통한 안전한 직렬화 처리- 실패 시 상세한 컨텍스트 로깅 (userId, eventType, payload 클래스명)
- 빈 userId 검증으로 잘못된 이벤트 발행 방지
이벤트 발행이 주 비즈니스 로직을 중단시키지 않으면서도 디버깅에 필요한 정보를 충분히 기록하고 있습니다.
src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventSender.kt (1)
18-66: 상태 기반 낙관적 잠금과 에러 핸들링이 우수합니다.과거 리뷰의 로깅 및 에러 처리 요청이 모두 반영되었습니다:
- 원자적 상태 전환(
markProcessingIfPending)으로 중복 전송 방지- 각 실패 지점에서 상세한 컨텍스트 로깅 (outboxEventId, userId, eventType 포함)
runCatching을 통한 안전한 전송 및 저장 처리- 연결 부재 시
PENDING으로 복구하여 재시도 가능하도록 구현상태(
OutboxEventStatus)를 통한 동시성 제어 방식은 비즈니스 로직과 잘 통합되어 있으며, 별도의Version필드 없이도 효과적으로 동작합니다.src/main/kotlin/org/appjam/smashing/domain/outbox/service/OutboxEventService.kt (1)
19-55: 비동기 실행과 에러 핸들링이 포괄적으로 구현되었습니다.과거 리뷰에서 요청된 모든 에러 처리가 완벽하게 적용되었습니다:
enqueuePendingSendsAsync메서드:
- Executor 거부 처리:
RejectedExecutionException캐치 및 로깅 (lines 36-38)- DB 조회 실패: 외부
runCatching으로 처리 (lines 24-34)- 개별 전송 실패: 내부
runCatching으로 각 ID 별 실패 로깅 (lines 28-31)
sendAsync메서드:
- 동일한 계층적 에러 처리 패턴 적용 (lines 44-54)
- 모든 실패 지점에서 컨텍스트 정보 포함
중첩된 에러 핸들링 구조가 Kotlin의 관용적 패턴을 잘 따르고 있으며, 운영 환경에서의 추적성을 보장합니다.
src/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.kt (1)
23-47: SSE 구독 플로우가 잘 설계되었습니다.과거 리뷰의 동기적 DB 조회 우려가 해결되었습니다:
enqueuePendingSendsAsync가 이제 executor를 통해 DB 조회를 별도 스레드로 오프로드- HTTP 요청 스레드는 즉시 emitter를 반환하여 블로킹 없음
system.connected이벤트를 통한 핸드셰이크 후 백로그 플러시는 좋은 패턴현재 코드의
runCatching사용도 적절합니다:
onFailure로 연결 실패 시 정리 수행isSuccess로 조건부 백로그 플러시 - 명확하고 읽기 쉬움src/main/kotlin/org/appjam/smashing/domain/outbox/components/SseEmitterRegistry.kt (1)
1-27: 전체적인 설계와 동시성 처리가 우수합니다!
ConcurrentHashMap과ConcurrentLinkedQueue를 사용한 스레드 안전한 구현이 매우 적절합니다.add메서드의computeIfAbsent사용과 자동 생명주기 바인딩도 잘 구현되어 있습니다.src/main/kotlin/org/appjam/smashing/domain/outbox/repository/OutboxEventRepository.kt (2)
13-24: 메서드 이름 개선이 잘 반영되었습니다!과거 리뷰에서 지적된 명명 문제가
findIdsByUserIdAndStatus로 개선되어 메서드의 동작이 명확하게 표현되고 있습니다. 쿼리 로직도 적절하며 페이징 처리가 올바르게 구현되어 있습니다.
26-39: 원자적 업데이트 패턴이 잘 구현되었습니다!조건부 업데이트 쿼리들이
@Modifying과 함께clearAutomatically,flushAutomatically옵션을 올바르게 사용하고 있으며,Int반환 타입을 통해 호출자가 업데이트 성공 여부를 확인할 수 있도록 설계되어 있습니다. 이는 동시성 환경에서 상태 전환을 안전하게 관리하는 좋은 패턴입니다.Also applies to: 41-54, 56-69
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI Agents
In @src/main/kotlin/org/appjam/smashing/domain/outbox/dto/SsePayload.kt:
- Around line 15-18: The data classes (e.g., MatchingReceivedPayload) expose the
`type` as a constructor parameter with a default, allowing callers to override
it; change each such class to remove `type` from the constructor and instead
declare it as an immutable class-body property (e.g., `override val type =
SseEventType.MATCHING_RECEIVED.eventName`) so callers cannot supply a custom
value; apply this same change to the other affected payload data classes
referenced in the comment to ensure `type` is fixed and not overridable.
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
src/main/kotlin/org/appjam/smashing/domain/notification/enums/NotificationType.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/dto/SsePayload.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/enums/MatchingUpdateStatus.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/enums/SseEventType.kt
🔇 Additional comments (3)
src/main/kotlin/org/appjam/smashing/domain/outbox/enums/MatchingUpdateStatus.kt (1)
3-7: 매칭 상태 enum 정의가 명확합니다.ACCEPTED, REJECTED, CANCELLED 세 가지 상태가 매칭 업데이트 흐름을 잘 표현하고 있습니다.
src/main/kotlin/org/appjam/smashing/domain/notification/enums/NotificationType.kt (1)
8-10: 알림 타입 추가가 적절합니다.MATCHING_RESULT_SUBMITTED가 매칭 관련 알림과 결과 관련 알림 사이에 논리적으로 배치되어 있습니다.
src/main/kotlin/org/appjam/smashing/domain/outbox/enums/SseEventType.kt (1)
3-14: SSE 이벤트 타입 enum이 잘 설계되었습니다.eventName 프로퍼티를 통해 타입 안전성을 확보하면서도 문자열 이벤트명을 제공하는 접근이 적절합니다. 이벤트명의 dot notation 규칙도 일관되게 적용되었습니다.
| data class MatchingReceivedPayload( | ||
| override val type: String = SseEventType.MATCHING_RECEIVED.eventName, | ||
| val matchingId: String, | ||
| ) : SsePayload |
There was a problem hiding this comment.
type 필드가 생성자에서 재정의 가능한 문제가 있습니다.
현재 각 data class에서 type을 생성자 파라미터로 default 값과 함께 정의하고 있어, 호출자가 의도치 않게 또는 실수로 잘못된 타입 값을 설정할 수 있습니다:
// 잘못된 타입 설정이 가능
MatchingReceivedPayload(type = "wrong.type", matchingId = "123")이는 이벤트 라우팅이나 역직렬화 시 타입 불일치 문제를 야기할 수 있습니다.
🔎 type 필드를 불변 속성으로 변경하는 제안
생성자 파라미터가 아닌 클래스 본문의 프로퍼티로 정의하여 재정의를 방지하세요:
data class MatchingReceivedPayload(
- override val type: String = SseEventType.MATCHING_RECEIVED.eventName,
val matchingId: String,
-) : SsePayload
+) : SsePayload {
+ override val type: String = SseEventType.MATCHING_RECEIVED.eventName
+}
data class MatchingUpdatedPayload(
- override val type: String = SseEventType.MATCHING_UPDATED.eventName,
val matchingId: String,
val status: MatchingUpdateStatus,
-) : SsePayload
+) : SsePayload {
+ override val type: String = SseEventType.MATCHING_UPDATED.eventName
+}
data class NotificationCreatedPayload(
- override val type: String = SseEventType.NOTIFICATION_CREATED.eventName,
val notificationId: String,
val notificationType: NotificationType,
val targetId: String? = null,
-) : SsePayload
+) : SsePayload {
+ override val type: String = SseEventType.NOTIFICATION_CREATED.eventName
+}이렇게 하면 type이 생성자에서 제외되어 재정의가 불가능해집니다.
Also applies to: 25-29, 35-40
🤖 Prompt for AI Agents
In @src/main/kotlin/org/appjam/smashing/domain/outbox/dto/SsePayload.kt around
lines 15 - 18, The data classes (e.g., MatchingReceivedPayload) expose the
`type` as a constructor parameter with a default, allowing callers to override
it; change each such class to remove `type` from the constructor and instead
declare it as an immutable class-body property (e.g., `override val type =
SseEventType.MATCHING_RECEIVED.eventName`) so callers cannot supply a custom
value; apply this same change to the other affected payload data classes
referenced in the comment to ensure `type` is fixed and not overridable.
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI Agents
In
@src/main/kotlin/org/appjam/smashing/global/auth/security/data/CustomUserDetails.kt:
- Line 1: Update the import in JwtProvider.kt: replace the old import
"org.appjam.smashing.global.auth.security.CustomUserDetails" with the new
package "org.appjam.smashing.global.auth.security.data.CustomUserDetails" so
JwtProvider.kt references the moved CustomUserDetails class; ensure the import
line in JwtProvider.kt matches the updated package used by
CurrentUserProvider.kt to fix the compile error.
In @src/main/kotlin/org/appjam/smashing/global/config/AsyncConfig.kt:
- Around line 11-23: The outboxEventExecutor bean (function outboxEventExecutor)
currently relies on the default AbortPolicy when the queue (queueCapacity 500)
is full and all threads (maxPoolSize 8) are busy; explicitly set a rejected
execution policy on the ThreadPoolTaskExecutor to avoid
RejectedExecutionException — e.g., call setRejectedExecutionHandler(new
ThreadPoolExecutor.CallerRunsPolicy()) on the executor or provide a custom
RejectedExecutionHandler that logs the rejection and enqueues/retries the task;
ensure the handler is set before executor.initialize() so the policy is active
at runtime.
🧹 Nitpick comments (2)
src/main/kotlin/org/appjam/smashing/global/auth/security/components/CurrentUserProvider.kt (1)
12-18: 예외 메시지의 명확성을 개선하면 디버깅에 도움이 됩니다.Line 14와 Line 17에서 서로 다른 예외 타입(
AuthenticationCredentialsNotFoundExceptionvsInsufficientAuthenticationException)을 던지고 있지만, 동일한 메시지 "Unauthenticated"를 사용하고 있습니다.이 두 상황은 의미적으로 다른 오류 상태입니다:
- Line 14: Authentication 객체가 null인 경우
- Line 17: Authentication 객체는 존재하지만 인증되지 않은 경우
메시지를 구체적으로 구분하면 로그 분석 시 문제를 더 빠르게 파악할 수 있습니다.
🔎 제안하는 개선 사항
fun currentUserId(): String { val authentication = SecurityContextHolder.getContext().authentication - ?: throw AuthenticationCredentialsNotFoundException("Unauthenticated") + ?: throw AuthenticationCredentialsNotFoundException("Authentication not found in SecurityContext") if (!authentication.isAuthenticated) { - throw InsufficientAuthenticationException("Unauthenticated") + throw InsufficientAuthenticationException("User is not authenticated") }src/main/kotlin/org/appjam/smashing/global/config/AsyncConfig.kt (1)
15-18: 스레드 풀 파라미터를 설정 파일로 외부화하는 것을 고려해보세요.현재 하드코딩된 값들(corePoolSize, maxPoolSize, queueCapacity, keepAliveSeconds)을
application.yml로 외부화하면 재배포 없이 운영 환경에서 튜닝이 가능합니다.과거 리뷰 코멘트에서 모니터링 기반 조정 계획이 언급되었으므로, 설정 외부화를 통해 더 유연하게 대응할 수 있습니다.
🔎 설정 외부화 예시
application.yml:async: outbox: core-pool-size: 4 max-pool-size: 8 queue-capacity: 500 keep-alive-seconds: 30코드:
+@ConfigurationProperties(prefix = "async.outbox") +data class OutboxExecutorProperties( + val corePoolSize: Int = 4, + val maxPoolSize: Int = 8, + val queueCapacity: Int = 500, + val keepAliveSeconds: Int = 30 +) @Configuration -class AsyncConfig { +class AsyncConfig( + private val properties: OutboxExecutorProperties +) { @Bean(name = [OUTBOX_EVENT_EXECUTOR]) fun outboxEventExecutor(): Executor { val executor = ThreadPoolTaskExecutor() - executor.corePoolSize = 4 - executor.maxPoolSize = 8 - executor.queueCapacity = 500 - executor.keepAliveSeconds = 30 + executor.corePoolSize = properties.corePoolSize + executor.maxPoolSize = properties.maxPoolSize + executor.queueCapacity = properties.queueCapacity + executor.keepAliveSeconds = properties.keepAliveSeconds executor.setThreadNamePrefix("Outbox-Sse-") executor.initialize() return executor }
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
src/main/kotlin/org/appjam/smashing/domain/outbox/components/SseEmitterRegistry.ktsrc/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.ktsrc/main/kotlin/org/appjam/smashing/global/auth/security/components/CurrentUserProvider.ktsrc/main/kotlin/org/appjam/smashing/global/auth/security/data/CustomUserDetails.ktsrc/main/kotlin/org/appjam/smashing/global/config/AsyncConfig.kt
🚧 Files skipped from review as they are similar to previous changes (2)
- src/main/kotlin/org/appjam/smashing/domain/outbox/components/SseEmitterRegistry.kt
- src/main/kotlin/org/appjam/smashing/domain/outbox/controller/SseSubscribeController.kt
🔇 Additional comments (1)
src/main/kotlin/org/appjam/smashing/global/auth/security/components/CurrentUserProvider.kt (1)
20-25: LGTM! Principal 타입 체크와 추출 로직이 안전합니다.
when표현식을 사용한 타입 체크가 적절하며, 지원하지 않는 principal 타입에 대해 클래스명을 포함한 명확한 예외 메시지를 제공하고 있습니다.참고: 이 구현은
CustomUserDetails.username이 사용자 ID를 포함한다고 가정합니다. SSE 구독 등에서 userId로 사용되므로, username이 실제로 고유한 사용자 식별자임을 확인해 주세요.
| @@ -1,4 +1,4 @@ | |||
| package org.appjam.smashing.global.auth.security | |||
| package org.appjam.smashing.global.auth.security.data | |||
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check for old package path references to CustomUserDetails
echo "=== Checking for old package path references ==="
rg "org\.appjam\.smashing\.global\.auth\.security\.CustomUserDetails" --type kotlin 2>/dev/null || echo "No old package path references found"
# Check for imports of CustomUserDetails
echo -e "\n=== All imports of CustomUserDetails ==="
rg "import.*CustomUserDetails" --type kotlin 2>/dev/null || echo "No imports found"
# Check for new package path usage
echo -e "\n=== Checking new package path usage ==="
rg "org\.appjam\.smashing\.global\.auth\.security\.data\.CustomUserDetails" --type kotlin 2>/dev/null || echo "No new package path found"
# Check CurrentUserProvider references
echo -e "\n=== CurrentUserProvider references ==="
rg "CurrentUserProvider" --type kotlin -B 1 -A 1 2>/dev/null | head -50 || echo "No CurrentUserProvider references found"
# Check if CurrentUserProvider file exists in new location
echo -e "\n=== Locating CurrentUserProvider file ==="
fd "CurrentUserProvider" --type f 2>/dev/nullRepository: TEAM-SMASHING/SMASHING-SERVER
Length of output: 2152
JwtProvider.kt의 import 문을 새 패키지 경로로 업데이트하세요.
패키지 이동 후 src/main/kotlin/org/appjam/smashing/global/auth/jwt/components/JwtProvider.kt에서 여전히 이전 패키지 경로로 CustomUserDetails를 import하고 있습니다:
import org.appjam.smashing.global.auth.security.CustomUserDetails
다음과 같이 수정하세요:
import org.appjam.smashing.global.auth.security.data.CustomUserDetails
현재 import 문은 컴파일 오류를 발생시킵니다. CurrentUserProvider.kt는 올바르게 업데이트되었으나, JwtProvider.kt가 누락되었습니다.
🤖 Prompt for AI Agents
In
@src/main/kotlin/org/appjam/smashing/global/auth/security/data/CustomUserDetails.kt
at line 1, Update the import in JwtProvider.kt: replace the old import
"org.appjam.smashing.global.auth.security.CustomUserDetails" with the new
package "org.appjam.smashing.global.auth.security.data.CustomUserDetails" so
JwtProvider.kt references the moved CustomUserDetails class; ensure the import
line in JwtProvider.kt matches the updated package used by
CurrentUserProvider.kt to fix the compile error.
| @Bean(name = [OUTBOX_EVENT_EXECUTOR]) | ||
| fun outboxEventExecutor(): Executor { | ||
| val executor = ThreadPoolTaskExecutor() | ||
|
|
||
| executor.corePoolSize = 4 | ||
| executor.maxPoolSize = 8 | ||
| executor.queueCapacity = 500 | ||
| executor.keepAliveSeconds = 30 | ||
| executor.setThreadNamePrefix("Outbox-Sse-") | ||
| executor.initialize() | ||
|
|
||
| return executor | ||
| } |
There was a problem hiding this comment.
Rejection policy를 명시적으로 설정하세요.
큐가 가득 찼을 때(500개) 모든 스레드가 사용 중이면(8개) 새로운 태스크가 거부됩니다. 기본 정책인 AbortPolicy는 RejectedExecutionException을 발생시켜 Outbox 이벤트 전송 시도가 실패할 수 있습니다.
Outbox 패턴의 신뢰성 보장을 위해 명시적인 rejection policy 설정을 권장합니다. 예: CallerRunsPolicy (호출 스레드에서 실행) 또는 커스텀 핸들러(로깅 + 재시도 로직).
🔎 Rejection policy 설정 예시
executor.corePoolSize = 4
executor.maxPoolSize = 8
executor.queueCapacity = 500
executor.keepAliveSeconds = 30
+executor.setRejectedExecutionHandler(ThreadPoolExecutor.CallerRunsPolicy())
executor.setThreadNamePrefix("Outbox-Sse-")
executor.initialize()또는 커스텀 핸들러로 로깅과 함께 적절한 처리:
+executor.setRejectedExecutionHandler { runnable, executor ->
+ logger.warn("Outbox event task rejected. Queue full or executor shutdown. Task: $runnable")
+ // 필요시 추가 처리 로직 (예: 메트릭 기록, 알림 등)
+}🤖 Prompt for AI Agents
In @src/main/kotlin/org/appjam/smashing/global/config/AsyncConfig.kt around
lines 11 - 23, The outboxEventExecutor bean (function outboxEventExecutor)
currently relies on the default AbortPolicy when the queue (queueCapacity 500)
is full and all threads (maxPoolSize 8) are busy; explicitly set a rejected
execution policy on the ThreadPoolTaskExecutor to avoid
RejectedExecutionException — e.g., call setRejectedExecutionHandler(new
ThreadPoolExecutor.CallerRunsPolicy()) on the executor or provide a custom
RejectedExecutionHandler that logs the rejection and enqueues/retries the task;
ensure the handler is set before executor.initialize() so the policy is active
at runtime.
📌 Related Issue
#️⃣ 요약 설명
SSE 관련 기능을 구현합니다.
실시간 SSE(Server-Sent Events) 기반 알림/매칭 이벤트 전송 구조를 도입하기 위한 구현으로, 단순한 즉시 푸시 방식이 아닌, Outbox 패턴을 기반으로 이벤트 유실을 방지하고 재시도가 가능한 구조를 목표로 설계했습니다.
전송 흐름 요약
-> 전송 실패 시 retryCount 기반 재시도 정책을 통해 일시적 실패 허용
-> SSE 구독 직후 system.connected 이벤트를 1회 전송하여 실제 연결 성공 여부 확인 후에만 backlog(PENDING 이벤트) flush
실시간 이벤트는 네트워크 상태, 앱 백그라운드 전환, 탭 종료 등으로 인해 전송 실패가 빈번하게 발생할 수 있습니다. 이를 단순 실패로 처리할 경우 UX 관점에서 중요한 매칭/알림 이벤트가 누락될 수 있다고 판단하여, 이벤트를 DB에 먼저 기록하고(Outbox), 트랜잭션 커밋 이후 비동기로 전송을 시도하는 구조를 채택했습니다.
SSE 전송 방식은 WebSocket이 아닌 SseEmitter 기반의 단방향 스트리밍 방식을 사용했습니다. 현재 서비스 요구사항이 “서버 → 클라이언트로의 상태 알림 전달”에 집중되어 있고, 연결 관리 및 인증 흐름을 단순화할 수 있다는 점에서 SSE가 더 적합하다고 판단했습니다. 유저의 멀티 탭 접속을 고려해 userId 단위로 여러 emitter를 관리하며, 전송 실패한 emitter는 즉시 제거하도록 설계했습니다.
Outbox 이벤트 전송은 트랜잭션 라이프사이클에 맞춰 처리됩니다. 도메인 이벤트 발생 시 OutboxEvent를 먼저 저장하고, 커밋 이후에만 실제 SSE 전송을 시도함으로써 DB 상태와 클라이언트에 전달되는 이벤트 간의 불일치를 최소화했습니다. SSE 연결이 존재하지 않는 경우에는 이벤트를 PENDING 상태로 유지하여, 유저가 다시 접속했을 때 누락된 이벤트를 재전송할 수 있도록 했습니다.
전송 실패에 대해서는 retryCount 기반의 정책을 두어 일시적인 네트워크 오류를 허용하고, 반복 실패 시에는 FAILED 상태로 전환하여 무한 재시도를 방지했습니다. 이를 통해 실시간성과 안정성 사이의 균형을 맞추는 방향으로 설계하고자 하였습니다.
비동기 처리는 @async 대신 Executor 기반으로 명시적으로 관리했습니다. 이를 통해 실제 어떤 스레드 풀에서 작업이 실행되는지 명확히 드러나도록 하고, SSE 전송과 관련된 작업들을 하나의 실행 정책으로 통제할 수 있도록 구성했습니다.
현재 구현은 매칭 관리(받은 요청, 보낸 요청, 매칭 확정)와 알림 생성 시점에 대한 실시간 반영을 중심으로 구성되어 있으며, SSE 이벤트 타입이나 정책은 추후 서비스 요구사항에 따라 확장·조정될 수 있습니다.
📝 작업 내용
👍 동작 확인
💬 리뷰 요구사항(선택)
코드 중 추후 수정이 필요할 수 있는 부분들은 모두 TODO: 주석으로 달아두었습니다. 이후 intellij에서 TODO 주석이 된 것들을 모아서 보실 수 있습니다!
현재 processing 상태로 변경 이후 네트워크 오류 등의 이유로 실패하는 경우엔 영영 processing에서 처리가 되지 않는 문제가 발생할 수 있습니다. 추후 스케쥴러 등의 도입을 통해 해당 문제를 해결할 예정입니다 (TODO 주석 달기 완료)