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..01bc08978 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/config/ChatMessageOutboxProperties.java @@ -0,0 +1,33 @@ +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; + +@Component +@Validated +@ConfigurationProperties(prefix = "chat.outbox") +@Getter +@Setter +public class ChatMessageOutboxProperties { + + @Positive + 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/chat/domain/ChatMessageOutbox.java b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java new file mode 100644 index 000000000..3f97ad37b --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/domain/ChatMessageOutbox.java @@ -0,0 +1,99 @@ +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.ColumnDefault; +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", nullable = true, length = 30) + private ChatMessageOutboxCompletionType completionType; + + @ColumnDefault("0") + @Column(name = "attempt_count", nullable = false) + private int attemptCount = 0; + + @Column(name = "processing_token", nullable = true, length = 36) + private String processingToken; + + @Column(name = "processing_started_at", nullable = true) + private ZonedDateTime processingStartedAt; + + @Column(name = "next_attempt_at", nullable = true) + private ZonedDateTime nextAttemptAt; + + @Column(name = "last_error", nullable = true, length = 1000) + private String lastError; + + @Column(name = "completed_at", nullable = true) + private ZonedDateTime completedAt; + + public ChatMessageOutbox( + ChatMessage chatMessage, + ChatParticipant recipient, + ZonedDateTime nextAttemptAt + ) { + this.chatMessage = chatMessage; + this.recipient = recipient; + this.status = ChatMessageOutboxStatus.PENDING; + 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/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/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..1e3f73436 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,21 @@ 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.scheduling.annotation.Async; 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; + @Async("chatMessageOutboxExecutor") @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..f2a7cf4c9 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/repository/ChatMessageOutboxRepository.java @@ -0,0 +1,172 @@ +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.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/ChatMessageHistoryQueryService.java b/src/main/java/com/example/solidconnection/chat/service/ChatMessageHistoryQueryService.java new file mode 100644 index 000000000..b0c9b5f79 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageHistoryQueryService.java @@ -0,0 +1,92 @@ +package com.example.solidconnection.chat.service; + +import static com.example.solidconnection.common.exception.ErrorCode.CHAT_PARTICIPANT_NOT_FOUND; + +import com.example.solidconnection.chat.domain.ChatMessage; +import com.example.solidconnection.chat.domain.ChatParticipant; +import com.example.solidconnection.chat.dto.ChatAttachmentResponse; +import com.example.solidconnection.chat.dto.ChatMessageHistoryResult; +import com.example.solidconnection.chat.dto.ChatMessageResponse; +import com.example.solidconnection.chat.repository.ChatMessageRepository; +import com.example.solidconnection.chat.repository.ChatParticipantRepository; +import com.example.solidconnection.common.dto.SliceResponse; +import com.example.solidconnection.common.exception.CustomException; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; +import lombok.RequiredArgsConstructor; +import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Slice; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +@Service +@RequiredArgsConstructor +public class ChatMessageHistoryQueryService { + + private final ChatMessageRepository chatMessageRepository; + private final ChatParticipantRepository chatParticipantRepository; + + @Transactional(readOnly = true) + 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); + 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/ChatMessageOutboxDispatcher.java b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java new file mode 100644 index 000000000..0e6036a43 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcher.java @@ -0,0 +1,103 @@ +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 java.util.concurrent.Semaphore; +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 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) { + try { + dispatchPermits.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.info("채팅 메시지 Outbox 디스패치 대기 중 인터럽트되었습니다. outboxId={}", outboxId); + return; + } + try { + dispatchWithPermit(outboxId); + } finally { + dispatchPermits.release(); + } + } + + private void dispatchWithPermit(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..65704d042 --- /dev/null +++ b/src/main/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionService.java @@ -0,0 +1,187 @@ +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(); + } + + 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()); + + 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.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..e3761e400 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.ChatMessageHistoryResult; 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,46 +31,53 @@ 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; -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; 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 ChatMessageHistoryQueryService chatMessageHistoryQueryService; + private final ChatMessageOutboxTransactionService outboxTransactionService; public ChatService(ChatRoomRepository chatRoomRepository, ChatMessageRepository chatMessageRepository, + ChatMessageOutboxRepository chatMessageOutboxRepository, ChatParticipantRepository chatParticipantRepository, ChatReadStatusRepository chatReadStatusRepository, SiteUserRepository siteUserRepository, MentorRepository mentorRepository, - ApplicationEventPublisher eventPublisher) { + ApplicationEventPublisher eventPublisher, + ChatMessageHistoryQueryService chatMessageHistoryQueryService, + 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.chatMessageHistoryQueryService = chatMessageHistoryQueryService; + this.outboxTransactionService = outboxTransactionService; } @Transactional(readOnly = true) @@ -118,42 +128,20 @@ private ChatRoomResponse createChatRoomResponse(ChatRoom chatRoom, long siteUser ); } - @Transactional(readOnly = true) public SliceResponse getChatMessages(long siteUserId, long roomId, Pageable pageable) { - validateChatRoomParticipant(siteUserId, roomId); - - Slice chatMessages = chatMessageRepository.findByRoomIdWithPaging(roomId, pageable); - - Map participantIdToParticipant = buildParticipantIdToParticipantMap(chatMessages); - List content = buildChatMessageResponses(chatMessages, participantIdToParticipant); - - 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) @@ -178,30 +166,7 @@ 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); - } - } - - 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 - ); + getChatParticipant(siteUserId, roomId); } @Transactional @@ -215,34 +180,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 +218,12 @@ public void sendChatImage(ChatImageSendRequest chatImageSendRequest, long siteUs } chatMessageRepository.save(chatMessage); + publishOutboxEvent(chatMessage, findOutboxRecipient(roomId, siteUserId)); + } - ChatMessageSendResponse chatMessageResponse = ChatMessageSendResponse.of(chatMessage, siteUserId); - eventPublisher.publishEvent(new ChatMessageBroadcastEvent(roomId, chatMessageResponse)); + private ChatParticipant getChatParticipant(long siteUserId, long roomId) { + return chatParticipantRepository.findByChatRoomIdAndSiteUserId(roomId, siteUserId) + .orElseThrow(() -> new CustomException(CHAT_PARTICIPANT_NOT_FOUND)); } private String generateThumbnailUrl(String originalUrl) { @@ -281,6 +243,26 @@ private String generateThumbnailUrl(String originalUrl) { } } + 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())); + } + + private ZonedDateTime now() { + return ZonedDateTime.now(UTC).truncatedTo(MICROS); + } + @Transactional public Long createMentoringChatRoom(Long mentoringId, Long mentorId, Long menteeId) { ChatRoom existingChatRoom = chatRoomRepository.findByMentoringId(mentoringId); 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..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 @@ -2,9 +2,12 @@ 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; @Configuration +@EnableAsync public class AsyncConfig { @Bean(name = "asyncExecutor") @@ -19,4 +22,12 @@ public ThreadPoolTaskExecutor asyncExecutor() { executor.initialize(); return executor; } + + @Bean(defaultCandidate = false) + public SimpleAsyncTaskExecutor chatMessageOutboxExecutor() { + SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("chat-message-outbox-"); + executor.setVirtualThreads(true); + executor.setTaskTerminationTimeout(10_000); + return executor; + } } 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 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/config/ChatMessageOutboxPropertiesTest.java b/src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java new file mode 100644 index 000000000..fc2033ef6 --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/config/ChatMessageOutboxPropertiesTest.java @@ -0,0 +1,127 @@ +package com.example.solidconnection.chat.config; + +import static org.assertj.core.api.Assertions.assertThat; + +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 { + + 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"); + } + + @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/chat/event/ChatMessageBroadcastEventListenerTest.java b/src/test/java/com/example/solidconnection/chat/event/ChatMessageBroadcastEventListenerTest.java index 2cbad42c1..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,79 +1,43 @@ 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 java.lang.reflect.Method; 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; +import org.springframework.scheduling.annotation.Async; @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); + @Test + void 송신_트랜잭션과_분리된_전용_실행기에서_디스패치한다() throws NoSuchMethodException { + // given + Method broadcast = ChatMessageBroadcastEventListener.class + .getDeclaredMethod("broadcast", ChatMessageBroadcastEvent.class); - logger = (Logger) LoggerFactory.getLogger(ChatMessageBroadcastEventListener.class); - listAppender = new ListAppender<>(); - listAppender.start(); - logger.addAppender(listAppender); - } + // when + Async async = broadcast.getAnnotation(Async.class); - @AfterEach - void tearDown() { - logger.detachAppender(listAppender); - listAppender.stop(); + // then + assertThat(async).isNotNull(); + assertThat(async.value()).isEqualTo("chatMessageOutboxExecutor"); } @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/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/ChatMessageOutboxDispatcherConcurrencyTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java new file mode 100644 index 000000000..1603befc6 --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherConcurrencyTest.java @@ -0,0 +1,112 @@ +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 인터럽트된_스레드에서_디스패치하면_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 + 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/chat/service/ChatMessageOutboxDispatcherTest.java b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java new file mode 100644 index 000000000..242976c51 --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxDispatcherTest.java @@ -0,0 +1,132 @@ +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.ChatMessageOutboxFixture; +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.Nested; +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; + + @Autowired + private ChatMessageOutboxFixture chatMessageOutboxFixture; + + @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 = chatMessageOutboxFixture.아웃박스(message, recipient, now.minusSeconds(1)); + } + + @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 new file mode 100644 index 000000000..630bb2dec --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxSchedulerTest.java @@ -0,0 +1,48 @@ +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.InOrder; +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); + 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( + 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..da8db706c --- /dev/null +++ b/src/test/java/com/example/solidconnection/chat/service/ChatMessageOutboxTransactionServiceTest.java @@ -0,0 +1,506 @@ +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.ChatMessageOutboxFixture; +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.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; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; + +@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; + + @Autowired + private ChatMessageOutboxFixture chatMessageOutboxFixture; + + @Autowired + private JdbcTemplate jdbcTemplate; + + 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 = chatMessageOutboxFixture.아웃박스(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(); + } + + @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 + 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 + 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 실패_메시지가_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 + 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) + ); + } + + @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 + class 내역_조회로_완료한다 { + + @Test + 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(); + + // when + int completedCount = outboxTransactionService.markHistoryServed( + recipient.getId(), + List.of(outbox.getChatMessage().getId()), + now.plusSeconds(1) + ); + boolean workerCompleted = outboxTransactionService.markCompleted( + outbox.getId(), + claimed.processingToken(), + 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(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/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 88c88a493..20d14a755 100644 --- a/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java +++ b/src/test/java/com/example/solidconnection/chat/service/ChatServiceTest.java @@ -5,9 +5,13 @@ 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; +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; @@ -18,12 +22,15 @@ 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.ChatMessageOutboxFixture; import com.example.solidconnection.chat.fixture.ChatParticipantFixture; 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 +68,12 @@ class ChatServiceTest { @Autowired private ChatMessageRepository chatMessageRepository; + @Autowired + private ChatMessageOutboxRepository chatMessageOutboxRepository; + + @Autowired + private ChatMessageOutboxTransactionService outboxTransactionService; + @Autowired private TransactionTemplate transactionTemplate; @@ -76,6 +89,9 @@ class ChatServiceTest { @Autowired private ChatMessageFixture chatMessageFixture; + @Autowired + private ChatMessageOutboxFixture chatMessageOutboxFixture; + @Autowired private ChatReadStatusFixture chatReadStatusFixture; @@ -219,12 +235,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 +339,116 @@ 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 = failedOutbox(message); + + // 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는_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 = chatMessageOutboxFixture.아웃박스( + 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 = chatMessageOutboxFixture.아웃박스( + 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 + ChatMessage message = chatMessageFixture.메시지("내가 보낸 메시지", participant.getId(), chatRoom); + ChatMessageOutbox outbox = chatMessageOutboxFixture.아웃박스( + 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 +542,7 @@ class 채팅_메시지를_전송한다 { private SiteUser sender; private ChatParticipant senderParticipant; + private ChatParticipant recipientParticipant; private ChatRoom chatRoom; @BeforeEach @@ -421,6 +550,7 @@ void setUp() { sender = siteUserFixture.사용자(111, "sender"); chatRoom = chatRoomFixture.채팅방(false); senderParticipant = chatParticipantFixture.참여자(sender.getId(), chatRoom); + recipientParticipant = chatParticipantFixture.참여자(mentor1.getId(), chatRoom); } @Test @@ -446,13 +576,17 @@ 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()), () -> 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 +606,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 +632,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 +640,7 @@ void setUp() { sender = siteUserFixture.사용자(111, "sender"); chatRoom = chatRoomFixture.채팅방(false); senderParticipant = chatParticipantFixture.참여자(sender.getId(), chatRoom); + recipientParticipant = chatParticipantFixture.참여자(mentor1.getId(), chatRoom); } @Test @@ -532,7 +669,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(); @@ -544,7 +682,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 +705,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() ); } @@ -581,7 +723,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(); @@ -618,7 +761,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..c2ab38e61 --- /dev/null +++ b/src/test/java/com/example/solidconnection/common/config/sync/AsyncConfigTest.java @@ -0,0 +1,29 @@ +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.SimpleAsyncTaskExecutor; + +@DisplayName("비동기 실행기 설정 테스트") +class AsyncConfigTest { + + @Test + void 채팅_Outbox_실행기는_작업을_버리지_않는_가상_스레드_실행기를_사용한다() { + // given + try (AnnotationConfigApplicationContext context = + new AnnotationConfigApplicationContext(AsyncConfig.class)) { + + // when + Object taskExecutor = context.getBean( + "chatMessageOutboxExecutor", + Object.class + ); + + // then + assertThat(taskExecutor).isInstanceOf(SimpleAsyncTaskExecutor.class); + } + } +} 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: