From ec082fcd37ea86422fdb2a06a3f65d14fd2c9694 Mon Sep 17 00:00:00 2001 From: Hexeong <123macanic@naver.com> Date: Mon, 2 Nov 2026 19:20:33 +0900 Subject: [PATCH 1/8] =?UTF-8?q?refactor:=20=EC=B1=84=ED=8C=85=20=EB=B0=A9?= =?UTF-8?q?=EC=86=A1=20Outbox=20=EC=9E=AC=EC=8B=9C=EB=8F=84=20=EB=8F=84?= =?UTF-8?q?=EC=9E=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../config/ChatMessageOutboxProperties.java | 20 ++ .../chat/domain/ChatMessageOutbox.java | 98 ++++++ .../ChatMessageOutboxCompletionType.java | 7 + .../chat/domain/ChatMessageOutboxStatus.java | 9 + .../chat/dto/ClaimedChatMessageOutbox.java | 10 + .../chat/event/ChatMessageBroadcastEvent.java | 6 +- .../ChatMessageBroadcastEventListener.java | 16 +- .../ChatMessageOutboxRepository.java | 173 ++++++++++ .../repository/ChatParticipantRepository.java | 2 + .../service/ChatMessageOutboxDispatcher.java | 84 +++++ .../service/ChatMessageOutboxScheduler.java | 28 ++ .../ChatMessageOutboxTransactionService.java | 178 ++++++++++ .../chat/service/ChatService.java | 75 +++-- .../resources/config/application-variable.yml | 11 + .../V64__create_chat_message_outbox.sql | 24 ++ ...ChatMessageBroadcastEventListenerTest.java | 75 +---- .../ChatMessageOutboxDispatcherTest.java | 123 +++++++ .../ChatMessageOutboxSchedulerTest.java | 46 +++ ...atMessageOutboxTransactionServiceTest.java | 303 ++++++++++++++++++ .../chat/service/ChatServiceTest.java | 73 ++++- src/test/resources/application.yml | 9 + 21 files changed, 1259 insertions(+), 111 deletions(-) create mode 100644 src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java create mode 100644 src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java create mode 100644 src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutboxCompletionType.java create mode 100644 src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutboxStatus.java create mode 100644 src/main/java/com/example/solidconnection/chat/dto/ClaimedChatMessageOutbox.java create mode 100644 src/main/java/com/example/solidconnection/chat/repository/ChatMessageOutboxRepository.java create mode 100644 src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java create mode 100644 src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxScheduler.java create mode 100644 src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionService.java create mode 100644 src/main/resources/db/migration/V64__create_chat_message_outbox.sql create mode 100644 src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java create mode 100644 src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxSchedulerTest.java create mode 100644 src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java diff --git a/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java b/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java new file mode 100644 index 000000000..5b8a3cef3 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java @@ -0,0 +1,20 @@ +package com.example.solidconnection.chat.config; + +import java.time.Duration; +import lombok.Getter; +import lombok.Setter; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +@Component +@ConfigurationProperties(prefix = "chat.outbox") +@Getter +@Setter +public class ChatMessageOutboxProperties { + + private int maxAttempts = 5; + private int batchSize = 100; + private Duration initialRetryDelay = Duration.ofSeconds(1); + private Duration maxRetryDelay = Duration.ofMinutes(1); + private Duration processingTimeout = Duration.ofMinutes(1); +} diff --git a/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java new file mode 100644 index 000000000..51237f55e --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java @@ -0,0 +1,98 @@ +package com.example.solidconnection.chat.domain; + +import com.example.solidconnection.common.BaseEntity; +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.FetchType; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.Index; +import jakarta.persistence.JoinColumn; +import jakarta.persistence.ManyToOne; +import jakarta.persistence.Table; +import jakarta.persistence.UniqueConstraint; +import java.time.ZonedDateTime; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.NoArgsConstructor; +import org.hibernate.annotations.OnDelete; +import org.hibernate.annotations.OnDeleteAction; + +@Entity +@Getter +@NoArgsConstructor(access = AccessLevel.PROTECTED) +@Table( + name = "chat_message_outbox", + uniqueConstraints = @UniqueConstraint( + name = "uk_chat_message_outbox_message_recipient", + columnNames = {"chat_message_id", "recipient_chat_participant_id"} + ), + indexes = { + @Index( + name = "idx_chat_message_outbox_pending", + columnList = "status, next_attempt_at" + ), + @Index( + name = "idx_chat_message_outbox_processing", + columnList = "status, processing_started_at" + ) + } +) +public class ChatMessageOutbox extends BaseEntity { + + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + @Column(name = "id") + private Long id; + + @ManyToOne(fetch = FetchType.LAZY, optional = false) + @JoinColumn(name = "chat_message_id", nullable = false) + @OnDelete(action = OnDeleteAction.CASCADE) + private ChatMessage chatMessage; + + @ManyToOne(fetch = FetchType.LAZY, optional = false) + @JoinColumn(name = "recipient_chat_participant_id", nullable = false) + @OnDelete(action = OnDeleteAction.CASCADE) + private ChatParticipant recipient; + + @Enumerated(EnumType.STRING) + @Column(name = "status", nullable = false, length = 20) + private ChatMessageOutboxStatus status; + + @Enumerated(EnumType.STRING) + @Column(name = "completion_type", length = 30) + private ChatMessageOutboxCompletionType completionType; + + @Column(name = "attempt_count", nullable = false) + private int attemptCount; + + @Column(name = "processing_token", length = 36) + private String processingToken; + + @Column(name = "processing_started_at") + private ZonedDateTime processingStartedAt; + + @Column(name = "next_attempt_at") + private ZonedDateTime nextAttemptAt; + + @Column(name = "last_error", length = 1000) + private String lastError; + + @Column(name = "completed_at") + private ZonedDateTime completedAt; + + public ChatMessageOutbox( + ChatMessage chatMessage, + ChatParticipant recipient, + ZonedDateTime nextAttemptAt + ) { + this.chatMessage = chatMessage; + this.recipient = recipient; + this.status = ChatMessageOutboxStatus.PENDING; + this.attemptCount = 0; + this.nextAttemptAt = nextAttemptAt; + } +} diff --git a/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutboxCompletionType.java b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutboxCompletionType.java new file mode 100644 index 000000000..024e14b38 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutboxCompletionType.java @@ -0,0 +1,7 @@ +package com.example.solidconnection.chat.domain; + +public enum ChatMessageOutboxCompletionType { + + BROKER_DISPATCHED, + HISTORY_SERVED +} diff --git a/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutboxStatus.java b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutboxStatus.java new file mode 100644 index 000000000..59ba01a20 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutboxStatus.java @@ -0,0 +1,9 @@ +package com.example.solidconnection.chat.domain; + +public enum ChatMessageOutboxStatus { + + PENDING, + PROCESSING, + COMPLETED, + FAILED +} diff --git a/src/main/java/com/example/solidconnection/chat/dto/ClaimedChatMessageOutbox.java b/src/main/java/com/example/solidconnection/chat/dto/ClaimedChatMessageOutbox.java new file mode 100644 index 000000000..1f1487fad --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/dto/ClaimedChatMessageOutbox.java @@ -0,0 +1,10 @@ +package com.example.solidconnection.chat.dto; + +public record ClaimedChatMessageOutbox( + long outboxId, + String processingToken, + int attemptCount, + long roomId, + ChatMessageSendResponse response +) { +} diff --git a/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEvent.java b/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEvent.java index 72c06f57f..ce3c144b3 100644 --- a/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEvent.java +++ b/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEvent.java @@ -1,10 +1,6 @@ package com.example.solidconnection.chat.event; -import com.example.solidconnection.chat.dto.ChatMessageSendResponse; - public record ChatMessageBroadcastEvent( - long roomId, - ChatMessageSendResponse response + long outboxId ) { - } diff --git a/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListener.java b/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListener.java index 3a3c65040..24273eb23 100644 --- a/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListener.java +++ b/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListener.java @@ -1,29 +1,19 @@ package com.example.solidconnection.chat.event; +import com.example.solidconnection.chat.service.ChatMessageOutboxDispatcher; import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.messaging.simp.SimpMessageSendingOperations; import org.springframework.stereotype.Component; import org.springframework.transaction.event.TransactionPhase; import org.springframework.transaction.event.TransactionalEventListener; @Component @RequiredArgsConstructor -@Slf4j public class ChatMessageBroadcastEventListener { - private final SimpMessageSendingOperations simpMessageSendingOperations; + private final ChatMessageOutboxDispatcher dispatcher; @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) public void broadcast(ChatMessageBroadcastEvent event) { - try { - simpMessageSendingOperations.convertAndSend( - "/topic/chat/" + event.roomId(), - event.response() - ); - } catch (RuntimeException e) { - log.error("채팅 메시지를 방송하지 못했습니다. roomId={}, messageId={}", - event.roomId(), event.response().messageId(), e); - } + dispatcher.dispatch(event.outboxId()); } } diff --git a/src/main/java/com/example/solidconnection/chat/repository/ChatMessageOutboxRepository.java b/src/main/java/com/example/solidconnection/chat/repository/ChatMessageOutboxRepository.java new file mode 100644 index 000000000..e73b8188b --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/repository/ChatMessageOutboxRepository.java @@ -0,0 +1,173 @@ +package com.example.solidconnection.chat.repository; + +import com.example.solidconnection.chat.domain.ChatMessageOutbox; +import com.example.solidconnection.chat.domain.ChatMessageOutboxCompletionType; +import com.example.solidconnection.chat.domain.ChatMessageOutboxStatus; +import java.time.ZonedDateTime; +import java.util.List; +import java.util.Optional; +import org.springframework.data.domain.Pageable; +import org.springframework.data.jpa.repository.EntityGraph; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; + +public interface ChatMessageOutboxRepository extends JpaRepository { + + @Modifying(clearAutomatically = true, flushAutomatically = true) + @Query(""" + UPDATE ChatMessageOutbox o + SET o.status = :processingStatus, + o.processingToken = :processingToken, + o.processingStartedAt = :now, + o.attemptCount = o.attemptCount + 1, + o.lastError = null, + o.updatedAt = :now + WHERE o.id = :outboxId + AND o.status = :pendingStatus + AND o.nextAttemptAt <= :now + AND o.attemptCount < :maxAttempts + """) + int claim( + @Param("outboxId") long outboxId, + @Param("processingToken") String processingToken, + @Param("now") ZonedDateTime now, + @Param("maxAttempts") int maxAttempts, + @Param("pendingStatus") ChatMessageOutboxStatus pendingStatus, + @Param("processingStatus") ChatMessageOutboxStatus processingStatus + ); + + @EntityGraph(attributePaths = { + "chatMessage", + "chatMessage.chatRoom", + "chatMessage.chatAttachments" + }) + @Query("SELECT o FROM ChatMessageOutbox o WHERE o.id = :outboxId") + Optional findWithMessageById(@Param("outboxId") long outboxId); + + @Query(""" + SELECT o.id FROM ChatMessageOutbox o + WHERE o.status = :status + AND o.nextAttemptAt <= :now + AND o.attemptCount < :maxAttempts + ORDER BY o.nextAttemptAt, o.id + """) + List findDispatchableIds( + @Param("status") ChatMessageOutboxStatus status, + @Param("now") ZonedDateTime now, + @Param("maxAttempts") int maxAttempts, + Pageable pageable + ); + + @Modifying(clearAutomatically = true, flushAutomatically = true) + @Query(""" + UPDATE ChatMessageOutbox o + SET o.status = :pendingStatus, + o.processingToken = null, + o.processingStartedAt = null, + o.nextAttemptAt = :now, + o.updatedAt = :now + WHERE o.status = :processingStatus + AND o.processingStartedAt < :cutoff + AND o.attemptCount < :maxAttempts + """) + int recoverRetryableStaleProcessing( + @Param("processingStatus") ChatMessageOutboxStatus processingStatus, + @Param("pendingStatus") ChatMessageOutboxStatus pendingStatus, + @Param("cutoff") ZonedDateTime cutoff, + @Param("now") ZonedDateTime now, + @Param("maxAttempts") int maxAttempts + ); + + @Modifying(clearAutomatically = true, flushAutomatically = true) + @Query(""" + UPDATE ChatMessageOutbox o + SET o.status = :failedStatus, + o.processingToken = null, + o.processingStartedAt = null, + o.nextAttemptAt = null, + o.updatedAt = :now + WHERE o.status = :processingStatus + AND o.processingStartedAt < :cutoff + AND o.attemptCount >= :maxAttempts + """) + int failExhaustedStaleProcessing( + @Param("processingStatus") ChatMessageOutboxStatus processingStatus, + @Param("failedStatus") ChatMessageOutboxStatus failedStatus, + @Param("cutoff") ZonedDateTime cutoff, + @Param("now") ZonedDateTime now, + @Param("maxAttempts") int maxAttempts + ); + + @Modifying(clearAutomatically = true, flushAutomatically = true) + @Query(""" + UPDATE ChatMessageOutbox o + SET o.status = :completedStatus, + o.completionType = :completionType, + o.processingToken = null, + o.processingStartedAt = null, + o.nextAttemptAt = null, + o.completedAt = :completedAt, + o.lastError = null, + o.updatedAt = :completedAt + WHERE o.id = :outboxId + AND o.status = :processingStatus + AND o.processingToken = :processingToken + """) + int markCompleted( + @Param("outboxId") long outboxId, + @Param("processingToken") String processingToken, + @Param("processingStatus") ChatMessageOutboxStatus processingStatus, + @Param("completedStatus") ChatMessageOutboxStatus completedStatus, + @Param("completionType") ChatMessageOutboxCompletionType completionType, + @Param("completedAt") ZonedDateTime completedAt + ); + + @Modifying(clearAutomatically = true, flushAutomatically = true) + @Query(""" + UPDATE ChatMessageOutbox o + SET o.status = :nextStatus, + o.processingToken = null, + o.processingStartedAt = null, + o.nextAttemptAt = :nextAttemptAt, + o.lastError = :lastError, + o.updatedAt = :failedAt + WHERE o.id = :outboxId + AND o.status = :processingStatus + AND o.processingToken = :processingToken + """) + int recordFailure( + @Param("outboxId") long outboxId, + @Param("processingToken") String processingToken, + @Param("processingStatus") ChatMessageOutboxStatus processingStatus, + @Param("nextStatus") ChatMessageOutboxStatus nextStatus, + @Param("nextAttemptAt") ZonedDateTime nextAttemptAt, + @Param("lastError") String lastError, + @Param("failedAt") ZonedDateTime failedAt + ); + + @Modifying(clearAutomatically = true, flushAutomatically = true) + @Query(""" + UPDATE ChatMessageOutbox o + SET o.status = :completedStatus, + o.completionType = :completionType, + o.processingToken = null, + o.processingStartedAt = null, + o.nextAttemptAt = null, + o.completedAt = :completedAt, + o.lastError = null, + o.updatedAt = :completedAt + WHERE o.recipient.id = :recipientParticipantId + AND o.chatMessage.id IN :messageIds + AND o.status IN :completableStatuses + """) + int markHistoryServed( + @Param("recipientParticipantId") long recipientParticipantId, + @Param("messageIds") List messageIds, + @Param("completableStatuses") List completableStatuses, + @Param("completedStatus") ChatMessageOutboxStatus completedStatus, + @Param("completionType") ChatMessageOutboxCompletionType completionType, + @Param("completedAt") ZonedDateTime completedAt + ); +} diff --git a/src/main/java/com/example/solidconnection/chat/repository/ChatParticipantRepository.java b/src/main/java/com/example/solidconnection/chat/repository/ChatParticipantRepository.java index 7a554e125..f78d3ac65 100644 --- a/src/main/java/com/example/solidconnection/chat/repository/ChatParticipantRepository.java +++ b/src/main/java/com/example/solidconnection/chat/repository/ChatParticipantRepository.java @@ -13,6 +13,8 @@ public interface ChatParticipantRepository extends JpaRepository findByChatRoomIdAndSiteUserId(long chatRoomId, long siteUserId); + List findAllByChatRoomIdAndSiteUserIdNot(long chatRoomId, long siteUserId); + void deleteAllByChatRoomIdIn(List chatRoomIds); @Query(""" diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java new file mode 100644 index 000000000..9fd857e99 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java @@ -0,0 +1,84 @@ +package com.example.solidconnection.chat.service; + +import static java.time.ZoneOffset.UTC; +import static java.time.temporal.ChronoUnit.MICROS; + +import com.example.solidconnection.chat.domain.ChatMessageOutboxCompletionType; +import com.example.solidconnection.chat.dto.ClaimedChatMessageOutbox; +import java.time.ZonedDateTime; +import java.util.Optional; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.messaging.simp.SimpMessageSendingOperations; +import org.springframework.stereotype.Service; + +@Service +@RequiredArgsConstructor +@Slf4j +public class ChatMessageOutboxDispatcher { + + private final ChatMessageOutboxTransactionService outboxTransactionService; + private final SimpMessageSendingOperations simpMessageSendingOperations; + + public void dispatch(long outboxId) { + Optional claim; + try { + claim = outboxTransactionService.claim(outboxId, now()); + } catch (RuntimeException e) { + log.error("채팅 메시지 Outbox 처리권을 획득하지 못했습니다. outboxId={}", outboxId, e); + return; + } + claim.ifPresent(this::broadcast); + } + + private void broadcast(ClaimedChatMessageOutbox claimed) { + try { + simpMessageSendingOperations.convertAndSend( + "/topic/chat/" + claimed.roomId(), + claimed.response() + ); + } catch (RuntimeException e) { + recordFailure(claimed, e); + return; + } + + try { + boolean updated = outboxTransactionService.markCompleted( + claimed.outboxId(), + claimed.processingToken(), + ChatMessageOutboxCompletionType.BROKER_DISPATCHED, + now() + ); + if (!updated) { + log.info("이미 다른 경로에서 완료된 채팅 메시지 Outbox입니다. outboxId={}, messageId={}", + claimed.outboxId(), claimed.response().messageId()); + } + } catch (RuntimeException e) { + log.error("방송한 채팅 메시지 Outbox의 완료 상태를 기록하지 못했습니다. outboxId={}, messageId={}", + claimed.outboxId(), claimed.response().messageId(), e); + } + } + + private void recordFailure(ClaimedChatMessageOutbox claimed, RuntimeException broadcastException) { + try { + boolean updated = outboxTransactionService.recordFailure( + claimed.outboxId(), + claimed.processingToken(), + broadcastException.getMessage(), + now() + ); + if (!updated) { + log.info("이미 다른 경로에서 상태가 변경된 채팅 메시지 Outbox입니다. outboxId={}, messageId={}", + claimed.outboxId(), claimed.response().messageId()); + } + } catch (RuntimeException stateException) { + broadcastException.addSuppressed(stateException); + } + log.error("채팅 메시지를 방송하지 못했습니다. outboxId={}, roomId={}, messageId={}", + claimed.outboxId(), claimed.roomId(), claimed.response().messageId(), broadcastException); + } + + private ZonedDateTime now() { + return ZonedDateTime.now(UTC).truncatedTo(MICROS); + } +} diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxScheduler.java b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxScheduler.java new file mode 100644 index 000000000..b4842a95f --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxScheduler.java @@ -0,0 +1,28 @@ +package com.example.solidconnection.chat.service; + +import static java.time.ZoneOffset.UTC; +import static java.time.temporal.ChronoUnit.MICROS; + +import com.example.solidconnection.chat.config.ChatMessageOutboxProperties; +import java.time.ZonedDateTime; +import lombok.RequiredArgsConstructor; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +@Component +@RequiredArgsConstructor +@ConditionalOnProperty(name = "chat.outbox.scheduling.enabled", havingValue = "true") +public class ChatMessageOutboxScheduler { + + private final ChatMessageOutboxTransactionService outboxTransactionService; + private final ChatMessageOutboxDispatcher dispatcher; + private final ChatMessageOutboxProperties properties; + + @Scheduled(fixedDelayString = "${chat.outbox.scheduling.delay:5000}") + public void dispatchPendingMessages() { + ZonedDateTime now = ZonedDateTime.now(UTC).truncatedTo(MICROS); + outboxTransactionService.recoverStaleProcessing(now.minus(properties.getProcessingTimeout()), now); + outboxTransactionService.findDispatchableIds(now).forEach(dispatcher::dispatch); + } +} diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionService.java b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionService.java new file mode 100644 index 000000000..da4368719 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionService.java @@ -0,0 +1,178 @@ +package com.example.solidconnection.chat.service; + +import com.example.solidconnection.chat.config.ChatMessageOutboxProperties; +import com.example.solidconnection.chat.domain.ChatMessage; +import com.example.solidconnection.chat.domain.ChatMessageOutbox; +import com.example.solidconnection.chat.domain.ChatMessageOutboxCompletionType; +import com.example.solidconnection.chat.domain.ChatMessageOutboxStatus; +import com.example.solidconnection.chat.domain.ChatParticipant; +import com.example.solidconnection.chat.dto.ChatMessageSendResponse; +import com.example.solidconnection.chat.dto.ClaimedChatMessageOutbox; +import com.example.solidconnection.chat.repository.ChatMessageOutboxRepository; +import com.example.solidconnection.chat.repository.ChatParticipantRepository; +import java.time.Duration; +import java.time.ZonedDateTime; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import lombok.RequiredArgsConstructor; +import org.springframework.data.domain.PageRequest; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; + +@Service +@RequiredArgsConstructor +public class ChatMessageOutboxTransactionService { + + private static final int LAST_ERROR_MAX_LENGTH = 1000; + + private final ChatMessageOutboxRepository outboxRepository; + private final ChatParticipantRepository chatParticipantRepository; + private final ChatMessageOutboxProperties properties; + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public Optional claim(long outboxId, ZonedDateTime now) { + String processingToken = UUID.randomUUID().toString(); + int updatedCount = outboxRepository.claim( + outboxId, + processingToken, + now, + properties.getMaxAttempts(), + ChatMessageOutboxStatus.PENDING, + ChatMessageOutboxStatus.PROCESSING + ); + if (updatedCount == 0) { + return Optional.empty(); + } + + ChatMessageOutbox outbox = outboxRepository.findWithMessageById(outboxId).orElseThrow(); + ChatMessage chatMessage = outbox.getChatMessage(); + ChatParticipant sender = chatParticipantRepository.findById(chatMessage.getSenderId()).orElseThrow(); + ChatMessageSendResponse response = ChatMessageSendResponse.of(chatMessage, sender.getSiteUserId()); + + return Optional.of(new ClaimedChatMessageOutbox( + outbox.getId(), + processingToken, + outbox.getAttemptCount(), + chatMessage.getChatRoom().getId(), + response + )); + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public boolean markCompleted( + long outboxId, + String processingToken, + ChatMessageOutboxCompletionType completionType, + ZonedDateTime completedAt + ) { + return outboxRepository.markCompleted( + outboxId, + processingToken, + ChatMessageOutboxStatus.PROCESSING, + ChatMessageOutboxStatus.COMPLETED, + completionType, + completedAt + ) == 1; + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public boolean recordFailure( + long outboxId, + String processingToken, + String lastError, + ZonedDateTime failedAt + ) { + ChatMessageOutbox outbox = outboxRepository.findById(outboxId).orElseThrow(); + boolean exhausted = outbox.getAttemptCount() >= properties.getMaxAttempts(); + ChatMessageOutboxStatus nextStatus = exhausted + ? ChatMessageOutboxStatus.FAILED + : ChatMessageOutboxStatus.PENDING; + ZonedDateTime nextAttemptAt = exhausted + ? null + : failedAt.plus(calculateBackoff(outbox.getAttemptCount())); + + return outboxRepository.recordFailure( + outboxId, + processingToken, + ChatMessageOutboxStatus.PROCESSING, + nextStatus, + nextAttemptAt, + truncate(lastError), + failedAt + ) == 1; + } + + private Duration calculateBackoff(int attemptCount) { + Duration delay = properties.getInitialRetryDelay(); + for (int attempt = 1; attempt < attemptCount; attempt++) { + if (delay.compareTo(properties.getMaxRetryDelay()) >= 0) { + return properties.getMaxRetryDelay(); + } + delay = delay.multipliedBy(2); + } + return delay.compareTo(properties.getMaxRetryDelay()) > 0 + ? properties.getMaxRetryDelay() + : delay; + } + + private String truncate(String value) { + if (value == null || value.length() <= LAST_ERROR_MAX_LENGTH) { + return value; + } + return value.substring(0, LAST_ERROR_MAX_LENGTH); + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public int recoverStaleProcessing(ZonedDateTime cutoff, ZonedDateTime now) { + int recoveredCount = outboxRepository.recoverRetryableStaleProcessing( + ChatMessageOutboxStatus.PROCESSING, + ChatMessageOutboxStatus.PENDING, + cutoff, + now, + properties.getMaxAttempts() + ); + int failedCount = outboxRepository.failExhaustedStaleProcessing( + ChatMessageOutboxStatus.PROCESSING, + ChatMessageOutboxStatus.FAILED, + cutoff, + now, + properties.getMaxAttempts() + ); + return recoveredCount + failedCount; + } + + @Transactional(propagation = Propagation.REQUIRES_NEW, readOnly = true) + public List findDispatchableIds(ZonedDateTime now) { + return outboxRepository.findDispatchableIds( + ChatMessageOutboxStatus.PENDING, + now, + properties.getMaxAttempts(), + PageRequest.of(0, properties.getBatchSize()) + ); + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public int markHistoryServed( + long recipientParticipantId, + List messageIds, + ZonedDateTime completedAt + ) { + if (messageIds.isEmpty()) { + return 0; + } + return outboxRepository.markHistoryServed( + recipientParticipantId, + messageIds, + List.of( + ChatMessageOutboxStatus.PENDING, + ChatMessageOutboxStatus.PROCESSING, + ChatMessageOutboxStatus.FAILED + ), + ChatMessageOutboxStatus.COMPLETED, + ChatMessageOutboxCompletionType.HISTORY_SERVED, + completedAt + ); + } +} diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatService.java b/src/main/java/com/example/solidconnection/chat/service/ChatService.java index 66ab3476a..c12d770b9 100644 --- a/src/main/java/com/example/solidconnection/chat/service/ChatService.java +++ b/src/main/java/com/example/solidconnection/chat/service/ChatService.java @@ -4,22 +4,25 @@ import static com.example.solidconnection.common.exception.ErrorCode.CHAT_PARTNER_NOT_FOUND; import static com.example.solidconnection.common.exception.ErrorCode.INVALID_CHAT_ROOM_STATE; import static com.example.solidconnection.common.exception.ErrorCode.USER_NOT_FOUND; +import static java.time.ZoneOffset.UTC; +import static java.time.temporal.ChronoUnit.MICROS; import com.example.solidconnection.chat.domain.ChatAttachment; import com.example.solidconnection.chat.domain.ChatMessage; +import com.example.solidconnection.chat.domain.ChatMessageOutbox; import com.example.solidconnection.chat.domain.ChatParticipant; import com.example.solidconnection.chat.domain.ChatRoom; import com.example.solidconnection.chat.dto.ChatAttachmentResponse; import com.example.solidconnection.chat.dto.ChatImageSendRequest; import com.example.solidconnection.chat.dto.ChatMessageResponse; import com.example.solidconnection.chat.dto.ChatMessageSendRequest; -import com.example.solidconnection.chat.dto.ChatMessageSendResponse; import com.example.solidconnection.chat.dto.ChatParticipantResponse; import com.example.solidconnection.chat.dto.ChatRoomData; import com.example.solidconnection.chat.dto.ChatRoomListResponse; import com.example.solidconnection.chat.dto.ChatRoomResponse; import com.example.solidconnection.chat.event.ChatMessageBroadcastEvent; import com.example.solidconnection.chat.repository.ChatMessageRepository; +import com.example.solidconnection.chat.repository.ChatMessageOutboxRepository; import com.example.solidconnection.chat.repository.ChatParticipantRepository; import com.example.solidconnection.chat.repository.ChatReadStatusRepository; import com.example.solidconnection.chat.repository.ChatRoomRepository; @@ -28,6 +31,7 @@ import com.example.solidconnection.mentor.repository.MentorRepository; import com.example.solidconnection.siteuser.domain.SiteUser; import com.example.solidconnection.siteuser.repository.SiteUserRepository; +import java.time.ZonedDateTime; import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -47,27 +51,33 @@ public class ChatService { private final ChatRoomRepository chatRoomRepository; private final ChatMessageRepository chatMessageRepository; + private final ChatMessageOutboxRepository chatMessageOutboxRepository; private final ChatParticipantRepository chatParticipantRepository; private final ChatReadStatusRepository chatReadStatusRepository; private final SiteUserRepository siteUserRepository; private final MentorRepository mentorRepository; private final ApplicationEventPublisher eventPublisher; + private final ChatMessageOutboxTransactionService outboxTransactionService; public ChatService(ChatRoomRepository chatRoomRepository, ChatMessageRepository chatMessageRepository, + ChatMessageOutboxRepository chatMessageOutboxRepository, ChatParticipantRepository chatParticipantRepository, ChatReadStatusRepository chatReadStatusRepository, SiteUserRepository siteUserRepository, MentorRepository mentorRepository, - ApplicationEventPublisher eventPublisher) { + ApplicationEventPublisher eventPublisher, + ChatMessageOutboxTransactionService outboxTransactionService) { this.chatRoomRepository = chatRoomRepository; this.chatMessageRepository = chatMessageRepository; + this.chatMessageOutboxRepository = chatMessageOutboxRepository; this.chatParticipantRepository = chatParticipantRepository; this.chatReadStatusRepository = chatReadStatusRepository; this.siteUserRepository = siteUserRepository; this.mentorRepository = mentorRepository; this.eventPublisher = eventPublisher; + this.outboxTransactionService = outboxTransactionService; } @Transactional(readOnly = true) @@ -120,12 +130,17 @@ private ChatRoomResponse createChatRoomResponse(ChatRoom chatRoom, long siteUser @Transactional(readOnly = true) public SliceResponse getChatMessages(long siteUserId, long roomId, Pageable pageable) { - validateChatRoomParticipant(siteUserId, roomId); + ChatParticipant participant = getChatParticipant(siteUserId, roomId); Slice chatMessages = chatMessageRepository.findByRoomIdWithPaging(roomId, pageable); Map participantIdToParticipant = buildParticipantIdToParticipantMap(chatMessages); List content = buildChatMessageResponses(chatMessages, participantIdToParticipant); + outboxTransactionService.markHistoryServed( + participant.getId(), + chatMessages.getContent().stream().map(ChatMessage::getId).toList(), + now() + ); return SliceResponse.of(content, chatMessages); } @@ -178,10 +193,12 @@ private ChatParticipant findPartner(ChatRoom chatRoom, long siteUserId) { } public void validateChatRoomParticipant(long siteUserId, long roomId) { - boolean isParticipant = chatParticipantRepository.existsByChatRoomIdAndSiteUserId(roomId, siteUserId); - if (!isParticipant) { - throw new CustomException(CHAT_PARTICIPANT_NOT_FOUND); - } + getChatParticipant(siteUserId, roomId); + } + + private ChatParticipant getChatParticipant(long siteUserId, long roomId) { + return chatParticipantRepository.findByChatRoomIdAndSiteUserId(roomId, siteUserId) + .orElseThrow(() -> new CustomException(CHAT_PARTICIPANT_NOT_FOUND)); } private ChatMessageResponse toChatMessageResponse(ChatMessage message, long externalSenderId) { @@ -215,34 +232,28 @@ public void markChatMessagesAsRead(long siteUserId, long roomId) { @Transactional public void sendChatMessage(ChatMessageSendRequest chatMessageSendRequest, long siteUserId, long roomId) { - long senderId = chatParticipantRepository.findByChatRoomIdAndSiteUserId(roomId, siteUserId) - .orElseThrow(() -> new CustomException(CHAT_PARTICIPANT_NOT_FOUND)) - .getId(); + ChatParticipant sender = getChatParticipant(siteUserId, roomId); + ChatRoom chatRoom = chatRoomRepository.findById(roomId) + .orElseThrow(() -> new CustomException(INVALID_CHAT_ROOM_STATE)); ChatMessage chatMessage = new ChatMessage( chatMessageSendRequest.content(), - senderId, - chatRoomRepository.findById(roomId) - .orElseThrow(() -> new CustomException(INVALID_CHAT_ROOM_STATE)) + sender.getId(), + chatRoom ); chatMessageRepository.save(chatMessage); - - ChatMessageSendResponse chatMessageResponse = ChatMessageSendResponse.of(chatMessage, siteUserId); - - eventPublisher.publishEvent(new ChatMessageBroadcastEvent(roomId, chatMessageResponse)); + publishOutboxEvent(chatMessage, findOutboxRecipient(roomId, siteUserId)); } @Transactional public void sendChatImage(ChatImageSendRequest chatImageSendRequest, long siteUserId, long roomId) { - long senderId = chatParticipantRepository.findByChatRoomIdAndSiteUserId(roomId, siteUserId) - .orElseThrow(() -> new CustomException(CHAT_PARTICIPANT_NOT_FOUND)) - .getId(); + ChatParticipant sender = getChatParticipant(siteUserId, roomId); ChatRoom chatRoom = chatRoomRepository.findById(roomId) .orElseThrow(() -> new CustomException(INVALID_CHAT_ROOM_STATE)); - ChatMessage chatMessage = new ChatMessage("", senderId, chatRoom); + ChatMessage chatMessage = new ChatMessage("", sender.getId(), chatRoom); // 이미지 판별을 위한 확장자 리스트 List imageExtensions = Arrays.asList("jpg", "jpeg", "png", "webp"); @@ -259,9 +270,27 @@ public void sendChatImage(ChatImageSendRequest chatImageSendRequest, long siteUs } chatMessageRepository.save(chatMessage); + publishOutboxEvent(chatMessage, findOutboxRecipient(roomId, siteUserId)); + } + + private ChatParticipant findOutboxRecipient(long roomId, long senderSiteUserId) { + List recipients = + chatParticipantRepository.findAllByChatRoomIdAndSiteUserIdNot(roomId, senderSiteUserId); + if (recipients.size() != 1) { + throw new CustomException(INVALID_CHAT_ROOM_STATE); + } + return recipients.get(0); + } + + private void publishOutboxEvent(ChatMessage chatMessage, ChatParticipant recipient) { + ChatMessageOutbox outbox = chatMessageOutboxRepository.save( + new ChatMessageOutbox(chatMessage, recipient, now()) + ); + eventPublisher.publishEvent(new ChatMessageBroadcastEvent(outbox.getId())); + } - ChatMessageSendResponse chatMessageResponse = ChatMessageSendResponse.of(chatMessage, siteUserId); - eventPublisher.publishEvent(new ChatMessageBroadcastEvent(roomId, chatMessageResponse)); + private ZonedDateTime now() { + return ZonedDateTime.now(UTC).truncatedTo(MICROS); } private String generateThumbnailUrl(String originalUrl) { diff --git a/src/main/resources/config/application-variable.yml b/src/main/resources/config/application-variable.yml index c8c1c974a..5c07ddcc2 100644 --- a/src/main/resources/config/application-variable.yml +++ b/src/main/resources/config/application-variable.yml @@ -3,6 +3,17 @@ view: scheduling: delay: 3000 +chat: + outbox: + max-attempts: 5 + batch-size: 100 + initial-retry-delay: 1s + max-retry-delay: 1m + processing-timeout: 1m + scheduling: + enabled: false + delay: 5000 + websocket: subscription-timeout: 30s thread-pool: diff --git a/src/main/resources/db/migration/V64__create_chat_message_outbox.sql b/src/main/resources/db/migration/V64__create_chat_message_outbox.sql new file mode 100644 index 000000000..c834266b5 --- /dev/null +++ b/src/main/resources/db/migration/V64__create_chat_message_outbox.sql @@ -0,0 +1,24 @@ +CREATE TABLE chat_message_outbox +( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + chat_message_id BIGINT NOT NULL, + recipient_chat_participant_id BIGINT NOT NULL, + status VARCHAR(20) NOT NULL, + completion_type VARCHAR(30) NULL, + attempt_count INT NOT NULL DEFAULT 0, + processing_token VARCHAR(36) NULL, + processing_started_at DATETIME(6) NULL, + next_attempt_at DATETIME(6) NULL, + last_error VARCHAR(1000) NULL, + completed_at DATETIME(6) NULL, + created_at DATETIME(6) NOT NULL, + updated_at DATETIME(6) NOT NULL, + CONSTRAINT uk_chat_message_outbox_message_recipient + UNIQUE (chat_message_id, recipient_chat_participant_id), + CONSTRAINT fk_chat_message_outbox_message + FOREIGN KEY (chat_message_id) REFERENCES chat_message (id) ON DELETE CASCADE, + CONSTRAINT fk_chat_message_outbox_recipient + FOREIGN KEY (recipient_chat_participant_id) REFERENCES chat_participant (id) ON DELETE CASCADE, + INDEX idx_chat_message_outbox_pending (status, next_attempt_at), + INDEX idx_chat_message_outbox_processing (status, processing_started_at) +); diff --git a/src/test/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListenerTest.java b/src/test/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListenerTest.java index 2cbad42c1..bfb6c1e03 100644 --- a/src/test/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListenerTest.java +++ b/src/test/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListenerTest.java @@ -1,79 +1,26 @@ package com.example.solidconnection.chat.event; -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatCode; -import static org.junit.jupiter.api.Assertions.assertAll; -import static org.mockito.Mockito.mock; +import static org.mockito.BDDMockito.then; -import ch.qos.logback.classic.Logger; -import ch.qos.logback.classic.spi.ILoggingEvent; -import ch.qos.logback.core.read.ListAppender; -import com.example.solidconnection.chat.domain.MessageType; -import com.example.solidconnection.chat.dto.ChatMessageSendResponse; -import java.util.List; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; +import com.example.solidconnection.chat.service.ChatMessageOutboxDispatcher; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; -import org.mockito.BDDMockito; -import org.slf4j.LoggerFactory; -import org.springframework.messaging.simp.SimpMessageSendingOperations; +import org.mockito.Mockito; @DisplayName("채팅 메시지 방송 이벤트 리스너 테스트") class ChatMessageBroadcastEventListenerTest { - private SimpMessageSendingOperations simpMessageSendingOperations; - private ChatMessageBroadcastEventListener listener; - private ListAppender listAppender; - private Logger logger; - - @BeforeEach - void setUp() { - simpMessageSendingOperations = mock(SimpMessageSendingOperations.class); - listener = new ChatMessageBroadcastEventListener(simpMessageSendingOperations); - - logger = (Logger) LoggerFactory.getLogger(ChatMessageBroadcastEventListener.class); - listAppender = new ListAppender<>(); - listAppender.start(); - logger.addAppender(listAppender); - } - - @AfterEach - void tearDown() { - logger.detachAppender(listAppender); - listAppender.stop(); - } - @Test - void 방송에_실패하면_예외를_전파하지_않고_로그를_남긴다() { + void Outbox_식별자를_디스패처에_전달한다() { // given - long roomId = 1L; - long messageId = 2L; - ChatMessageSendResponse response = new ChatMessageSendResponse( - messageId, - "메시지", - 3L, - MessageType.TEXT, - List.of() - ); - ChatMessageBroadcastEvent event = new ChatMessageBroadcastEvent(roomId, response); - RuntimeException exception = new RuntimeException("broadcast failed"); - BDDMockito.willThrow(exception) - .given(simpMessageSendingOperations) - .convertAndSend("/topic/chat/" + roomId, response); + ChatMessageOutboxDispatcher dispatcher = Mockito.mock(ChatMessageOutboxDispatcher.class); + ChatMessageBroadcastEventListener listener = new ChatMessageBroadcastEventListener(dispatcher); + ChatMessageBroadcastEvent event = new ChatMessageBroadcastEvent(1L); - // when & then - assertThatCode(() -> listener.broadcast(event)).doesNotThrowAnyException(); + // when + listener.broadcast(event); - ILoggingEvent logEvent = listAppender.list.stream() - .filter(eventLog -> eventLog.getFormattedMessage().contains("채팅 메시지를 방송하지 못했습니다.")) - .findFirst() - .orElseThrow(); - assertAll( - () -> assertThat(logEvent.getLevel().toString()).isEqualTo("ERROR"), - () -> assertThat(logEvent.getFormattedMessage()).contains("roomId=1", "messageId=2"), - () -> assertThat(logEvent.getThrowableProxy()).isNotNull(), - () -> assertThat(logEvent.getThrowableProxy().getMessage()).isEqualTo("broadcast failed") - ); + // then + then(dispatcher).should().dispatch(1L); } } diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java new file mode 100644 index 000000000..504928e96 --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java @@ -0,0 +1,123 @@ +package com.example.solidconnection.chat.service; + +import static java.time.ZoneOffset.UTC; +import static java.time.temporal.ChronoUnit.MICROS; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.junit.jupiter.api.Assertions.assertAll; + +import com.example.solidconnection.chat.domain.ChatMessage; +import com.example.solidconnection.chat.domain.ChatMessageOutbox; +import com.example.solidconnection.chat.domain.ChatMessageOutboxCompletionType; +import com.example.solidconnection.chat.domain.ChatMessageOutboxStatus; +import com.example.solidconnection.chat.domain.ChatParticipant; +import com.example.solidconnection.chat.domain.ChatRoom; +import com.example.solidconnection.chat.dto.ChatMessageSendResponse; +import com.example.solidconnection.chat.fixture.ChatMessageFixture; +import com.example.solidconnection.chat.fixture.ChatParticipantFixture; +import com.example.solidconnection.chat.fixture.ChatRoomFixture; +import com.example.solidconnection.chat.repository.ChatMessageOutboxRepository; +import com.example.solidconnection.siteuser.domain.SiteUser; +import com.example.solidconnection.siteuser.fixture.SiteUserFixture; +import com.example.solidconnection.support.TestContainerSpringBootTest; +import java.time.ZonedDateTime; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.BDDMockito; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.messaging.simp.SimpMessagingTemplate; +import org.springframework.test.context.bean.override.mockito.MockitoBean; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +@TestContainerSpringBootTest +@DisplayName("채팅 메시지 Outbox 디스패처 테스트") +class ChatMessageOutboxDispatcherTest { + + @Autowired + private ChatMessageOutboxDispatcher dispatcher; + + @Autowired + private ChatMessageOutboxRepository outboxRepository; + + @Autowired + private SiteUserFixture siteUserFixture; + + @Autowired + private ChatRoomFixture chatRoomFixture; + + @Autowired + private ChatParticipantFixture chatParticipantFixture; + + @Autowired + private ChatMessageFixture chatMessageFixture; + + @MockitoBean + private SimpMessagingTemplate simpMessagingTemplate; + + private ChatMessageOutbox outbox; + private long roomId; + + @BeforeEach + void setUp() { + SiteUser senderUser = siteUserFixture.사용자(1, "sender"); + SiteUser recipientUser = siteUserFixture.사용자(2, "recipient"); + ChatRoom chatRoom = chatRoomFixture.채팅방(false); + roomId = chatRoom.getId(); + ChatParticipant sender = chatParticipantFixture.참여자(senderUser.getId(), chatRoom); + ChatParticipant recipient = chatParticipantFixture.참여자(recipientUser.getId(), chatRoom); + ChatMessage message = chatMessageFixture.메시지("메시지", sender.getId(), chatRoom); + ZonedDateTime now = ZonedDateTime.now(UTC).truncatedTo(MICROS); + outbox = outboxRepository.save(new ChatMessageOutbox(message, recipient, now.minusSeconds(1))); + } + + @Test + void DB_트랜잭션_밖에서_방송하고_성공_상태를_별도_커밋한다() { + // given + BDDMockito.willAnswer(invocation -> { + assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isFalse(); + return null; + }).given(simpMessagingTemplate).convertAndSend( + BDDMockito.anyString(), + BDDMockito.any(ChatMessageSendResponse.class) + ); + + // when + dispatcher.dispatch(outbox.getId()); + + // then + ArgumentCaptor responseCaptor = + ArgumentCaptor.forClass(ChatMessageSendResponse.class); + BDDMockito.then(simpMessagingTemplate).should().convertAndSend( + BDDMockito.eq("/topic/chat/" + roomId), + responseCaptor.capture() + ); + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(responseCaptor.getValue().messageId()) + .isEqualTo(outbox.getChatMessage().getId()), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), + () -> assertThat(saved.getCompletionType()) + .isEqualTo(ChatMessageOutboxCompletionType.BROKER_DISPATCHED) + ); + } + + @Test + void 방송에_실패하면_예외를_전파하지_않고_재시도를_예약한다() { + // given + BDDMockito.willThrow(new RuntimeException("broadcast failed")) + .given(simpMessagingTemplate) + .convertAndSend(BDDMockito.anyString(), BDDMockito.any(ChatMessageSendResponse.class)); + + // when & then + assertThatCode(() -> dispatcher.dispatch(outbox.getId())).doesNotThrowAnyException(); + + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.PENDING), + () -> assertThat(saved.getAttemptCount()).isEqualTo(1), + () -> assertThat(saved.getLastError()).contains("broadcast failed") + ); + } +} diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxSchedulerTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxSchedulerTest.java new file mode 100644 index 000000000..f4af8007c --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxSchedulerTest.java @@ -0,0 +1,46 @@ +package com.example.solidconnection.chat.service; + +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.then; + +import com.example.solidconnection.chat.config.ChatMessageOutboxProperties; +import java.time.Duration; +import java.time.ZonedDateTime; +import java.util.List; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +@DisplayName("채팅 메시지 Outbox 스케줄러 테스트") +class ChatMessageOutboxSchedulerTest { + + @Test + void 오래된_처리를_회수한_뒤_전송_가능한_Outbox를_배치로_처리한다() { + // given + ChatMessageOutboxTransactionService transactionService = + Mockito.mock(ChatMessageOutboxTransactionService.class); + ChatMessageOutboxDispatcher dispatcher = Mockito.mock(ChatMessageOutboxDispatcher.class); + ChatMessageOutboxProperties properties = new ChatMessageOutboxProperties(); + properties.setProcessingTimeout(Duration.ofMinutes(2)); + given(transactionService.findDispatchableIds(Mockito.any(ZonedDateTime.class))) + .willReturn(List.of(1L, 2L)); + ChatMessageOutboxScheduler scheduler = + new ChatMessageOutboxScheduler(transactionService, dispatcher, properties); + + // when + scheduler.dispatchPendingMessages(); + + // then + ArgumentCaptor cutoffCaptor = ArgumentCaptor.forClass(ZonedDateTime.class); + ArgumentCaptor nowCaptor = ArgumentCaptor.forClass(ZonedDateTime.class); + then(transactionService).should().recoverStaleProcessing(cutoffCaptor.capture(), nowCaptor.capture()); + then(transactionService).should().findDispatchableIds(nowCaptor.getValue()); + then(dispatcher).should().dispatch(1L); + then(dispatcher).should().dispatch(2L); + org.assertj.core.api.Assertions.assertThat(Duration.between( + cutoffCaptor.getValue(), + nowCaptor.getValue() + )).isEqualTo(Duration.ofMinutes(2)); + } +} diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java new file mode 100644 index 000000000..c10fb5926 --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java @@ -0,0 +1,303 @@ +package com.example.solidconnection.chat.service; + +import static java.time.ZoneOffset.UTC; +import static java.time.temporal.ChronoUnit.MICROS; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertAll; + +import com.example.solidconnection.chat.domain.ChatMessage; +import com.example.solidconnection.chat.domain.ChatMessageOutbox; +import com.example.solidconnection.chat.domain.ChatMessageOutboxCompletionType; +import com.example.solidconnection.chat.domain.ChatMessageOutboxStatus; +import com.example.solidconnection.chat.domain.ChatParticipant; +import com.example.solidconnection.chat.domain.ChatRoom; +import com.example.solidconnection.chat.dto.ClaimedChatMessageOutbox; +import com.example.solidconnection.chat.fixture.ChatMessageFixture; +import com.example.solidconnection.chat.fixture.ChatParticipantFixture; +import com.example.solidconnection.chat.fixture.ChatRoomFixture; +import com.example.solidconnection.chat.repository.ChatMessageOutboxRepository; +import com.example.solidconnection.siteuser.domain.SiteUser; +import com.example.solidconnection.siteuser.fixture.SiteUserFixture; +import com.example.solidconnection.support.TestContainerSpringBootTest; +import java.time.ZonedDateTime; +import java.util.List; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; + +@TestContainerSpringBootTest +@DisplayName("채팅 메시지 Outbox 트랜잭션 서비스 테스트") +class ChatMessageOutboxTransactionServiceTest { + + @Autowired + private ChatMessageOutboxTransactionService outboxTransactionService; + + @Autowired + private ChatMessageOutboxRepository outboxRepository; + + @Autowired + private SiteUserFixture siteUserFixture; + + @Autowired + private ChatRoomFixture chatRoomFixture; + + @Autowired + private ChatParticipantFixture chatParticipantFixture; + + @Autowired + private ChatMessageFixture chatMessageFixture; + + private ChatParticipant recipient; + private ChatMessageOutbox outbox; + private ZonedDateTime now; + + @BeforeEach + void setUp() { + SiteUser senderUser = siteUserFixture.사용자(1, "sender"); + SiteUser recipientUser = siteUserFixture.사용자(2, "recipient"); + ChatRoom chatRoom = chatRoomFixture.채팅방(false); + ChatParticipant sender = chatParticipantFixture.참여자(senderUser.getId(), chatRoom); + recipient = chatParticipantFixture.참여자(recipientUser.getId(), chatRoom); + ChatMessage message = chatMessageFixture.메시지("메시지", sender.getId(), chatRoom); + now = ZonedDateTime.now(UTC).truncatedTo(MICROS); + outbox = outboxRepository.save(new ChatMessageOutbox(message, recipient, now.minusMinutes(10))); + } + + @Nested + class 처리권을_획득한다 { + + @Test + void 처리권을_획득하면_시도_횟수와_token을_기록한다() { + // when + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim(outbox.getId(), now) + .orElseThrow(); + + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.PROCESSING), + () -> assertThat(saved.getAttemptCount()).isEqualTo(1), + () -> assertThat(saved.getProcessingToken()).isEqualTo(claimed.processingToken()), + () -> assertThat(claimed.response().messageId()).isEqualTo(outbox.getChatMessage().getId()) + ); + } + + @Test + void 이미_처리권을_획득한_Outbox는_다시_획득할_수_없다() { + // given + outboxTransactionService.claim(outbox.getId(), now).orElseThrow(); + + // when & then + assertThat(outboxTransactionService.claim(outbox.getId(), now)).isEmpty(); + } + } + + @Nested + class 처리_결과를_기록한다 { + + @Test + void 방송에_성공하면_BROKER_DISPATCHED로_완료한다() { + // given + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim(outbox.getId(), now) + .orElseThrow(); + + // when + boolean updated = outboxTransactionService.markCompleted( + outbox.getId(), + claimed.processingToken(), + ChatMessageOutboxCompletionType.BROKER_DISPATCHED, + now.plusSeconds(1) + ); + + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(updated).isTrue(), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), + () -> assertThat(saved.getCompletionType()) + .isEqualTo(ChatMessageOutboxCompletionType.BROKER_DISPATCHED), + () -> assertThat(saved.getProcessingToken()).isNull() + ); + } + + @Test + void 방송에_실패하면_다음_시도를_예약한다() { + // given + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim(outbox.getId(), now) + .orElseThrow(); + + // when + boolean updated = outboxTransactionService.recordFailure( + outbox.getId(), + claimed.processingToken(), + "broadcast failed", + now.plusSeconds(1) + ); + + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(updated).isTrue(), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.PENDING), + () -> assertThat(saved.getLastError()).isEqualTo("broadcast failed"), + () -> assertThat(saved.getNextAttemptAt()).isAfter(now.plusSeconds(1)) + ); + } + + @Test + void 최대_시도_횟수에_도달하면_FAILED로_종료한다() { + // given + ZonedDateTime attemptTime = now; + for (int attempt = 1; attempt <= 3; attempt++) { + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim(outbox.getId(), attemptTime) + .orElseThrow(); + outboxTransactionService.recordFailure( + outbox.getId(), + claimed.processingToken(), + "failure-" + attempt, + attemptTime + ); + attemptTime = outboxRepository.findById(outbox.getId()).orElseThrow().getNextAttemptAt(); + } + + // when + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + + // then + assertAll( + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.FAILED), + () -> assertThat(saved.getAttemptCount()).isEqualTo(3), + () -> assertThat(saved.getLastError()).isEqualTo("failure-3") + ); + } + } + + @Nested + class 오래된_처리를_복구한다 { + + @Test + void 최대_시도_횟수를_사용한_처리가_만료되면_FAILED로_종료한다() { + // given + ZonedDateTime attemptTime = now.minusMinutes(10); + for (int attempt = 1; attempt < 3; attempt++) { + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim(outbox.getId(), attemptTime) + .orElseThrow(); + outboxTransactionService.recordFailure( + outbox.getId(), + claimed.processingToken(), + "failure-" + attempt, + attemptTime + ); + attemptTime = outboxRepository.findById(outbox.getId()).orElseThrow().getNextAttemptAt(); + } + outboxTransactionService.claim(outbox.getId(), attemptTime).orElseThrow(); + + // when + int recoveredCount = outboxTransactionService.recoverStaleProcessing( + attemptTime.plusMinutes(1), + attemptTime.plusMinutes(2) + ); + + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(recoveredCount).isEqualTo(1), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.FAILED), + () -> assertThat(saved.getAttemptCount()).isEqualTo(3) + ); + } + + @Test + void 회수된_이전_worker의_늦은_결과가_새로운_완료_상태를_덮지_못한다() { + // given + ZonedDateTime firstAttemptAt = now.minusMinutes(2); + ClaimedChatMessageOutbox firstClaim = outboxTransactionService.claim(outbox.getId(), firstAttemptAt) + .orElseThrow(); + outboxTransactionService.recoverStaleProcessing(now.minusMinutes(1), now); + ClaimedChatMessageOutbox secondClaim = outboxTransactionService.claim(outbox.getId(), now) + .orElseThrow(); + outboxTransactionService.markCompleted( + outbox.getId(), + secondClaim.processingToken(), + ChatMessageOutboxCompletionType.BROKER_DISPATCHED, + now.plusSeconds(1) + ); + + // when + boolean staleUpdate = outboxTransactionService.recordFailure( + outbox.getId(), + firstClaim.processingToken(), + "late failure", + now.plusSeconds(2) + ); + + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(staleUpdate).isFalse(), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), + () -> assertThat(saved.getCompletionType()) + .isEqualTo(ChatMessageOutboxCompletionType.BROKER_DISPATCHED) + ); + } + } + + @Nested + class 내역_조회로_완료한다 { + + @Test + void 처리_중인_메시지도_HISTORY_SERVED로_완료하고_늦은_worker가_되돌리지_못한다() { + // given + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim(outbox.getId(), now) + .orElseThrow(); + + // when + int completedCount = outboxTransactionService.markHistoryServed( + recipient.getId(), + List.of(outbox.getChatMessage().getId()), + now.plusSeconds(1) + ); + boolean staleUpdate = outboxTransactionService.recordFailure( + outbox.getId(), + claimed.processingToken(), + "late failure", + now.plusSeconds(2) + ); + + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(completedCount).isEqualTo(1), + () -> assertThat(staleUpdate).isFalse(), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), + () -> assertThat(saved.getCompletionType()) + .isEqualTo(ChatMessageOutboxCompletionType.HISTORY_SERVED) + ); + } + + @Test + void 다른_참여자의_조회는_완료_처리하지_않는다() { + // given + ChatParticipant other = chatParticipantFixture.참여자( + siteUserFixture.사용자(3, "other").getId(), + chatRoomFixture.채팅방(false) + ); + + // when + int completedCount = outboxTransactionService.markHistoryServed( + other.getId(), + List.of(outbox.getChatMessage().getId()), + now + ); + + // then + assertAll( + () -> assertThat(completedCount).isZero(), + () -> assertThat(outboxRepository.findById(outbox.getId()).orElseThrow().getStatus()) + .isEqualTo(ChatMessageOutboxStatus.PENDING) + ); + } + } +} diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java index 88c88a493..8436fb06c 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java @@ -8,6 +8,9 @@ import com.example.solidconnection.chat.domain.ChatAttachment; import com.example.solidconnection.chat.domain.ChatMessage; +import com.example.solidconnection.chat.domain.ChatMessageOutbox; +import com.example.solidconnection.chat.domain.ChatMessageOutboxCompletionType; +import com.example.solidconnection.chat.domain.ChatMessageOutboxStatus; import com.example.solidconnection.chat.domain.ChatParticipant; import com.example.solidconnection.chat.domain.ChatReadStatus; import com.example.solidconnection.chat.domain.ChatRoom; @@ -24,6 +27,7 @@ import com.example.solidconnection.chat.fixture.ChatReadStatusFixture; import com.example.solidconnection.chat.fixture.ChatRoomFixture; import com.example.solidconnection.chat.repository.ChatMessageRepository; +import com.example.solidconnection.chat.repository.ChatMessageOutboxRepository; import com.example.solidconnection.chat.repository.ChatReadStatusRepositoryForTest; import com.example.solidconnection.common.dto.SliceResponse; import com.example.solidconnection.common.exception.CustomException; @@ -61,6 +65,9 @@ class ChatServiceTest { @Autowired private ChatMessageRepository chatMessageRepository; + @Autowired + private ChatMessageOutboxRepository chatMessageOutboxRepository; + @Autowired private TransactionTemplate transactionTemplate; @@ -219,12 +226,14 @@ class 채팅_메시지를_조회한다 { private ChatRoom chatRoom; private Pageable pageable; + private ChatParticipant participant; + private ChatParticipant partnerParticipant; @BeforeEach void setUp() { chatRoom = chatRoomFixture.채팅방(false); - chatParticipantFixture.참여자(user.getId(), chatRoom); - chatParticipantFixture.참여자(mentor1.getId(), chatRoom); + participant = chatParticipantFixture.참여자(user.getId(), chatRoom); + partnerParticipant = chatParticipantFixture.참여자(mentor1.getId(), chatRoom); pageable = PageRequest.of(0, 20, Sort.by(Sort.Direction.DESC, "createdAt")); } @@ -321,6 +330,46 @@ void setUp() { .isInstanceOf(CustomException.class) .hasMessage(CHAT_PARTICIPANT_NOT_FOUND.getMessage()); } + + @Test + void 조회_응답에_포함된_상대방_메시지의_Outbox를_HISTORY_SERVED로_완료한다() { + // given + ChatMessage message = chatMessageFixture.메시지("조회 복구 메시지", partnerParticipant.getId(), chatRoom); + ChatMessageOutbox outbox = chatMessageOutboxRepository.save(new ChatMessageOutbox( + message, + participant, + ZonedDateTime.now().minusMinutes(1) + )); + + // when + chatService.getChatMessages(user.getId(), chatRoom.getId(), pageable); + + // then + ChatMessageOutbox saved = chatMessageOutboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), + () -> assertThat(saved.getCompletionType()) + .isEqualTo(ChatMessageOutboxCompletionType.HISTORY_SERVED) + ); + } + + @Test + void 송신자의_조회는_상대방_Outbox를_완료하지_않는다() { + // given + ChatMessage message = chatMessageFixture.메시지("내가 보낸 메시지", participant.getId(), chatRoom); + ChatMessageOutbox outbox = chatMessageOutboxRepository.save(new ChatMessageOutbox( + message, + partnerParticipant, + ZonedDateTime.now().minusMinutes(1) + )); + + // when + chatService.getChatMessages(user.getId(), chatRoom.getId(), pageable); + + // then + assertThat(chatMessageOutboxRepository.findById(outbox.getId()).orElseThrow().getStatus()) + .isEqualTo(ChatMessageOutboxStatus.PENDING); + } } @Nested @@ -414,6 +463,7 @@ class 채팅_메시지를_전송한다 { private SiteUser sender; private ChatParticipant senderParticipant; + private ChatParticipant recipientParticipant; private ChatRoom chatRoom; @BeforeEach @@ -421,6 +471,7 @@ void setUp() { sender = siteUserFixture.사용자(111, "sender"); chatRoom = chatRoomFixture.채팅방(false); senderParticipant = chatParticipantFixture.참여자(sender.getId(), chatRoom); + recipientParticipant = chatParticipantFixture.참여자(mentor1.getId(), chatRoom); } @Test @@ -452,7 +503,10 @@ public void beforeCommit(boolean readOnly) { () -> assertThat(destinationCaptor.getValue()).isEqualTo("/topic/chat/" + chatRoom.getId()), () -> assertThat(payloadCaptor.getValue().content()).isEqualTo(content), () -> assertThat(payloadCaptor.getValue().siteUserId()).isEqualTo(sender.getId()), - () -> assertThat(chatMessageRepository.count()).isEqualTo(messageCount + 1) + () -> assertThat(chatMessageRepository.count()).isEqualTo(messageCount + 1), + () -> assertThat(chatMessageOutboxRepository.findAll()).singleElement() + .extracting(outbox -> outbox.getRecipient().getId()) + .isEqualTo(recipientParticipant.getId()) ); } @@ -472,7 +526,8 @@ public void beforeCommit(boolean readOnly) { // then assertAll( () -> BDDMockito.then(simpMessagingTemplate).shouldHaveNoInteractions(), - () -> assertThat(chatMessageRepository.count()).isEqualTo(messageCount) + () -> assertThat(chatMessageRepository.count()).isEqualTo(messageCount), + () -> assertThat(chatMessageOutboxRepository.count()).isZero() ); } @@ -497,6 +552,7 @@ class 채팅_이미지를_전송한다 { private static final String EXPECTED_THUMBNAIL_URL = "https://bucket.s3.ap-northeast-2.amazonaws.com/chat/thumbnails/example_thumb.jpg"; private SiteUser sender; private ChatParticipant senderParticipant; + private ChatParticipant recipientParticipant; private ChatRoom chatRoom; @BeforeEach @@ -504,6 +560,7 @@ void setUp() { sender = siteUserFixture.사용자(111, "sender"); chatRoom = chatRoomFixture.채팅방(false); senderParticipant = chatParticipantFixture.참여자(sender.getId(), chatRoom); + recipientParticipant = chatParticipantFixture.참여자(mentor1.getId(), chatRoom); } @Test @@ -544,7 +601,10 @@ public void beforeCommit(boolean readOnly) { () -> assertThat(response.messageType()).isEqualTo(MessageType.IMAGE), () -> assertThat(response.siteUserId()).isEqualTo(sender.getId()), () -> assertThat(response.content()).isEmpty(), - () -> assertThat(chatMessageRepository.count()).isEqualTo(messageCount + 1) + () -> assertThat(chatMessageRepository.count()).isEqualTo(messageCount + 1), + () -> assertThat(chatMessageOutboxRepository.findAll()).singleElement() + .extracting(outbox -> outbox.getRecipient().getId()) + .isEqualTo(recipientParticipant.getId()) ); } @@ -564,7 +624,8 @@ public void beforeCommit(boolean readOnly) { // then assertAll( () -> BDDMockito.then(simpMessagingTemplate).shouldHaveNoInteractions(), - () -> assertThat(chatMessageRepository.count()).isEqualTo(messageCount) + () -> assertThat(chatMessageRepository.count()).isEqualTo(messageCount), + () -> assertThat(chatMessageOutboxRepository.count()).isZero() ); } diff --git a/src/test/resources/application.yml b/src/test/resources/application.yml index 3f8cbeaff..819d4c253 100644 --- a/src/test/resources/application.yml +++ b/src/test/resources/application.yml @@ -49,6 +49,15 @@ view: scheduling: enabled: false delay: 3000 +chat: + outbox: + max-attempts: 3 + batch-size: 10 + initial-retry-delay: 1s + max-retry-delay: 10s + processing-timeout: 1m + scheduling: + enabled: false websocket: subscription-timeout: 300ms thread-pool: From c9c19b581c2160cb9f4c9b36f90bcd370716f40e Mon Sep 17 00:00:00 2001 From: Hexeong <123macanic@naver.com> Date: Mon, 2 Nov 2026 19:56:45 +0900 Subject: [PATCH 2/8] =?UTF-8?q?fix:=20=EC=B1=84=ED=8C=85=20Outbox=20?= =?UTF-8?q?=EB=A6=AC=EB=B7=B0=20=EC=A7=80=EC=A0=81=20=EB=B0=98=EC=98=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../config/ChatMessageOutboxProperties.java | 5 + .../chat/dto/ChatMessageHistoryResult.java | 11 ++ .../ChatMessageOutboxRepository.java | 1 - .../ChatMessageHistoryQueryService.java | 88 ++++++++++++ .../ChatMessageOutboxTransactionService.java | 21 ++- .../chat/service/ChatService.java | 85 +++--------- .../ChatMessageOutboxPropertiesTest.java | 49 +++++++ .../ChatMessageOutboxDispatcherTest.java | 99 +++++++------- .../ChatMessageOutboxSchedulerTest.java | 6 +- ...atMessageOutboxTransactionServiceTest.java | 129 +++++++++++++++++- .../ChatServiceHistoryTransactionTest.java | 95 +++++++++++++ .../chat/service/ChatServiceTest.java | 84 +++++++++++- 12 files changed, 542 insertions(+), 131 deletions(-) create mode 100644 src/main/java/com/example/solidconnection/chat/dto/ChatMessageHistoryResult.java create mode 100644 src/main/java/com/example/solidconnection/chat/service/ChatMessageHistoryQueryService.java create mode 100644 src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java create mode 100644 src/test/java/com/example/solidconnection/chat/service/ChatServiceHistoryTransactionTest.java diff --git a/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java b/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java index 5b8a3cef3..bbf81ef35 100644 --- a/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java +++ b/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java @@ -1,18 +1,23 @@ package com.example.solidconnection.chat.config; +import jakarta.validation.constraints.Positive; import java.time.Duration; import lombok.Getter; import lombok.Setter; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; +import org.springframework.validation.annotation.Validated; @Component +@Validated @ConfigurationProperties(prefix = "chat.outbox") @Getter @Setter public class ChatMessageOutboxProperties { + @Positive private int maxAttempts = 5; + @Positive private int batchSize = 100; private Duration initialRetryDelay = Duration.ofSeconds(1); private Duration maxRetryDelay = Duration.ofMinutes(1); diff --git a/src/main/java/com/example/solidconnection/chat/dto/ChatMessageHistoryResult.java b/src/main/java/com/example/solidconnection/chat/dto/ChatMessageHistoryResult.java new file mode 100644 index 000000000..62332578e --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/dto/ChatMessageHistoryResult.java @@ -0,0 +1,11 @@ +package com.example.solidconnection.chat.dto; + +import com.example.solidconnection.common.dto.SliceResponse; +import java.util.List; + +public record ChatMessageHistoryResult( + SliceResponse response, + long participantId, + List messageIds +) { +} diff --git a/src/main/java/com/example/solidconnection/chat/repository/ChatMessageOutboxRepository.java b/src/main/java/com/example/solidconnection/chat/repository/ChatMessageOutboxRepository.java index e73b8188b..f2a7cf4c9 100644 --- a/src/main/java/com/example/solidconnection/chat/repository/ChatMessageOutboxRepository.java +++ b/src/main/java/com/example/solidconnection/chat/repository/ChatMessageOutboxRepository.java @@ -22,7 +22,6 @@ public interface ChatMessageOutboxRepository extends JpaRepository new CustomException(CHAT_PARTICIPANT_NOT_FOUND)); + Slice chatMessages = chatMessageRepository.findByRoomIdWithPaging(roomId, pageable); + Map participants = findParticipants(chatMessages); + List content = toResponses(chatMessages, participants); + + return new ChatMessageHistoryResult( + SliceResponse.of(content, chatMessages), + participant.getId(), + chatMessages.getContent().stream().map(ChatMessage::getId).toList() + ); + } + + private Map findParticipants(Slice chatMessages) { + Set participantIds = chatMessages.getContent().stream() + .map(ChatMessage::getSenderId) + .collect(Collectors.toSet()); + + return chatParticipantRepository.findAllById(participantIds).stream() + .collect(Collectors.toMap(ChatParticipant::getId, Function.identity())); + } + + private List toResponses( + Slice chatMessages, + Map participants + ) { + return chatMessages.getContent().stream() + .map(message -> toResponse(message, participants)) + .toList(); + } + + private ChatMessageResponse toResponse(ChatMessage message, Map participants) { + ChatParticipant sender = participants.get(message.getSenderId()); + if (sender == null) { + throw new CustomException(CHAT_PARTICIPANT_NOT_FOUND); + } + List attachments = message.getChatAttachments().stream() + .map(attachment -> ChatAttachmentResponse.of( + attachment.getId(), + attachment.getIsImage(), + attachment.getUrl(), + attachment.getThumbnailUrl(), + attachment.getCreatedAt() + )) + .toList(); + + return ChatMessageResponse.of( + message.getId(), + message.getContent(), + sender.getSiteUserId(), + message.getCreatedAt(), + attachments + ); + } +} diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionService.java b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionService.java index da4368719..65704d042 100644 --- a/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionService.java +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionService.java @@ -46,7 +46,20 @@ public Optional claim(long outboxId, ZonedDateTime now return Optional.empty(); } - ChatMessageOutbox outbox = outboxRepository.findWithMessageById(outboxId).orElseThrow(); + Optional foundOutbox = outboxRepository.findWithMessageById(outboxId); + if (foundOutbox.isEmpty()) { + outboxRepository.recordFailure( + outboxId, + processingToken, + ChatMessageOutboxStatus.PROCESSING, + ChatMessageOutboxStatus.FAILED, + null, + "message unavailable", + now + ); + return Optional.empty(); + } + ChatMessageOutbox outbox = foundOutbox.get(); ChatMessage chatMessage = outbox.getChatMessage(); ChatParticipant sender = chatParticipantRepository.findById(chatMessage.getSenderId()).orElseThrow(); ChatMessageSendResponse response = ChatMessageSendResponse.of(chatMessage, sender.getSiteUserId()); @@ -165,11 +178,7 @@ public int markHistoryServed( return outboxRepository.markHistoryServed( recipientParticipantId, messageIds, - List.of( - ChatMessageOutboxStatus.PENDING, - ChatMessageOutboxStatus.PROCESSING, - ChatMessageOutboxStatus.FAILED - ), + List.of(ChatMessageOutboxStatus.FAILED), ChatMessageOutboxStatus.COMPLETED, ChatMessageOutboxCompletionType.HISTORY_SERVED, completedAt diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatService.java b/src/main/java/com/example/solidconnection/chat/service/ChatService.java index c12d770b9..3394ae536 100644 --- a/src/main/java/com/example/solidconnection/chat/service/ChatService.java +++ b/src/main/java/com/example/solidconnection/chat/service/ChatService.java @@ -12,8 +12,8 @@ import com.example.solidconnection.chat.domain.ChatMessageOutbox; import com.example.solidconnection.chat.domain.ChatParticipant; import com.example.solidconnection.chat.domain.ChatRoom; -import com.example.solidconnection.chat.dto.ChatAttachmentResponse; import com.example.solidconnection.chat.dto.ChatImageSendRequest; +import com.example.solidconnection.chat.dto.ChatMessageHistoryResult; import com.example.solidconnection.chat.dto.ChatMessageResponse; import com.example.solidconnection.chat.dto.ChatMessageSendRequest; import com.example.solidconnection.chat.dto.ChatParticipantResponse; @@ -35,18 +35,15 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.function.Function; -import java.util.stream.Collectors; +import lombok.extern.slf4j.Slf4j; import org.springframework.context.ApplicationEventPublisher; import org.springframework.data.domain.Pageable; -import org.springframework.data.domain.Slice; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.StringUtils; @Service +@Slf4j public class ChatService { private final ChatRoomRepository chatRoomRepository; @@ -58,6 +55,7 @@ public class ChatService { private final MentorRepository mentorRepository; private final ApplicationEventPublisher eventPublisher; + private final ChatMessageHistoryQueryService chatMessageHistoryQueryService; private final ChatMessageOutboxTransactionService outboxTransactionService; public ChatService(ChatRoomRepository chatRoomRepository, @@ -68,6 +66,7 @@ public ChatService(ChatRoomRepository chatRoomRepository, SiteUserRepository siteUserRepository, MentorRepository mentorRepository, ApplicationEventPublisher eventPublisher, + ChatMessageHistoryQueryService chatMessageHistoryQueryService, ChatMessageOutboxTransactionService outboxTransactionService) { this.chatRoomRepository = chatRoomRepository; this.chatMessageRepository = chatMessageRepository; @@ -77,6 +76,7 @@ public ChatService(ChatRoomRepository chatRoomRepository, this.siteUserRepository = siteUserRepository; this.mentorRepository = mentorRepository; this.eventPublisher = eventPublisher; + this.chatMessageHistoryQueryService = chatMessageHistoryQueryService; this.outboxTransactionService = outboxTransactionService; } @@ -128,47 +128,20 @@ private ChatRoomResponse createChatRoomResponse(ChatRoom chatRoom, long siteUser ); } - @Transactional(readOnly = true) public SliceResponse getChatMessages(long siteUserId, long roomId, Pageable pageable) { - ChatParticipant participant = getChatParticipant(siteUserId, roomId); - - Slice chatMessages = chatMessageRepository.findByRoomIdWithPaging(roomId, pageable); - - Map participantIdToParticipant = buildParticipantIdToParticipantMap(chatMessages); - List content = buildChatMessageResponses(chatMessages, participantIdToParticipant); - outboxTransactionService.markHistoryServed( - participant.getId(), - chatMessages.getContent().stream().map(ChatMessage::getId).toList(), - now() - ); - - return SliceResponse.of(content, chatMessages); - } - - // senderId(chatParticipantId)로 chatParticipant 맵 생성 - private Map buildParticipantIdToParticipantMap(Slice chatMessages) { - Set participantIds = chatMessages.getContent().stream() - .map(ChatMessage::getSenderId) - .collect(Collectors.toSet()); - - return chatParticipantRepository.findAllById(participantIds).stream() - .collect(Collectors.toMap(ChatParticipant::getId, Function.identity())); - } - - private List buildChatMessageResponses( - Slice chatMessages, - Map participantIdToParticipant - ) { - return chatMessages.getContent().stream() - .map(message -> { - ChatParticipant senderParticipant = participantIdToParticipant.get(message.getSenderId()); - if (senderParticipant == null) { - throw new CustomException(CHAT_PARTICIPANT_NOT_FOUND); - } - long externalSenderId = senderParticipant.getSiteUserId(); - return toChatMessageResponse(message, externalSenderId); - }) - .toList(); + ChatMessageHistoryResult result = + chatMessageHistoryQueryService.getChatMessages(siteUserId, roomId, pageable); + try { + outboxTransactionService.markHistoryServed( + result.participantId(), + result.messageIds(), + now() + ); + } catch (RuntimeException e) { + log.error("채팅 내역 조회 Outbox 완료 기록에 실패했습니다. participantId={}, roomId={}", + result.participantId(), roomId, e); + } + return result.response(); } @Transactional(readOnly = true) @@ -201,26 +174,6 @@ private ChatParticipant getChatParticipant(long siteUserId, long roomId) { .orElseThrow(() -> new CustomException(CHAT_PARTICIPANT_NOT_FOUND)); } - private ChatMessageResponse toChatMessageResponse(ChatMessage message, long externalSenderId) { - List attachments = message.getChatAttachments().stream() - .map(attachment -> ChatAttachmentResponse.of( - attachment.getId(), - attachment.getIsImage(), - attachment.getUrl(), - attachment.getThumbnailUrl(), - attachment.getCreatedAt() - )) - .toList(); - - return ChatMessageResponse.of( - message.getId(), - message.getContent(), - externalSenderId, - message.getCreatedAt(), - attachments - ); - } - @Transactional public void markChatMessagesAsRead(long siteUserId, long roomId) { ChatParticipant participant = chatParticipantRepository diff --git a/src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java b/src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java new file mode 100644 index 000000000..95070d484 --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java @@ -0,0 +1,49 @@ +package com.example.solidconnection.chat.config; + +import static org.assertj.core.api.Assertions.assertThat; + +import jakarta.validation.Validation; +import jakarta.validation.Validator; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +@DisplayName("채팅 메시지 Outbox 설정 테스트") +class ChatMessageOutboxPropertiesTest { + + private Validator validator; + + @BeforeEach + void setUp() { + validator = Validation.buildDefaultValidatorFactory().getValidator(); + } + + @Nested + class 양수_설정을_검증한다 { + + @Test + void 최대_시도_횟수가_0이면_검증에_실패한다() { + // given + ChatMessageOutboxProperties properties = new ChatMessageOutboxProperties(); + properties.setMaxAttempts(0); + + // when & then + assertThat(validator.validate(properties)) + .extracting(violation -> violation.getPropertyPath().toString()) + .containsExactly("maxAttempts"); + } + + @Test + void 배치_크기가_0이면_검증에_실패한다() { + // given + ChatMessageOutboxProperties properties = new ChatMessageOutboxProperties(); + properties.setBatchSize(0); + + // when & then + assertThat(validator.validate(properties)) + .extracting(violation -> violation.getPropertyPath().toString()) + .containsExactly("batchSize"); + } + } +} diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java index 504928e96..2a527626f 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java @@ -23,6 +23,7 @@ import java.time.ZonedDateTime; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.mockito.BDDMockito; @@ -72,52 +73,56 @@ void setUp() { outbox = outboxRepository.save(new ChatMessageOutbox(message, recipient, now.minusSeconds(1))); } - @Test - void DB_트랜잭션_밖에서_방송하고_성공_상태를_별도_커밋한다() { - // given - BDDMockito.willAnswer(invocation -> { - assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isFalse(); - return null; - }).given(simpMessagingTemplate).convertAndSend( - BDDMockito.anyString(), - BDDMockito.any(ChatMessageSendResponse.class) - ); - - // when - dispatcher.dispatch(outbox.getId()); - - // then - ArgumentCaptor responseCaptor = - ArgumentCaptor.forClass(ChatMessageSendResponse.class); - BDDMockito.then(simpMessagingTemplate).should().convertAndSend( - BDDMockito.eq("/topic/chat/" + roomId), - responseCaptor.capture() - ); - ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); - assertAll( - () -> assertThat(responseCaptor.getValue().messageId()) - .isEqualTo(outbox.getChatMessage().getId()), - () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), - () -> assertThat(saved.getCompletionType()) - .isEqualTo(ChatMessageOutboxCompletionType.BROKER_DISPATCHED) - ); - } - - @Test - void 방송에_실패하면_예외를_전파하지_않고_재시도를_예약한다() { - // given - BDDMockito.willThrow(new RuntimeException("broadcast failed")) - .given(simpMessagingTemplate) - .convertAndSend(BDDMockito.anyString(), BDDMockito.any(ChatMessageSendResponse.class)); - - // when & then - assertThatCode(() -> dispatcher.dispatch(outbox.getId())).doesNotThrowAnyException(); - - ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); - assertAll( - () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.PENDING), - () -> assertThat(saved.getAttemptCount()).isEqualTo(1), - () -> assertThat(saved.getLastError()).contains("broadcast failed") - ); + @Nested + class 메시지를_방송한다 { + + @Test + void DB_트랜잭션_밖에서_방송하고_성공_상태를_별도_커밋한다() { + // given + BDDMockito.willAnswer(invocation -> { + assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isFalse(); + return null; + }).given(simpMessagingTemplate).convertAndSend( + BDDMockito.anyString(), + BDDMockito.any(ChatMessageSendResponse.class) + ); + + // when + dispatcher.dispatch(outbox.getId()); + + // then + ArgumentCaptor responseCaptor = + ArgumentCaptor.forClass(ChatMessageSendResponse.class); + BDDMockito.then(simpMessagingTemplate).should().convertAndSend( + BDDMockito.eq("/topic/chat/" + roomId), + responseCaptor.capture() + ); + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(responseCaptor.getValue().messageId()) + .isEqualTo(outbox.getChatMessage().getId()), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), + () -> assertThat(saved.getCompletionType()) + .isEqualTo(ChatMessageOutboxCompletionType.BROKER_DISPATCHED) + ); + } + + @Test + void 방송에_실패하면_예외를_전파하지_않고_재시도를_예약한다() { + // given + BDDMockito.willThrow(new RuntimeException("broadcast failed")) + .given(simpMessagingTemplate) + .convertAndSend(BDDMockito.anyString(), BDDMockito.any(ChatMessageSendResponse.class)); + + // when & then + assertThatCode(() -> dispatcher.dispatch(outbox.getId())).doesNotThrowAnyException(); + + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.PENDING), + () -> assertThat(saved.getAttemptCount()).isEqualTo(1), + () -> assertThat(saved.getLastError()).contains("broadcast failed") + ); + } } } diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxSchedulerTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxSchedulerTest.java index f4af8007c..630bb2dec 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxSchedulerTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxSchedulerTest.java @@ -10,6 +10,7 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import org.mockito.InOrder; import org.mockito.Mockito; @DisplayName("채팅 메시지 Outbox 스케줄러 테스트") @@ -34,8 +35,9 @@ class ChatMessageOutboxSchedulerTest { // then ArgumentCaptor cutoffCaptor = ArgumentCaptor.forClass(ZonedDateTime.class); ArgumentCaptor nowCaptor = ArgumentCaptor.forClass(ZonedDateTime.class); - then(transactionService).should().recoverStaleProcessing(cutoffCaptor.capture(), nowCaptor.capture()); - then(transactionService).should().findDispatchableIds(nowCaptor.getValue()); + InOrder inOrder = Mockito.inOrder(transactionService); + inOrder.verify(transactionService).recoverStaleProcessing(cutoffCaptor.capture(), nowCaptor.capture()); + inOrder.verify(transactionService).findDispatchableIds(nowCaptor.getValue()); then(dispatcher).should().dispatch(1L); then(dispatcher).should().dispatch(2L); org.assertj.core.api.Assertions.assertThat(Duration.between( diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java index c10fb5926..6b60fd78c 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java @@ -21,11 +21,13 @@ import com.example.solidconnection.support.TestContainerSpringBootTest; import java.time.ZonedDateTime; import java.util.List; +import java.util.Optional; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; @TestContainerSpringBootTest @DisplayName("채팅 메시지 Outbox 트랜잭션 서비스 테스트") @@ -49,6 +51,9 @@ class ChatMessageOutboxTransactionServiceTest { @Autowired private ChatMessageFixture chatMessageFixture; + @Autowired + private JdbcTemplate jdbcTemplate; + private ChatParticipant recipient; private ChatMessageOutbox outbox; private ZonedDateTime now; @@ -92,6 +97,28 @@ class 처리권을_획득한다 { // when & then assertThat(outboxTransactionService.claim(outbox.getId(), now)).isEmpty(); } + + @Test + void 삭제되어_방송할_수_없는_메시지는_FAILED로_종료한다() { + // given + jdbcTemplate.update( + "UPDATE chat_message SET is_deleted = true WHERE id = ?", + outbox.getChatMessage().getId() + ); + + // when + Optional claimed = + outboxTransactionService.claim(outbox.getId(), now); + + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(claimed).isEmpty(), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.FAILED), + () -> assertThat(saved.getAttemptCount()).isEqualTo(1), + () -> assertThat(saved.getLastError()).contains("message unavailable") + ); + } } @Nested @@ -146,6 +173,50 @@ class 처리_결과를_기록한다 { ); } + @Test + void 실패_메시지가_1000자를_넘으면_잘라서_기록한다() { + // given + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim(outbox.getId(), now) + .orElseThrow(); + + // when + boolean updated = outboxTransactionService.recordFailure( + outbox.getId(), + claimed.processingToken(), + "a".repeat(1001), + now.plusSeconds(1) + ); + + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(updated).isTrue(), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.PENDING), + () -> assertThat(saved.getLastError()).hasSize(1000) + ); + } + + @Test + void 다음_처리권을_획득해도_직전_실패_원인을_보존한다() { + // given + ClaimedChatMessageOutbox firstClaim = outboxTransactionService.claim(outbox.getId(), now) + .orElseThrow(); + outboxTransactionService.recordFailure( + outbox.getId(), + firstClaim.processingToken(), + "first failure", + now + ); + ZonedDateTime nextAttemptAt = outboxRepository.findById(outbox.getId()).orElseThrow().getNextAttemptAt(); + + // when + outboxTransactionService.claim(outbox.getId(), nextAttemptAt).orElseThrow(); + + // then + assertThat(outboxRepository.findById(outbox.getId()).orElseThrow().getLastError()) + .isEqualTo("first failure"); + } + @Test void 최대_시도_횟수에_도달하면_FAILED로_종료한다() { // given @@ -248,7 +319,24 @@ class 오래된_처리를_복구한다 { class 내역_조회로_완료한다 { @Test - void 처리_중인_메시지도_HISTORY_SERVED로_완료하고_늦은_worker가_되돌리지_못한다() { + void 방송을_한_번도_시도하지_않은_메시지는_내역_조회로_완료하지_않는다() { + // when + int completedCount = outboxTransactionService.markHistoryServed( + recipient.getId(), + List.of(outbox.getChatMessage().getId()), + now + ); + + // then + assertAll( + () -> assertThat(completedCount).isZero(), + () -> assertThat(outboxRepository.findById(outbox.getId()).orElseThrow().getStatus()) + .isEqualTo(ChatMessageOutboxStatus.PENDING) + ); + } + + @Test + void 처리_중인_메시지는_내역_조회로_완료하지_않아_송신자_방송을_보존한다() { // given ClaimedChatMessageOutbox claimed = outboxTransactionService.claim(outbox.getId(), now) .orElseThrow(); @@ -259,18 +347,51 @@ class 내역_조회로_완료한다 { List.of(outbox.getChatMessage().getId()), now.plusSeconds(1) ); - boolean staleUpdate = outboxTransactionService.recordFailure( + boolean workerCompleted = outboxTransactionService.markCompleted( outbox.getId(), claimed.processingToken(), - "late failure", + ChatMessageOutboxCompletionType.BROKER_DISPATCHED, now.plusSeconds(2) ); + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(completedCount).isZero(), + () -> assertThat(workerCompleted).isTrue(), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), + () -> assertThat(saved.getCompletionType()) + .isEqualTo(ChatMessageOutboxCompletionType.BROKER_DISPATCHED) + ); + } + + @Test + void 재시도_상한에_도달한_메시지는_HISTORY_SERVED로_완료한다() { + // given + ZonedDateTime attemptTime = now; + for (int attempt = 1; attempt <= 3; attempt++) { + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim(outbox.getId(), attemptTime) + .orElseThrow(); + outboxTransactionService.recordFailure( + outbox.getId(), + claimed.processingToken(), + "failure-" + attempt, + attemptTime + ); + attemptTime = outboxRepository.findById(outbox.getId()).orElseThrow().getNextAttemptAt(); + } + + // when + int completedCount = outboxTransactionService.markHistoryServed( + recipient.getId(), + List.of(outbox.getChatMessage().getId()), + now.plusMinutes(1) + ); + // then ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); assertAll( () -> assertThat(completedCount).isEqualTo(1), - () -> assertThat(staleUpdate).isFalse(), () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), () -> assertThat(saved.getCompletionType()) .isEqualTo(ChatMessageOutboxCompletionType.HISTORY_SERVED) diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatServiceHistoryTransactionTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatServiceHistoryTransactionTest.java new file mode 100644 index 000000000..9a99b8e3c --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/service/ChatServiceHistoryTransactionTest.java @@ -0,0 +1,95 @@ +package com.example.solidconnection.chat.service; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.BDDMockito.given; + +import com.example.solidconnection.chat.domain.ChatParticipant; +import com.example.solidconnection.chat.domain.ChatRoom; +import com.example.solidconnection.chat.fixture.ChatMessageFixture; +import com.example.solidconnection.chat.fixture.ChatParticipantFixture; +import com.example.solidconnection.chat.fixture.ChatRoomFixture; +import com.example.solidconnection.siteuser.domain.SiteUser; +import com.example.solidconnection.siteuser.fixture.SiteUserFixture; +import com.example.solidconnection.support.TestContainerSpringBootTest; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.domain.PageRequest; +import org.springframework.test.context.bean.override.mockito.MockitoBean; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +@TestContainerSpringBootTest +@DisplayName("채팅 내역 조회 트랜잭션 테스트") +class ChatServiceHistoryTransactionTest { + + @Autowired + private ChatService chatService; + + @Autowired + private SiteUserFixture siteUserFixture; + + @Autowired + private ChatRoomFixture chatRoomFixture; + + @Autowired + private ChatParticipantFixture chatParticipantFixture; + + @Autowired + private ChatMessageFixture chatMessageFixture; + + @MockitoBean + private ChatMessageOutboxTransactionService outboxTransactionService; + + @Test + void 조회_트랜잭션을_종료한_뒤_Outbox_완료를_기록한다() { + // given + SiteUser user = siteUserFixture.사용자(1, "user"); + SiteUser partner = siteUserFixture.사용자(2, "partner"); + ChatRoom chatRoom = chatRoomFixture.채팅방(false); + ChatParticipant userParticipant = chatParticipantFixture.참여자(user.getId(), chatRoom); + ChatParticipant partnerParticipant = chatParticipantFixture.참여자(partner.getId(), chatRoom); + chatMessageFixture.메시지("메시지", partnerParticipant.getId(), chatRoom); + given(outboxTransactionService.markHistoryServed( + org.mockito.ArgumentMatchers.eq(userParticipant.getId()), + org.mockito.ArgumentMatchers.anyList(), + org.mockito.ArgumentMatchers.any() + )).willAnswer(invocation -> { + assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isFalse(); + return 1; + }); + + // when + chatService.getChatMessages(user.getId(), chatRoom.getId(), PageRequest.of(0, 20)); + + // then + org.mockito.BDDMockito.then(outboxTransactionService).should().markHistoryServed( + org.mockito.ArgumentMatchers.eq(userParticipant.getId()), + org.mockito.ArgumentMatchers.anyList(), + org.mockito.ArgumentMatchers.any() + ); + } + + @Test + void Outbox_완료_기록에_실패해도_조회_응답은_반환한다() { + // given + SiteUser user = siteUserFixture.사용자(1, "user"); + SiteUser partner = siteUserFixture.사용자(2, "partner"); + ChatRoom chatRoom = chatRoomFixture.채팅방(false); + ChatParticipant userParticipant = chatParticipantFixture.참여자(user.getId(), chatRoom); + ChatParticipant partnerParticipant = chatParticipantFixture.참여자(partner.getId(), chatRoom); + chatMessageFixture.메시지("메시지", partnerParticipant.getId(), chatRoom); + given(outboxTransactionService.markHistoryServed( + org.mockito.ArgumentMatchers.eq(userParticipant.getId()), + org.mockito.ArgumentMatchers.anyList(), + org.mockito.ArgumentMatchers.any() + )).willThrow(new RuntimeException("state update failed")); + + // when & then + assertThatCode(() -> chatService.getChatMessages( + user.getId(), + chatRoom.getId(), + PageRequest.of(0, 20) + )).doesNotThrowAnyException(); + } +} diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java index 8436fb06c..e92b900e4 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java @@ -21,6 +21,7 @@ import com.example.solidconnection.chat.dto.ChatMessageSendResponse; import com.example.solidconnection.chat.dto.ChatParticipantResponse; import com.example.solidconnection.chat.dto.ChatRoomListResponse; +import com.example.solidconnection.chat.dto.ClaimedChatMessageOutbox; import com.example.solidconnection.chat.fixture.ChatAttachmentFixture; import com.example.solidconnection.chat.fixture.ChatMessageFixture; import com.example.solidconnection.chat.fixture.ChatParticipantFixture; @@ -68,6 +69,9 @@ class ChatServiceTest { @Autowired private ChatMessageOutboxRepository chatMessageOutboxRepository; + @Autowired + private ChatMessageOutboxTransactionService outboxTransactionService; + @Autowired private TransactionTemplate transactionTemplate; @@ -335,11 +339,7 @@ void setUp() { void 조회_응답에_포함된_상대방_메시지의_Outbox를_HISTORY_SERVED로_완료한다() { // given ChatMessage message = chatMessageFixture.메시지("조회 복구 메시지", partnerParticipant.getId(), chatRoom); - ChatMessageOutbox outbox = chatMessageOutboxRepository.save(new ChatMessageOutbox( - message, - participant, - ZonedDateTime.now().minusMinutes(1) - )); + ChatMessageOutbox outbox = failedOutbox(message); // when chatService.getChatMessages(user.getId(), chatRoom.getId(), pageable); @@ -353,6 +353,80 @@ void setUp() { ); } + @Test + void 조회_페이지에_포함되지_않은_메시지의_Outbox는_PENDING으로_유지한다() { + // given + ChatMessage olderMessage = chatMessageFixture.메시지( + "페이지 밖 메시지", + partnerParticipant.getId(), + chatRoom + ); + ChatMessageOutbox olderOutbox = pendingOutbox(olderMessage); + ChatMessage newerMessage = chatMessageFixture.메시지( + "조회된 메시지", + partnerParticipant.getId(), + chatRoom + ); + ChatMessageOutbox newerOutbox = failedOutbox(newerMessage); + Pageable firstMessageOnly = PageRequest.of(0, 1, Sort.by(Sort.Direction.DESC, "createdAt")); + + // when + chatService.getChatMessages(user.getId(), chatRoom.getId(), firstMessageOnly); + + // then + ChatMessageOutbox savedOlder = chatMessageOutboxRepository.findById(olderOutbox.getId()).orElseThrow(); + ChatMessageOutbox savedNewer = chatMessageOutboxRepository.findById(newerOutbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(savedOlder.getStatus()).isEqualTo(ChatMessageOutboxStatus.PENDING), + () -> assertThat(savedNewer.getStatus()).isEqualTo(ChatMessageOutboxStatus.COMPLETED), + () -> assertThat(savedNewer.getCompletionType()) + .isEqualTo(ChatMessageOutboxCompletionType.HISTORY_SERVED) + ); + } + + private ChatMessageOutbox failedOutbox(ChatMessage message) { + ChatMessageOutbox outbox = chatMessageOutboxRepository.save(new ChatMessageOutbox( + message, + participant, + ZonedDateTime.now().minusMinutes(1) + )); + ZonedDateTime attemptTime = ZonedDateTime.now(); + for (int attempt = 1; attempt <= 3; attempt++) { + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim( + outbox.getId(), + attemptTime + ).orElseThrow(); + outboxTransactionService.recordFailure( + outbox.getId(), + claimed.processingToken(), + "broadcast failed", + attemptTime + ); + attemptTime = chatMessageOutboxRepository.findById(outbox.getId()).orElseThrow().getNextAttemptAt(); + } + return outbox; + } + + private ChatMessageOutbox pendingOutbox(ChatMessage message) { + ChatMessageOutbox outbox = chatMessageOutboxRepository.save(new ChatMessageOutbox( + message, + participant, + ZonedDateTime.now().minusMinutes(1) + )); + ZonedDateTime attemptTime = ZonedDateTime.now(); + ClaimedChatMessageOutbox claimed = outboxTransactionService.claim( + outbox.getId(), + attemptTime + ).orElseThrow(); + outboxTransactionService.recordFailure( + outbox.getId(), + claimed.processingToken(), + "broadcast failed", + attemptTime + ); + return outbox; + } + @Test void 송신자의_조회는_상대방_Outbox를_완료하지_않는다() { // given From 256203de6228c67803375f6dce0df54aba6d2731 Mon Sep 17 00:00:00 2001 From: Hexeong <123macanic@naver.com> Date: Mon, 2 Nov 2026 20:31:36 +0900 Subject: [PATCH 3/8] =?UTF-8?q?fix:=20=EC=B1=84=ED=8C=85=20Outbox=20?= =?UTF-8?q?=EB=94=94=EC=8A=A4=ED=8C=A8=EC=B9=98=EB=A5=BC=20=EB=B9=84?= =?UTF-8?q?=EB=8F=99=EA=B8=B0=EB=A1=9C=20=EB=B6=84=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ChatMessageBroadcastEventListener.java | 2 ++ .../common/config/sync/AsyncConfig.java | 9 +++++++ .../DiscordNotificationEventListener.java | 2 -- .../resources/config/application-variable.yml | 2 +- ...ChatMessageBroadcastEventListenerTest.java | 17 ++++++++++++ .../chat/service/ChatServiceTest.java | 13 +++++++--- .../common/config/sync/AsyncConfigTest.java | 26 +++++++++++++++++++ 7 files changed, 64 insertions(+), 7 deletions(-) create mode 100644 src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java diff --git a/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListener.java b/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListener.java index 24273eb23..1e3f73436 100644 --- a/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListener.java +++ b/src/main/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListener.java @@ -2,6 +2,7 @@ import com.example.solidconnection.chat.service.ChatMessageOutboxDispatcher; import lombok.RequiredArgsConstructor; +import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; import org.springframework.transaction.event.TransactionPhase; import org.springframework.transaction.event.TransactionalEventListener; @@ -12,6 +13,7 @@ public class ChatMessageBroadcastEventListener { private final ChatMessageOutboxDispatcher dispatcher; + @Async("chatMessageOutboxExecutor") @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) public void broadcast(ChatMessageBroadcastEvent event) { dispatcher.dispatch(event.outboxId()); diff --git a/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java b/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java index 852e3de02..9853ab171 100644 --- a/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java +++ b/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java @@ -2,9 +2,13 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.core.task.TaskExecutor; +import org.springframework.core.task.VirtualThreadTaskExecutor; +import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @Configuration +@EnableAsync public class AsyncConfig { @Bean(name = "asyncExecutor") @@ -19,4 +23,9 @@ public ThreadPoolTaskExecutor asyncExecutor() { executor.initialize(); return executor; } + + @Bean(defaultCandidate = false) + public TaskExecutor chatMessageOutboxExecutor() { + return new VirtualThreadTaskExecutor("chat-message-outbox-"); + } } diff --git a/src/main/java/com/example/solidconnection/common/discord/DiscordNotificationEventListener.java b/src/main/java/com/example/solidconnection/common/discord/DiscordNotificationEventListener.java index 4642ab326..aa1235ed9 100644 --- a/src/main/java/com/example/solidconnection/common/discord/DiscordNotificationEventListener.java +++ b/src/main/java/com/example/solidconnection/common/discord/DiscordNotificationEventListener.java @@ -6,12 +6,10 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Async; -import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.stereotype.Component; import org.springframework.transaction.event.TransactionalEventListener; @Component -@EnableAsync @RequiredArgsConstructor @Slf4j public class DiscordNotificationEventListener { diff --git a/src/main/resources/config/application-variable.yml b/src/main/resources/config/application-variable.yml index 5c07ddcc2..90d12ed4d 100644 --- a/src/main/resources/config/application-variable.yml +++ b/src/main/resources/config/application-variable.yml @@ -11,7 +11,7 @@ chat: max-retry-delay: 1m processing-timeout: 1m scheduling: - enabled: false + enabled: true delay: 5000 websocket: diff --git a/src/test/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListenerTest.java b/src/test/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListenerTest.java index bfb6c1e03..8a96b84a2 100644 --- a/src/test/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListenerTest.java +++ b/src/test/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListenerTest.java @@ -1,15 +1,32 @@ package com.example.solidconnection.chat.event; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.BDDMockito.then; import com.example.solidconnection.chat.service.ChatMessageOutboxDispatcher; +import java.lang.reflect.Method; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.mockito.Mockito; +import org.springframework.scheduling.annotation.Async; @DisplayName("채팅 메시지 방송 이벤트 리스너 테스트") class ChatMessageBroadcastEventListenerTest { + @Test + void 송신_트랜잭션과_분리된_전용_실행기에서_디스패치한다() throws NoSuchMethodException { + // given + Method broadcast = ChatMessageBroadcastEventListener.class + .getDeclaredMethod("broadcast", ChatMessageBroadcastEvent.class); + + // when + Async async = broadcast.getAnnotation(Async.class); + + // then + assertThat(async).isNotNull(); + assertThat(async.value()).isEqualTo("chatMessageOutboxExecutor"); + } + @Test void Outbox_식별자를_디스패처에_전달한다() { // given diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java index e92b900e4..ff5193ab1 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java @@ -5,6 +5,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.AssertionsForClassTypes.assertThatCode; import static org.junit.jupiter.api.Assertions.assertAll; +import static org.mockito.Mockito.timeout; import com.example.solidconnection.chat.domain.ChatAttachment; import com.example.solidconnection.chat.domain.ChatMessage; @@ -571,7 +572,8 @@ public void beforeCommit(boolean readOnly) { ArgumentCaptor destinationCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor payloadCaptor = ArgumentCaptor.forClass(ChatMessageSendResponse.class); - BDDMockito.verify(simpMessagingTemplate).convertAndSend(destinationCaptor.capture(), payloadCaptor.capture()); + BDDMockito.verify(simpMessagingTemplate, timeout(5000)) + .convertAndSend(destinationCaptor.capture(), payloadCaptor.capture()); assertAll( () -> assertThat(destinationCaptor.getValue()).isEqualTo("/topic/chat/" + chatRoom.getId()), @@ -663,7 +665,8 @@ public void beforeCommit(boolean readOnly) { ArgumentCaptor destinationCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor payloadCaptor = ArgumentCaptor.forClass(ChatMessageSendResponse.class); - BDDMockito.verify(simpMessagingTemplate).convertAndSend(destinationCaptor.capture(), payloadCaptor.capture()); + BDDMockito.verify(simpMessagingTemplate, timeout(5000)) + .convertAndSend(destinationCaptor.capture(), payloadCaptor.capture()); ChatMessageSendResponse response = payloadCaptor.getValue(); @@ -716,7 +719,8 @@ public void beforeCommit(boolean readOnly) { // then ArgumentCaptor payloadCaptor = ArgumentCaptor.forClass(ChatMessageSendResponse.class); - BDDMockito.verify(simpMessagingTemplate).convertAndSend(BDDMockito.anyString(), payloadCaptor.capture()); + BDDMockito.verify(simpMessagingTemplate, timeout(5000)) + .convertAndSend(BDDMockito.anyString(), payloadCaptor.capture()); ChatMessageSendResponse response = payloadCaptor.getValue(); @@ -753,7 +757,8 @@ public void beforeCommit(boolean readOnly) { // then ArgumentCaptor payloadCaptor = ArgumentCaptor.forClass(ChatMessageSendResponse.class); - BDDMockito.verify(simpMessagingTemplate).convertAndSend(BDDMockito.anyString(), payloadCaptor.capture()); + BDDMockito.verify(simpMessagingTemplate, timeout(5000)) + .convertAndSend(BDDMockito.anyString(), payloadCaptor.capture()); ChatMessageSendResponse response = payloadCaptor.getValue(); diff --git a/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java b/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java new file mode 100644 index 000000000..fde3f0c38 --- /dev/null +++ b/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java @@ -0,0 +1,26 @@ +package com.example.solidconnection.common.config.sync; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.core.task.VirtualThreadTaskExecutor; + +@DisplayName("비동기 실행기 설정 테스트") +class AsyncConfigTest { + + @Test + void 채팅_Outbox는_전용_가상_스레드_실행기를_사용한다() { + // given + try (AnnotationConfigApplicationContext context = + new AnnotationConfigApplicationContext(AsyncConfig.class)) { + + // when + Object taskExecutor = context.getBean("chatMessageOutboxExecutor"); + + // then + assertThat(taskExecutor).isInstanceOf(VirtualThreadTaskExecutor.class); + } + } +} From 979ff760a815baa7155af1c45d70bee870c3ef6f Mon Sep 17 00:00:00 2001 From: Hexeong <123macanic@naver.com> Date: Mon, 2 Nov 2026 21:03:00 +0900 Subject: [PATCH 4/8] =?UTF-8?q?fix:=20=EC=B1=84=ED=8C=85=20Outbox=20?= =?UTF-8?q?=EC=9A=B4=EC=98=81=20=EC=95=88=EC=A0=84=EC=84=B1=20=EB=B3=B4?= =?UTF-8?q?=EC=99=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../config/ChatMessageOutboxProperties.java | 8 ++ .../common/config/sync/AsyncConfig.java | 16 +++- .../resources/config/application-variable.yml | 2 +- .../ChatMessageOutboxPropertiesTest.java | 78 +++++++++++++++++++ .../common/config/sync/AsyncConfigTest.java | 15 +++- 5 files changed, 110 insertions(+), 9 deletions(-) diff --git a/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java b/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java index bbf81ef35..01bc08978 100644 --- a/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java +++ b/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java @@ -1,9 +1,11 @@ package com.example.solidconnection.chat.config; +import jakarta.validation.constraints.NotNull; import jakarta.validation.constraints.Positive; import java.time.Duration; import lombok.Getter; import lombok.Setter; +import org.hibernate.validator.constraints.time.DurationMin; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; import org.springframework.validation.annotation.Validated; @@ -19,7 +21,13 @@ public class ChatMessageOutboxProperties { private int maxAttempts = 5; @Positive private int batchSize = 100; + @NotNull + @DurationMin(nanos = 1) private Duration initialRetryDelay = Duration.ofSeconds(1); + @NotNull + @DurationMin(nanos = 1) private Duration maxRetryDelay = Duration.ofMinutes(1); + @NotNull + @DurationMin(nanos = 1) private Duration processingTimeout = Duration.ofMinutes(1); } diff --git a/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java b/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java index 9853ab171..90eb7c1b3 100644 --- a/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java +++ b/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java @@ -1,9 +1,8 @@ package com.example.solidconnection.common.config.sync; +import java.util.concurrent.ThreadPoolExecutor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.core.task.TaskExecutor; -import org.springframework.core.task.VirtualThreadTaskExecutor; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @@ -25,7 +24,16 @@ public ThreadPoolTaskExecutor asyncExecutor() { } @Bean(defaultCandidate = false) - public TaskExecutor chatMessageOutboxExecutor() { - return new VirtualThreadTaskExecutor("chat-message-outbox-"); + public ThreadPoolTaskExecutor chatMessageOutboxExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(2); + executor.setMaxPoolSize(4); + executor.setQueueCapacity(100); + executor.setThreadNamePrefix("chat-message-outbox-"); + executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy()); + executor.setWaitForTasksToCompleteOnShutdown(true); + executor.setAwaitTerminationSeconds(10); + executor.initialize(); + return executor; } } diff --git a/src/main/resources/config/application-variable.yml b/src/main/resources/config/application-variable.yml index 90d12ed4d..5c07ddcc2 100644 --- a/src/main/resources/config/application-variable.yml +++ b/src/main/resources/config/application-variable.yml @@ -11,7 +11,7 @@ chat: max-retry-delay: 1m processing-timeout: 1m scheduling: - enabled: true + enabled: false delay: 5000 websocket: diff --git a/src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java b/src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java index 95070d484..fc2033ef6 100644 --- a/src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java +++ b/src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java @@ -4,10 +4,15 @@ import jakarta.validation.Validation; import jakarta.validation.Validator; +import java.io.IOException; +import java.time.Duration; +import java.util.Objects; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.springframework.boot.env.YamlPropertySourceLoader; +import org.springframework.core.io.ClassPathResource; @DisplayName("채팅 메시지 Outbox 설정 테스트") class ChatMessageOutboxPropertiesTest { @@ -45,5 +50,78 @@ class 양수_설정을_검증한다 { .extracting(violation -> violation.getPropertyPath().toString()) .containsExactly("batchSize"); } + + @Test + void 초기_재시도_지연이_0이면_검증에_실패한다() { + // given + ChatMessageOutboxProperties properties = new ChatMessageOutboxProperties(); + properties.setInitialRetryDelay(Duration.ZERO); + + // when & then + assertThat(validator.validate(properties)) + .extracting(violation -> violation.getPropertyPath().toString()) + .containsExactly("initialRetryDelay"); + } + + @Test + void 최대_재시도_지연이_0이면_검증에_실패한다() { + // given + ChatMessageOutboxProperties properties = new ChatMessageOutboxProperties(); + properties.setMaxRetryDelay(Duration.ZERO); + + // when & then + assertThat(validator.validate(properties)) + .extracting(violation -> violation.getPropertyPath().toString()) + .containsExactly("maxRetryDelay"); + } + + @Test + void 처리_타임아웃이_0이면_검증에_실패한다() { + // given + ChatMessageOutboxProperties properties = new ChatMessageOutboxProperties(); + properties.setProcessingTimeout(Duration.ZERO); + + // when & then + assertThat(validator.validate(properties)) + .extracting(violation -> violation.getPropertyPath().toString()) + .containsExactly("processingTimeout"); + } + + @Test + void Duration_설정이_null이면_검증에_실패한다() { + // given + ChatMessageOutboxProperties properties = new ChatMessageOutboxProperties(); + properties.setInitialRetryDelay(null); + properties.setMaxRetryDelay(null); + properties.setProcessingTimeout(null); + + // when & then + assertThat(validator.validate(properties)) + .extracting(violation -> violation.getPropertyPath().toString()) + .containsExactlyInAnyOrder("initialRetryDelay", "maxRetryDelay", "processingTimeout"); + } + } + + @Nested + class 배포_기본값을_검증한다 { + + @Test + void 클라이언트_멱등_처리_전에는_재시도_스케줄러를_비활성화한다() throws IOException { + // given + YamlPropertySourceLoader loader = new YamlPropertySourceLoader(); + + // when + Object enabled = loader.load( + "application-variable", + new ClassPathResource("config/application-variable.yml") + ).stream() + .map(source -> source.getProperty("chat.outbox.scheduling.enabled")) + .filter(Objects::nonNull) + .findFirst() + .orElseThrow(); + + // then + assertThat(enabled).isEqualTo(false); + } } } diff --git a/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java b/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java index fde3f0c38..65952395b 100644 --- a/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java +++ b/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java @@ -2,25 +2,32 @@ import static org.assertj.core.api.Assertions.assertThat; +import java.util.concurrent.ThreadPoolExecutor; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import org.springframework.core.task.VirtualThreadTaskExecutor; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @DisplayName("비동기 실행기 설정 테스트") class AsyncConfigTest { @Test - void 채팅_Outbox는_전용_가상_스레드_실행기를_사용한다() { + void 채팅_Outbox_실행기는_동시_실행과_대기_작업에_상한을_둔다() { // given try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(AsyncConfig.class)) { // when - Object taskExecutor = context.getBean("chatMessageOutboxExecutor"); + ThreadPoolTaskExecutor taskExecutor = context.getBean( + "chatMessageOutboxExecutor", + ThreadPoolTaskExecutor.class + ); // then - assertThat(taskExecutor).isInstanceOf(VirtualThreadTaskExecutor.class); + assertThat(taskExecutor.getMaxPoolSize()).isEqualTo(4); + assertThat(taskExecutor.getThreadPoolExecutor().getQueue().remainingCapacity()).isEqualTo(100); + assertThat(taskExecutor.getThreadPoolExecutor().getRejectedExecutionHandler()) + .isInstanceOf(ThreadPoolExecutor.DiscardPolicy.class); } } } From 14f04512c8fd67b2990b5c94a459c65bd7e0e602 Mon Sep 17 00:00:00 2001 From: Hexeong <123macanic@naver.com> Date: Mon, 2 Nov 2026 21:27:18 +0900 Subject: [PATCH 5/8] =?UTF-8?q?fix:=20Outbox=20=ED=8F=AC=ED=99=94=20?= =?UTF-8?q?=EC=8B=9C=20=EB=94=94=EC=8A=A4=ED=8C=A8=EC=B9=98=20=EC=9C=A0?= =?UTF-8?q?=EC=8B=A4=20=EB=B0=A9=EC=A7=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/ChatMessageOutboxDispatcher.java | 13 +++ .../common/config/sync/AsyncConfig.java | 16 ++-- ...essageOutboxDispatcherConcurrencyTest.java | 88 +++++++++++++++++++ .../common/config/sync/AsyncConfigTest.java | 14 ++- 4 files changed, 111 insertions(+), 20 deletions(-) create mode 100644 src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java index 9fd857e99..c77827d78 100644 --- a/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java @@ -7,6 +7,7 @@ import com.example.solidconnection.chat.dto.ClaimedChatMessageOutbox; import java.time.ZonedDateTime; import java.util.Optional; +import java.util.concurrent.Semaphore; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.messaging.simp.SimpMessageSendingOperations; @@ -17,10 +18,22 @@ @Slf4j public class ChatMessageOutboxDispatcher { + private static final int MAX_CONCURRENT_DISPATCHES = 4; + private final ChatMessageOutboxTransactionService outboxTransactionService; private final SimpMessageSendingOperations simpMessageSendingOperations; + private final Semaphore dispatchPermits = new Semaphore(MAX_CONCURRENT_DISPATCHES, true); public void dispatch(long outboxId) { + dispatchPermits.acquireUninterruptibly(); + try { + dispatchWithPermit(outboxId); + } finally { + dispatchPermits.release(); + } + } + + private void dispatchWithPermit(long outboxId) { Optional claim; try { claim = outboxTransactionService.claim(outboxId, now()); diff --git a/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java b/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java index 90eb7c1b3..e87d602d6 100644 --- a/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java +++ b/src/main/java/com/example/solidconnection/common/config/sync/AsyncConfig.java @@ -1,8 +1,8 @@ package com.example.solidconnection.common.config.sync; -import java.util.concurrent.ThreadPoolExecutor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @@ -24,16 +24,10 @@ public ThreadPoolTaskExecutor asyncExecutor() { } @Bean(defaultCandidate = false) - public ThreadPoolTaskExecutor chatMessageOutboxExecutor() { - ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); - executor.setCorePoolSize(2); - executor.setMaxPoolSize(4); - executor.setQueueCapacity(100); - executor.setThreadNamePrefix("chat-message-outbox-"); - executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy()); - executor.setWaitForTasksToCompleteOnShutdown(true); - executor.setAwaitTerminationSeconds(10); - executor.initialize(); + public SimpleAsyncTaskExecutor chatMessageOutboxExecutor() { + SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("chat-message-outbox-"); + executor.setVirtualThreads(true); + executor.setTaskTerminationTimeout(10_000); return executor; } } diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java new file mode 100644 index 000000000..76941572e --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java @@ -0,0 +1,88 @@ +package com.example.solidconnection.chat.service; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import com.example.solidconnection.chat.dto.ChatMessageSendResponse; +import com.example.solidconnection.chat.dto.ClaimedChatMessageOutbox; +import com.example.solidconnection.chat.domain.MessageType; +import java.time.Duration; +import java.time.ZonedDateTime; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import org.mockito.BDDMockito; +import org.springframework.messaging.simp.SimpMessageSendingOperations; + +class ChatMessageOutboxDispatcherConcurrencyTest { + + @Test + void 동시_디스패치를_네_개로_제한하면서_대기_작업을_버리지_않는다() throws Exception { + // given + ChatMessageOutboxTransactionService transactionService = + BDDMockito.mock(ChatMessageOutboxTransactionService.class); + SimpMessageSendingOperations messagingOperations = + BDDMockito.mock(SimpMessageSendingOperations.class); + ChatMessageOutboxDispatcher dispatcher = + new ChatMessageOutboxDispatcher(transactionService, messagingOperations); + AtomicInteger claimedCount = new AtomicInteger(); + CountDownLatch fourBroadcastsStarted = new CountDownLatch(4); + CountDownLatch releaseBroadcasts = new CountDownLatch(1); + + BDDMockito.given(transactionService.claim(BDDMockito.anyLong(), BDDMockito.any(ZonedDateTime.class))) + .willAnswer(invocation -> { + long outboxId = invocation.getArgument(0); + claimedCount.incrementAndGet(); + return Optional.of(claimed(outboxId)); + }); + BDDMockito.willAnswer(invocation -> { + fourBroadcastsStarted.countDown(); + releaseBroadcasts.await(3, TimeUnit.SECONDS); + return null; + }).given(messagingOperations).convertAndSend( + BDDMockito.anyString(), + BDDMockito.any(ChatMessageSendResponse.class) + ); + BDDMockito.given(transactionService.markCompleted( + BDDMockito.anyLong(), + BDDMockito.anyString(), + BDDMockito.any(), + BDDMockito.any(ZonedDateTime.class) + )).willReturn(true); + + try (var executor = Executors.newFixedThreadPool(5)) { + // when + List> futures = java.util.stream.LongStream.rangeClosed(1, 5) + .mapToObj(outboxId -> executor.submit(() -> dispatcher.dispatch(outboxId))) + .toList(); + + // then + assertThat(fourBroadcastsStarted.await(1, TimeUnit.SECONDS)).isTrue(); + await().during(Duration.ofMillis(100)) + .atMost(Duration.ofSeconds(1)) + .untilAsserted(() -> assertThat(claimedCount).hasValue(4)); + + releaseBroadcasts.countDown(); + for (var future : futures) { + future.get(1, TimeUnit.SECONDS); + } + assertThat(claimedCount).hasValue(5); + } finally { + releaseBroadcasts.countDown(); + } + } + + private ClaimedChatMessageOutbox claimed(long outboxId) { + return new ClaimedChatMessageOutbox( + outboxId, + "token-" + outboxId, + 1, + 1, + new ChatMessageSendResponse(outboxId, "content", 1, MessageType.TEXT, List.of()) + ); + } +} diff --git a/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java b/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java index 65952395b..c2ab38e61 100644 --- a/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java +++ b/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java @@ -2,32 +2,28 @@ import static org.assertj.core.api.Assertions.assertThat; -import java.util.concurrent.ThreadPoolExecutor; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.core.task.SimpleAsyncTaskExecutor; @DisplayName("비동기 실행기 설정 테스트") class AsyncConfigTest { @Test - void 채팅_Outbox_실행기는_동시_실행과_대기_작업에_상한을_둔다() { + void 채팅_Outbox_실행기는_작업을_버리지_않는_가상_스레드_실행기를_사용한다() { // given try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(AsyncConfig.class)) { // when - ThreadPoolTaskExecutor taskExecutor = context.getBean( + Object taskExecutor = context.getBean( "chatMessageOutboxExecutor", - ThreadPoolTaskExecutor.class + Object.class ); // then - assertThat(taskExecutor.getMaxPoolSize()).isEqualTo(4); - assertThat(taskExecutor.getThreadPoolExecutor().getQueue().remainingCapacity()).isEqualTo(100); - assertThat(taskExecutor.getThreadPoolExecutor().getRejectedExecutionHandler()) - .isInstanceOf(ThreadPoolExecutor.DiscardPolicy.class); + assertThat(taskExecutor).isInstanceOf(SimpleAsyncTaskExecutor.class); } } } From bb3701628585f61e626d27e2d04c6ed9183be740 Mon Sep 17 00:00:00 2001 From: Hexeong <123macanic@naver.com> Date: Mon, 2 Nov 2026 22:03:29 +0900 Subject: [PATCH 6/8] =?UTF-8?q?refactor:=20=EC=B1=84=ED=8C=85=20Outbox=20?= =?UTF-8?q?=EC=BB=A8=EB=B2=A4=EC=85=98=20=EC=A0=95=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../chat/domain/ChatMessageOutbox.java | 17 +++---- .../ChatMessageHistoryQueryService.java | 6 ++- .../chat/service/ChatService.java | 44 ++++++++--------- .../fixture/ChatMessageOutboxFixture.java | 27 +++++++++++ .../ChatMessageOutboxFixtureBuilder.java | 48 +++++++++++++++++++ .../ChatMessageOutboxDispatcherTest.java | 6 ++- ...atMessageOutboxTransactionServiceTest.java | 6 ++- .../chat/service/ChatServiceTest.java | 16 ++++--- 8 files changed, 131 insertions(+), 39 deletions(-) create mode 100644 src/test/java/com/example/solidconnection/chat/fixture/ChatMessageOutboxFixture.java create mode 100644 src/test/java/com/example/solidconnection/chat/fixture/ChatMessageOutboxFixtureBuilder.java diff --git a/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java index 51237f55e..3f97ad37b 100644 --- a/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java +++ b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java @@ -18,6 +18,7 @@ import lombok.AccessLevel; import lombok.Getter; import lombok.NoArgsConstructor; +import org.hibernate.annotations.ColumnDefault; import org.hibernate.annotations.OnDelete; import org.hibernate.annotations.OnDeleteAction; @@ -63,25 +64,26 @@ public class ChatMessageOutbox extends BaseEntity { private ChatMessageOutboxStatus status; @Enumerated(EnumType.STRING) - @Column(name = "completion_type", length = 30) + @Column(name = "completion_type", nullable = true, length = 30) private ChatMessageOutboxCompletionType completionType; + @ColumnDefault("0") @Column(name = "attempt_count", nullable = false) - private int attemptCount; + private int attemptCount = 0; - @Column(name = "processing_token", length = 36) + @Column(name = "processing_token", nullable = true, length = 36) private String processingToken; - @Column(name = "processing_started_at") + @Column(name = "processing_started_at", nullable = true) private ZonedDateTime processingStartedAt; - @Column(name = "next_attempt_at") + @Column(name = "next_attempt_at", nullable = true) private ZonedDateTime nextAttemptAt; - @Column(name = "last_error", length = 1000) + @Column(name = "last_error", nullable = true, length = 1000) private String lastError; - @Column(name = "completed_at") + @Column(name = "completed_at", nullable = true) private ZonedDateTime completedAt; public ChatMessageOutbox( @@ -92,7 +94,6 @@ public ChatMessageOutbox( this.chatMessage = chatMessage; this.recipient = recipient; this.status = ChatMessageOutboxStatus.PENDING; - this.attemptCount = 0; this.nextAttemptAt = nextAttemptAt; } } diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatMessageHistoryQueryService.java b/src/main/java/com/example/solidconnection/chat/service/ChatMessageHistoryQueryService.java index 6241b6b9f..b0c9b5f79 100644 --- a/src/main/java/com/example/solidconnection/chat/service/ChatMessageHistoryQueryService.java +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageHistoryQueryService.java @@ -30,7 +30,11 @@ public class ChatMessageHistoryQueryService { private final ChatParticipantRepository chatParticipantRepository; @Transactional(readOnly = true) - public ChatMessageHistoryResult getChatMessages(long siteUserId, long roomId, Pageable pageable) { + public ChatMessageHistoryResult getChatMessages( + long siteUserId, + long roomId, + Pageable pageable + ) { ChatParticipant participant = chatParticipantRepository.findByChatRoomIdAndSiteUserId(roomId, siteUserId) .orElseThrow(() -> new CustomException(CHAT_PARTICIPANT_NOT_FOUND)); Slice chatMessages = chatMessageRepository.findByRoomIdWithPaging(roomId, pageable); diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatService.java b/src/main/java/com/example/solidconnection/chat/service/ChatService.java index 3394ae536..e3761e400 100644 --- a/src/main/java/com/example/solidconnection/chat/service/ChatService.java +++ b/src/main/java/com/example/solidconnection/chat/service/ChatService.java @@ -169,11 +169,6 @@ public void validateChatRoomParticipant(long siteUserId, long roomId) { getChatParticipant(siteUserId, roomId); } - private ChatParticipant getChatParticipant(long siteUserId, long roomId) { - return chatParticipantRepository.findByChatRoomIdAndSiteUserId(roomId, siteUserId) - .orElseThrow(() -> new CustomException(CHAT_PARTICIPANT_NOT_FOUND)); - } - @Transactional public void markChatMessagesAsRead(long siteUserId, long roomId) { ChatParticipant participant = chatParticipantRepository @@ -226,6 +221,28 @@ public void sendChatImage(ChatImageSendRequest chatImageSendRequest, long siteUs publishOutboxEvent(chatMessage, findOutboxRecipient(roomId, siteUserId)); } + private ChatParticipant getChatParticipant(long siteUserId, long roomId) { + return chatParticipantRepository.findByChatRoomIdAndSiteUserId(roomId, siteUserId) + .orElseThrow(() -> new CustomException(CHAT_PARTICIPANT_NOT_FOUND)); + } + + private String generateThumbnailUrl(String originalUrl) { + try { + String fileName = originalUrl.substring(originalUrl.lastIndexOf('/') + 1); + + String nameWithoutExt = fileName.substring(0, fileName.lastIndexOf('.')); + String extension = fileName.substring(fileName.lastIndexOf('.')); + + String thumbnailFileName = nameWithoutExt + "_thumb" + extension; + + return originalUrl.replace("chat/files/", "chat/thumbnails/") + .replace(fileName, thumbnailFileName); + + } catch (Exception e) { + return originalUrl; + } + } + private ChatParticipant findOutboxRecipient(long roomId, long senderSiteUserId) { List recipients = chatParticipantRepository.findAllByChatRoomIdAndSiteUserIdNot(roomId, senderSiteUserId); @@ -246,23 +263,6 @@ private ZonedDateTime now() { return ZonedDateTime.now(UTC).truncatedTo(MICROS); } - private String generateThumbnailUrl(String originalUrl) { - try { - String fileName = originalUrl.substring(originalUrl.lastIndexOf('/') + 1); - - String nameWithoutExt = fileName.substring(0, fileName.lastIndexOf('.')); - String extension = fileName.substring(fileName.lastIndexOf('.')); - - String thumbnailFileName = nameWithoutExt + "_thumb" + extension; - - return originalUrl.replace("chat/files/", "chat/thumbnails/") - .replace(fileName, thumbnailFileName); - - } catch (Exception e) { - return originalUrl; - } - } - @Transactional public Long createMentoringChatRoom(Long mentoringId, Long mentorId, Long menteeId) { ChatRoom existingChatRoom = chatRoomRepository.findByMentoringId(mentoringId); diff --git a/src/test/java/com/example/solidconnection/chat/fixture/ChatMessageOutboxFixture.java b/src/test/java/com/example/solidconnection/chat/fixture/ChatMessageOutboxFixture.java new file mode 100644 index 000000000..1241b211d --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/fixture/ChatMessageOutboxFixture.java @@ -0,0 +1,27 @@ +package com.example.solidconnection.chat.fixture; + +import com.example.solidconnection.chat.domain.ChatMessage; +import com.example.solidconnection.chat.domain.ChatMessageOutbox; +import com.example.solidconnection.chat.domain.ChatParticipant; +import java.time.ZonedDateTime; +import lombok.RequiredArgsConstructor; +import org.springframework.boot.test.context.TestComponent; + +@TestComponent +@RequiredArgsConstructor +public class ChatMessageOutboxFixture { + + private final ChatMessageOutboxFixtureBuilder chatMessageOutboxFixtureBuilder; + + public ChatMessageOutbox 아웃박스( + ChatMessage chatMessage, + ChatParticipant recipient, + ZonedDateTime nextAttemptAt + ) { + return chatMessageOutboxFixtureBuilder.chatMessageOutbox() + .chatMessage(chatMessage) + .recipient(recipient) + .nextAttemptAt(nextAttemptAt) + .create(); + } +} diff --git a/src/test/java/com/example/solidconnection/chat/fixture/ChatMessageOutboxFixtureBuilder.java b/src/test/java/com/example/solidconnection/chat/fixture/ChatMessageOutboxFixtureBuilder.java new file mode 100644 index 000000000..8d6df9ec5 --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/fixture/ChatMessageOutboxFixtureBuilder.java @@ -0,0 +1,48 @@ +package com.example.solidconnection.chat.fixture; + +import com.example.solidconnection.chat.domain.ChatMessage; +import com.example.solidconnection.chat.domain.ChatMessageOutbox; +import com.example.solidconnection.chat.domain.ChatParticipant; +import com.example.solidconnection.chat.repository.ChatMessageOutboxRepository; +import java.time.ZonedDateTime; +import lombok.RequiredArgsConstructor; +import org.springframework.boot.test.context.TestComponent; + +@TestComponent +@RequiredArgsConstructor +public class ChatMessageOutboxFixtureBuilder { + + private final ChatMessageOutboxRepository chatMessageOutboxRepository; + + private ChatMessage chatMessage; + private ChatParticipant recipient; + private ZonedDateTime nextAttemptAt; + + public ChatMessageOutboxFixtureBuilder chatMessageOutbox() { + return new ChatMessageOutboxFixtureBuilder(chatMessageOutboxRepository); + } + + public ChatMessageOutboxFixtureBuilder chatMessage(ChatMessage chatMessage) { + this.chatMessage = chatMessage; + return this; + } + + public ChatMessageOutboxFixtureBuilder recipient(ChatParticipant recipient) { + this.recipient = recipient; + return this; + } + + public ChatMessageOutboxFixtureBuilder nextAttemptAt(ZonedDateTime nextAttemptAt) { + this.nextAttemptAt = nextAttemptAt; + return this; + } + + public ChatMessageOutbox create() { + ChatMessageOutbox outbox = new ChatMessageOutbox( + chatMessage, + recipient, + nextAttemptAt + ); + return chatMessageOutboxRepository.save(outbox); + } +} diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java index 2a527626f..242976c51 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java @@ -14,6 +14,7 @@ import com.example.solidconnection.chat.domain.ChatRoom; import com.example.solidconnection.chat.dto.ChatMessageSendResponse; import com.example.solidconnection.chat.fixture.ChatMessageFixture; +import com.example.solidconnection.chat.fixture.ChatMessageOutboxFixture; import com.example.solidconnection.chat.fixture.ChatParticipantFixture; import com.example.solidconnection.chat.fixture.ChatRoomFixture; import com.example.solidconnection.chat.repository.ChatMessageOutboxRepository; @@ -54,6 +55,9 @@ class ChatMessageOutboxDispatcherTest { @Autowired private ChatMessageFixture chatMessageFixture; + @Autowired + private ChatMessageOutboxFixture chatMessageOutboxFixture; + @MockitoBean private SimpMessagingTemplate simpMessagingTemplate; @@ -70,7 +74,7 @@ void setUp() { ChatParticipant recipient = chatParticipantFixture.참여자(recipientUser.getId(), chatRoom); ChatMessage message = chatMessageFixture.메시지("메시지", sender.getId(), chatRoom); ZonedDateTime now = ZonedDateTime.now(UTC).truncatedTo(MICROS); - outbox = outboxRepository.save(new ChatMessageOutbox(message, recipient, now.minusSeconds(1))); + outbox = chatMessageOutboxFixture.아웃박스(message, recipient, now.minusSeconds(1)); } @Nested diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java index 6b60fd78c..557c04ea3 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java @@ -13,6 +13,7 @@ import com.example.solidconnection.chat.domain.ChatRoom; import com.example.solidconnection.chat.dto.ClaimedChatMessageOutbox; import com.example.solidconnection.chat.fixture.ChatMessageFixture; +import com.example.solidconnection.chat.fixture.ChatMessageOutboxFixture; import com.example.solidconnection.chat.fixture.ChatParticipantFixture; import com.example.solidconnection.chat.fixture.ChatRoomFixture; import com.example.solidconnection.chat.repository.ChatMessageOutboxRepository; @@ -51,6 +52,9 @@ class ChatMessageOutboxTransactionServiceTest { @Autowired private ChatMessageFixture chatMessageFixture; + @Autowired + private ChatMessageOutboxFixture chatMessageOutboxFixture; + @Autowired private JdbcTemplate jdbcTemplate; @@ -67,7 +71,7 @@ void setUp() { recipient = chatParticipantFixture.참여자(recipientUser.getId(), chatRoom); ChatMessage message = chatMessageFixture.메시지("메시지", sender.getId(), chatRoom); now = ZonedDateTime.now(UTC).truncatedTo(MICROS); - outbox = outboxRepository.save(new ChatMessageOutbox(message, recipient, now.minusMinutes(10))); + outbox = chatMessageOutboxFixture.아웃박스(message, recipient, now.minusMinutes(10)); } @Nested diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java index ff5193ab1..20d14a755 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java @@ -25,6 +25,7 @@ import com.example.solidconnection.chat.dto.ClaimedChatMessageOutbox; import com.example.solidconnection.chat.fixture.ChatAttachmentFixture; import com.example.solidconnection.chat.fixture.ChatMessageFixture; +import com.example.solidconnection.chat.fixture.ChatMessageOutboxFixture; import com.example.solidconnection.chat.fixture.ChatParticipantFixture; import com.example.solidconnection.chat.fixture.ChatReadStatusFixture; import com.example.solidconnection.chat.fixture.ChatRoomFixture; @@ -88,6 +89,9 @@ class ChatServiceTest { @Autowired private ChatMessageFixture chatMessageFixture; + @Autowired + private ChatMessageOutboxFixture chatMessageOutboxFixture; + @Autowired private ChatReadStatusFixture chatReadStatusFixture; @@ -386,11 +390,11 @@ void setUp() { } private ChatMessageOutbox failedOutbox(ChatMessage message) { - ChatMessageOutbox outbox = chatMessageOutboxRepository.save(new ChatMessageOutbox( + ChatMessageOutbox outbox = chatMessageOutboxFixture.아웃박스( message, participant, ZonedDateTime.now().minusMinutes(1) - )); + ); ZonedDateTime attemptTime = ZonedDateTime.now(); for (int attempt = 1; attempt <= 3; attempt++) { ClaimedChatMessageOutbox claimed = outboxTransactionService.claim( @@ -409,11 +413,11 @@ private ChatMessageOutbox failedOutbox(ChatMessage message) { } private ChatMessageOutbox pendingOutbox(ChatMessage message) { - ChatMessageOutbox outbox = chatMessageOutboxRepository.save(new ChatMessageOutbox( + ChatMessageOutbox outbox = chatMessageOutboxFixture.아웃박스( message, participant, ZonedDateTime.now().minusMinutes(1) - )); + ); ZonedDateTime attemptTime = ZonedDateTime.now(); ClaimedChatMessageOutbox claimed = outboxTransactionService.claim( outbox.getId(), @@ -432,11 +436,11 @@ private ChatMessageOutbox pendingOutbox(ChatMessage message) { void 송신자의_조회는_상대방_Outbox를_완료하지_않는다() { // given ChatMessage message = chatMessageFixture.메시지("내가 보낸 메시지", participant.getId(), chatRoom); - ChatMessageOutbox outbox = chatMessageOutboxRepository.save(new ChatMessageOutbox( + ChatMessageOutbox outbox = chatMessageOutboxFixture.아웃박스( message, partnerParticipant, ZonedDateTime.now().minusMinutes(1) - )); + ); // when chatService.getChatMessages(user.getId(), chatRoom.getId(), pageable); From 8941cb7f97251b6d382b6096f17eac60819f87ae Mon Sep 17 00:00:00 2001 From: Hexeong <123macanic@naver.com> Date: Mon, 2 Nov 2026 22:32:24 +0900 Subject: [PATCH 7/8] =?UTF-8?q?fix:=20Outbox=20=EB=94=94=EC=8A=A4=ED=8C=A8?= =?UTF-8?q?=EC=B9=98=20=EC=A2=85=EB=A3=8C=20=EC=9D=B8=ED=84=B0=EB=9F=BD?= =?UTF-8?q?=ED=8A=B8=20=EC=B2=98=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/ChatMessageOutboxDispatcher.java | 8 ++++++- ...essageOutboxDispatcherConcurrencyTest.java | 24 +++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java index c77827d78..0e6036a43 100644 --- a/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java @@ -25,7 +25,13 @@ public class ChatMessageOutboxDispatcher { private final Semaphore dispatchPermits = new Semaphore(MAX_CONCURRENT_DISPATCHES, true); public void dispatch(long outboxId) { - dispatchPermits.acquireUninterruptibly(); + try { + dispatchPermits.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.info("채팅 메시지 Outbox 디스패치 대기 중 인터럽트되었습니다. outboxId={}", outboxId); + return; + } try { dispatchWithPermit(outboxId); } finally { diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java index 76941572e..1603befc6 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java @@ -20,6 +20,30 @@ class ChatMessageOutboxDispatcherConcurrencyTest { + @Test + void 인터럽트된_스레드에서_디스패치하면_Outbox를_claim하지_않는다() { + // given + ChatMessageOutboxTransactionService transactionService = + BDDMockito.mock(ChatMessageOutboxTransactionService.class); + SimpMessageSendingOperations messagingOperations = + BDDMockito.mock(SimpMessageSendingOperations.class); + ChatMessageOutboxDispatcher dispatcher = + new ChatMessageOutboxDispatcher(transactionService, messagingOperations); + Thread.currentThread().interrupt(); + + try { + // when + dispatcher.dispatch(1L); + + // then + assertThat(Thread.currentThread().isInterrupted()).isTrue(); + BDDMockito.then(transactionService).shouldHaveNoInteractions(); + BDDMockito.then(messagingOperations).shouldHaveNoInteractions(); + } finally { + Thread.interrupted(); + } + } + @Test void 동시_디스패치를_네_개로_제한하면서_대기_작업을_버리지_않는다() throws Exception { // given From 62db10e41a62fbeb99ce10cf9ee7099c85882df3 Mon Sep 17 00:00:00 2001 From: Hexeong <123macanic@naver.com> Date: Tue, 3 Nov 2026 11:41:24 +0900 Subject: [PATCH 8/8] =?UTF-8?q?test:=20Outbox=20=EA=B2=BD=ED=95=A9=20?= =?UTF-8?q?=EC=A1=B0=EA=B1=B4=20=EA=B2=80=EC=A6=9D=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...atMessageOutboxTransactionServiceTest.java | 78 +++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java index 557c04ea3..da8db706c 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java @@ -21,8 +21,13 @@ import com.example.solidconnection.siteuser.fixture.SiteUserFixture; import com.example.solidconnection.support.TestContainerSpringBootTest; import java.time.ZonedDateTime; +import java.util.ArrayList; import java.util.List; import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Nested; @@ -102,6 +107,51 @@ class 처리권을_획득한다 { assertThat(outboxTransactionService.claim(outbox.getId(), now)).isEmpty(); } + @Test + void 동일한_Outbox를_동시에_claim하면_하나의_worker만_처리권을_획득한다() throws Exception { + // given + CountDownLatch workersReady = new CountDownLatch(2); + CountDownLatch start = new CountDownLatch(1); + + try (var executor = Executors.newFixedThreadPool(2)) { + try { + List>> claims = List.of( + executor.submit(() -> claimAfterStart(workersReady, start)), + executor.submit(() -> claimAfterStart(workersReady, start)) + ); + boolean allWorkersReady = workersReady.await(5, TimeUnit.SECONDS); + + // when + start.countDown(); + + // then + assertThat(allWorkersReady).isTrue(); + List> claimResults = new ArrayList<>(); + for (Future> claim : claims) { + claimResults.add(claim.get(10, TimeUnit.SECONDS)); + } + long claimedCount = claimResults.stream().filter(Optional::isPresent).count(); + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(claimedCount).isEqualTo(1), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.PROCESSING), + () -> assertThat(saved.getAttemptCount()).isEqualTo(1) + ); + } finally { + start.countDown(); + } + } + } + + private Optional claimAfterStart( + CountDownLatch workersReady, + CountDownLatch start + ) throws InterruptedException { + workersReady.countDown(); + start.await(); + return outboxTransactionService.claim(outbox.getId(), now); + } + @Test void 삭제되어_방송할_수_없는_메시지는_FAILED로_종료한다() { // given @@ -317,6 +367,34 @@ class 오래된_처리를_복구한다 { .isEqualTo(ChatMessageOutboxCompletionType.BROKER_DISPATCHED) ); } + + @Test + void 회수된_이전_worker는_재획득된_Outbox를_완료할_수_없다() { + // given + ZonedDateTime firstAttemptAt = now.minusMinutes(2); + ClaimedChatMessageOutbox firstClaim = outboxTransactionService.claim(outbox.getId(), firstAttemptAt) + .orElseThrow(); + outboxTransactionService.recoverStaleProcessing(now.minusMinutes(1), now); + ClaimedChatMessageOutbox secondClaim = outboxTransactionService.claim(outbox.getId(), now) + .orElseThrow(); + + // when + boolean staleUpdate = outboxTransactionService.markCompleted( + outbox.getId(), + firstClaim.processingToken(), + ChatMessageOutboxCompletionType.BROKER_DISPATCHED, + now.plusSeconds(1) + ); + + // then + ChatMessageOutbox saved = outboxRepository.findById(outbox.getId()).orElseThrow(); + assertAll( + () -> assertThat(staleUpdate).isFalse(), + () -> assertThat(saved.getStatus()).isEqualTo(ChatMessageOutboxStatus.PROCESSING), + () -> assertThat(saved.getProcessingToken()).isEqualTo(secondClaim.processingToken()), + () -> assertThat(saved.getCompletionType()).isNull() + ); + } } @Nested