diff --git a/src/main/java/com/jobdri/jobdri_api/JobdriApiApplication.java b/src/main/java/com/jobdri/jobdri_api/JobdriApiApplication.java index 897d2d4..b96a502 100644 --- a/src/main/java/com/jobdri/jobdri_api/JobdriApiApplication.java +++ b/src/main/java/com/jobdri/jobdri_api/JobdriApiApplication.java @@ -4,10 +4,12 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.retry.annotation.EnableRetry; import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication @EnableAsync @EnableRetry +@EnableScheduling public class JobdriApiApplication { public static void main(String[] args) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java index 23b0585..c9985a3 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java @@ -4,6 +4,7 @@ import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncSubmitResponse; import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisResponse; import com.jobdri.jobdri_api.domain.analysis.service.AnalysisAsyncFacadeService; +import com.jobdri.jobdri_api.domain.analysis.service.AnalysisAsyncSseService; import com.jobdri.jobdri_api.domain.analysis.service.AnalysisService; import com.jobdri.jobdri_api.global.apiPayload.ApiResponse; import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode; @@ -13,11 +14,13 @@ import io.swagger.v3.oas.annotations.tags.Tag; import lombok.RequiredArgsConstructor; import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @RestController @RequiredArgsConstructor @@ -27,6 +30,7 @@ public class AnalysisController { private final AnalysisService analysisService; private final AnalysisAsyncFacadeService analysisAsyncFacadeService; + private final AnalysisAsyncSseService analysisAsyncSseService; @Operation(summary = "자소서 분석 비동기 실행", description = "저장된 문항 답변과 공고 정보를 기반으로 자소서 분석 작업을 접수합니다.") @PostMapping @@ -53,6 +57,20 @@ public ApiResponse getAnalysisTask( ); } + @Operation(summary = "자소서 분석 비동기 작업 상태 SSE 구독", description = "taskId로 자소서 분석 비동기 작업 상태를 SSE 스트림으로 구독합니다.") + @GetMapping(value = "/async/{taskId}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter streamAnalysisTask( + @AuthenticationPrincipal UserDetailsImpl userDetails, + @PathVariable Long mockApplyId, + @PathVariable String taskId + ) { + var user = getAuthenticatedUser(userDetails); + return analysisAsyncSseService.subscribe( + taskId, + () -> analysisAsyncFacadeService.getTask(user, mockApplyId, taskId) + ); + } + @Operation(summary = "자소서 분석 결과 조회", description = "저장된 자소서 분석 결과를 조회합니다.") @GetMapping public ApiResponse getAnalysis( diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.java index 045c904..4426619 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.java @@ -5,6 +5,7 @@ import org.springframework.data.jpa.repository.JpaRepository; import java.util.Collection; +import java.util.List; import java.util.Optional; public interface AnalysisAsyncTaskRepository extends JpaRepository { @@ -15,4 +16,6 @@ Optional findFirstByUserIdAndMockApplyIdAndStatusInOrderByCre Long mockApplyId, Collection statuses ); + + List findByStatusIn(Collection statuses); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAdminDebugService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAdminDebugService.java index f7bb3f2..d073b10 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAdminDebugService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAdminDebugService.java @@ -19,6 +19,7 @@ @Service @RequiredArgsConstructor @Transactional(readOnly = true) +// 관리자용으로 분석 참조 데이터와 retrieval 결과를 미리 점검하는 서비스다. public class AnalysisAdminDebugService { private final MockApplyRepository mockApplyRepository; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.java index 9f8e4ed..d58015a 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.java @@ -24,6 +24,7 @@ @Component @Slf4j @RequiredArgsConstructor +// 자소서 분석에 필요한 프롬프트를 만들고 LLM 호출을 수행하는 클라이언트다. public class AnalysisAiClient { private static final int MAX_REFERENCE_SECTION_LENGTH = 3000; private static final int MAX_REFERENCE_FIELD_LENGTH = 300; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.java index 3fa23e0..53d9678 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.java @@ -15,6 +15,7 @@ @Service @RequiredArgsConstructor +// 분석 비동기 작업의 접수와 상태 조회를 외부 API 관점에서 조율하는 서비스다. public class AnalysisAsyncFacadeService { private static final String ACTIVE_TASK_UNIQUE_CONSTRAINT = "uk_analysis_async_tasks_active_user_mock_apply"; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncProcessor.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncProcessor.java index 3b632c0..e113255 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncProcessor.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncProcessor.java @@ -6,6 +6,7 @@ @Service @RequiredArgsConstructor +// 분석 비동기 작업을 MQ 메시지로 변환해 워커 실행 경로로 넘기는 서비스다. public class AnalysisAsyncProcessor { private final AnalysisTaskMessagePublisher analysisTaskMessagePublisher; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java new file mode 100644 index 0000000..895359a --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java @@ -0,0 +1,46 @@ +package com.jobdri.jobdri_api.domain.analysis.service; + +import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncStatusResponse; +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.TaskStatus; +import com.jobdri.jobdri_api.global.sse.SseSubscriptionRegistry; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.function.Supplier; + +@Service +@RequiredArgsConstructor +public class AnalysisAsyncSseService { + + private static final String EVENT_NAME = "analysis-status"; + + private final SseSubscriptionRegistry sseSubscriptionRegistry; + + public SseEmitter subscribe(String taskId, Supplier initialStatusSupplier) { + return sseSubscriptionRegistry.subscribe( + channelKey(taskId), + EVENT_NAME, + initialStatusSupplier, + this::isTerminal + ); + } + + public void publish(AnalysisAsyncStatusResponse statusResponse) { + sseSubscriptionRegistry.publish( + channelKey(statusResponse.taskId()), + EVENT_NAME, + statusResponse, + isTerminal(statusResponse) + ); + } + + private String channelKey(String taskId) { + return "analysis:" + taskId; + } + + private boolean isTerminal(AnalysisAsyncStatusResponse statusResponse) { + TaskStatus status = TaskStatus.valueOf(statusResponse.status()); + return status == TaskStatus.SUCCEEDED || status == TaskStatus.FAILED; + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java new file mode 100644 index 0000000..50b7b85 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java @@ -0,0 +1,110 @@ +package com.jobdri.jobdri_api.domain.analysis.service; + +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.CreditStatus; +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.FailureReason; +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.TaskStatus; +import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; +import com.jobdri.jobdri_api.domain.user.entity.User; +import com.jobdri.jobdri_api.domain.user.service.UserService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; + +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.EnumSet; + +@Slf4j +@Service +@RequiredArgsConstructor +public class AnalysisAsyncSweepService { + + private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; + private final AnalysisAsyncTaskService analysisAsyncTaskService; + private final AnalysisService analysisService; + private final UserService userService; + private final TransactionTemplate transactionTemplate; + + @Value("${app.worker.analysis.queue-timeout-minutes:10}") + private long queueTimeoutMinutes; + + @Value("${app.worker.analysis.processing-timeout-minutes:20}") + private long processingTimeoutMinutes; + + public int sweepTimedOutTasks() { + int expiredCount = 0; + for (AnalysisAsyncTask task : analysisAsyncTaskRepository.findByStatusIn(EnumSet.of(TaskStatus.PENDING, TaskStatus.RUNNING))) { + try { + expiredCount += transactionTemplate.execute(status -> sweepTimedOutTask(task.getTaskId())); + } catch (RuntimeException e) { + log.error("Analysis async task sweep failed for taskId={}", task.getTaskId(), e); + } + } + return expiredCount; + } + + private int sweepTimedOutTask(String taskId) { + AnalysisAsyncTask task = analysisAsyncTaskRepository.findById(taskId).orElse(null); + if (task == null || task.getStatus() == TaskStatus.SUCCEEDED || task.getStatus() == TaskStatus.FAILED) { + return 0; + } + + ExpirationDecision expirationDecision = resolveExpiration(task); + if (expirationDecision == null) { + return 0; + } + + releaseCreditIfNeeded(task); + analysisAsyncTaskService.markFailed( + task.getTaskId(), + expirationDecision.failureReason(), + expirationDecision.errorMessage(), + task.getRetryCount() + ); + return 1; + } + + private ExpirationDecision resolveExpiration(AnalysisAsyncTask task) { + LocalDateTime now = LocalDateTime.now(); + if (task.getStatus() == TaskStatus.PENDING && isExpired(task.getSubmittedAt(), now, queueTimeoutMinutes)) { + return new ExpirationDecision( + FailureReason.QUEUE_TIMEOUT, + "자소서 분석 작업이 대기열에서 시간 내 처리되지 않았습니다." + ); + } + + LocalDateTime lastActivityAt = task.getLastAttemptAt() != null ? task.getLastAttemptAt() : task.getStartedAt(); + if (task.getStatus() == TaskStatus.RUNNING && isExpired(lastActivityAt, now, processingTimeoutMinutes)) { + return new ExpirationDecision( + FailureReason.INTERNAL_ERROR, + "자소서 분석 작업이 처리 제한 시간을 초과했습니다." + ); + } + + return null; + } + + private boolean isExpired(LocalDateTime baseTime, LocalDateTime now, long timeoutMinutes) { + if (baseTime == null || timeoutMinutes <= 0) { + return false; + } + return Duration.between(baseTime, now).toMinutes() >= timeoutMinutes; + } + + private void releaseCreditIfNeeded(AnalysisAsyncTask task) { + if (task.getCreditStatus() != CreditStatus.RESERVED || task.getCreditReferenceId() == null) { + return; + } + + User user = userService.getUser(task.getUserId()); + analysisService.refundAnalysisCredit(user, task.getCreditReferenceId()); + analysisAsyncTaskService.markCreditReleased(task.getTaskId()); + } + + private record ExpirationDecision(FailureReason failureReason, String errorMessage) { + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java index 816cf5d..c9970d1 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java @@ -12,6 +12,8 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; import java.time.Instant; import java.util.EnumSet; @@ -19,9 +21,11 @@ @Service @RequiredArgsConstructor +// 분석 비동기 task 엔티티의 생성, 상태 전이, 조회를 전담하는 서비스다. public class AnalysisAsyncTaskService { private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; + private final AnalysisAsyncSseService analysisAsyncSseService; @Value("${app.worker.analysis.max-retry-count:3}") private int maxRetryCount; @@ -49,22 +53,30 @@ public Optional findActiveTask(Long userId, Long mockApplyId) @Transactional public void markRunning(String taskId, String workerId, int retryCount, Instant submittedAt) { - getTask(taskId).markRunning(workerId, retryCount, submittedAt); + AnalysisAsyncTask task = getTask(taskId); + task.markRunning(workerId, retryCount, submittedAt); + publishAfterCommit(toStatusResponse(task)); } @Transactional public void markSuccess(String taskId) { - getTask(taskId).markSuccess(); + AnalysisAsyncTask task = getTask(taskId); + task.markSuccess(); + publishAfterCommit(toStatusResponse(task)); } @Transactional public void markRetryScheduled(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { - getTask(taskId).markRetryScheduled(failureReason, errorMessage, retryCount); + AnalysisAsyncTask task = getTask(taskId); + task.markRetryScheduled(failureReason, errorMessage, retryCount); + publishAfterCommit(toStatusResponse(task)); } @Transactional public void markFailed(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { - getTask(taskId).markFailed(failureReason, errorMessage, retryCount); + AnalysisAsyncTask task = getTask(taskId); + task.markFailed(failureReason, errorMessage, retryCount); + publishAfterCommit(toStatusResponse(task)); } @Transactional @@ -135,4 +147,17 @@ private AnalysisAsyncStatusResponse toStatusResponse(AnalysisAsyncTask task) { .result(null) .build(); } + + private void publishAfterCommit(AnalysisAsyncStatusResponse statusResponse) { + if (!TransactionSynchronizationManager.isSynchronizationActive()) { + analysisAsyncSseService.publish(statusResponse); + return; + } + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void afterCommit() { + analysisAsyncSseService.publish(statusResponse); + } + }); + } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisExecutionPayload.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisExecutionPayload.java index 5461bf8..078c996 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisExecutionPayload.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisExecutionPayload.java @@ -5,6 +5,7 @@ import java.util.List; +// 실제 분석 실행에 필요한 공고, 문항, 답변 데이터를 묶어 전달하는 payload다. public record AnalysisExecutionPayload( Long userId, Long mockApplyId, diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisImprovementRules.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisImprovementRules.java index 6f4c012..cee9ba6 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisImprovementRules.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisImprovementRules.java @@ -2,6 +2,7 @@ import java.util.List; +// 개선 문구를 후처리할 때 사용하는 내부 규칙 모음이다. final class AnalysisImprovementRules { private static final List BANNED_PHRASES = List.of( "추가하세요", diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisQueueProperties.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisQueueProperties.java index 6c155eb..f3462f8 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisQueueProperties.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisQueueProperties.java @@ -4,6 +4,7 @@ import org.springframework.stereotype.Component; @Component +// 분석 워커 큐 관련 설정값을 주입받아 묶어두는 설정 객체다. public record AnalysisQueueProperties( @Value("${app.worker.analysis.routing-key:analysis.execute}") String routingKey ) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java index 0385b48..01b7650 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java @@ -47,6 +47,7 @@ @Slf4j @RequiredArgsConstructor @Transactional(readOnly = true) +// 자소서 분석의 핵심 비즈니스 로직과 결과 저장을 담당하는 메인 서비스다. public class AnalysisService { private static final int MIN_SCORE = 0; private static final int MAX_SCORE = 100; @@ -73,16 +74,15 @@ public class AnalysisService { public AnalysisResponse analyze(User user, Long mockApplyId) { validateAnalysisRequest(user, mockApplyId); String referenceId = "mockApplyId=" + mockApplyId; - reserveAnalysisCredit(user, referenceId); + deductAnalysisCredit(user, referenceId); try { AnalysisExecutionPayload payload = prepareAnalysisExecution(user, mockApplyId); AnalysisLlmResponse llmResponse = executeAnalysis(payload); AnalysisResponse response = finalizeAnalysis(user, mockApplyId, payload, llmResponse); - confirmAnalysisCredit(user, referenceId); return response; } catch (RuntimeException e) { - releaseAnalysisCredit(user, referenceId); + refundAnalysisCredit(user, referenceId); throw e; } } @@ -103,20 +103,13 @@ public void validateAnalysisRequest(User user, Long mockApplyId) { } @Transactional - public void reserveAnalysisCredit(User user, String referenceId) { - creditService.use(user, 1, "자소서 분석 크레딧 예약", referenceId); + public void deductAnalysisCredit(User user, String referenceId) { + creditService.use(user, 1, "자소서 분석 크레딧 차감", referenceId); } @Transactional - public void confirmAnalysisCredit(User user, String referenceId) { - if (user == null || referenceId == null) { - return; - } - } - - @Transactional - public void releaseAnalysisCredit(User user, String referenceId) { - creditService.refund(user, 1, "자소서 분석 크레딧 예약 해제", referenceId); + public void refundAnalysisCredit(User user, String referenceId) { + creditService.refund(user, 1, "자소서 분석 크레딧 환불", referenceId); } @Transactional(readOnly = true) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisTaskMessagePublisher.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisTaskMessagePublisher.java index 6a8e781..2104ad2 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisTaskMessagePublisher.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisTaskMessagePublisher.java @@ -9,6 +9,7 @@ @Service @RequiredArgsConstructor +// 분석 작업 메시지를 공통 worker exchange로 발행하는 publisher다. public class AnalysisTaskMessagePublisher { private final RabbitPublishSupport rabbitPublishSupport; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java index 1d28b78..d2a2121 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java @@ -23,6 +23,7 @@ @Service @RequiredArgsConstructor +// 외부 분석 워커와 내부 분석 도메인 상태를 연결해 주는 브리지 서비스다. public class AnalysisWorkerBridgeService { private final AnalysisAsyncTaskService analysisAsyncTaskService; @@ -156,7 +157,7 @@ private void reserveCreditIfNeeded(AnalysisAsyncTask task) { User user = userService.getUser(task.getUserId()); String creditReferenceId = "analysisTaskId=" + task.getTaskId(); - analysisService.reserveAnalysisCredit(user, creditReferenceId); + analysisService.deductAnalysisCredit(user, creditReferenceId); analysisAsyncTaskService.markCreditReserved(task.getTaskId(), creditReferenceId); } @@ -164,8 +165,6 @@ private void confirmCreditIfNeeded(AnalysisAsyncTask task, User user) { if (task.getCreditStatus() != CreditStatus.RESERVED || task.getCreditReferenceId() == null) { return; } - - analysisService.confirmAnalysisCredit(user, task.getCreditReferenceId()); analysisAsyncTaskService.markCreditConfirmed(task.getTaskId()); } @@ -174,7 +173,7 @@ private void releaseCreditIfNeeded(AnalysisAsyncTask task) { return; } User user = userService.getUser(task.getUserId()); - analysisService.releaseAnalysisCredit(user, task.getCreditReferenceId()); + analysisService.refundAnalysisCredit(user, task.getCreditReferenceId()); analysisAsyncTaskService.markCreditReleased(task.getTaskId()); } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/QuestionService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/QuestionService.java index c355f83..f72e28f 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/QuestionService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/QuestionService.java @@ -34,6 +34,7 @@ @Service @RequiredArgsConstructor @Transactional(readOnly = true) +// 분석 전 단계에서 문항 선택, 추가, 답변 저장을 담당하는 서비스다. public class QuestionService { private static final int MIN_SELECTION_COUNT = 1; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java index 3e3fedb..103a98e 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java @@ -7,6 +7,7 @@ import com.jobdri.jobdri_api.domain.jobposting.dto.response.JobPostingExtractResponse; import com.jobdri.jobdri_api.domain.jobposting.service.JobPostingAiService; import com.jobdri.jobdri_api.domain.jobposting.service.JobPostingAsyncFacadeService; +import com.jobdri.jobdri_api.domain.jobposting.service.JobPostingAsyncSseService; import com.jobdri.jobdri_api.domain.user.service.UserService; import com.jobdri.jobdri_api.global.apiPayload.ApiResponse; import com.jobdri.jobdri_api.global.security.UserDetailsImpl; @@ -20,6 +21,7 @@ import lombok.RequiredArgsConstructor; import org.springframework.http.MediaType; import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; @@ -35,6 +37,7 @@ public class JobPostingAiController { private final JobPostingAiService jobPostingAiService; private final JobPostingAsyncFacadeService jobPostingAsyncFacadeService; + private final JobPostingAsyncSseService jobPostingAsyncSseService; private final UserService userService; @Operation( @@ -108,6 +111,22 @@ public ApiResponse getIngestJobPostingAsyncStatus ); } + @Operation( + summary = "채용 공고 비동기 작업 상태 SSE 구독", + description = "taskId로 비동기 작업 상태를 SSE 스트림으로 구독합니다." + ) + @GetMapping(value = "/ingest/async/{taskId}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter streamIngestJobPostingAsyncStatus( + @AuthenticationPrincipal UserDetailsImpl userDetails, + @PathVariable String taskId + ) { + var user = validateAuthenticatedUser(userDetails); + return jobPostingAsyncSseService.subscribe( + taskId, + () -> jobPostingAsyncFacadeService.getTask(user, taskId) + ); + } + private com.jobdri.jobdri_api.domain.user.entity.User validateAuthenticatedUser(UserDetailsImpl userDetails) { return userService.validateUser(userDetails == null ? null : userDetails.getUser()); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingWorkerInternalController.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingWorkerInternalController.java index a5cfa8a..1d22206 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingWorkerInternalController.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingWorkerInternalController.java @@ -55,7 +55,14 @@ public ApiResponse markRetry( @Valid @RequestBody JobPostingWorkerRetryRequest request ) { internalApiKeyValidator.validate(internalApiKey); - jobPostingWorkerBridgeService.markRetry(taskId, request.failureReason(), request.errorMessage(), request.retryCount()); + jobPostingWorkerBridgeService.markRetry( + taskId, + request.failureReason(), + request.errorMessage(), + request.retryCount(), + request.workerId(), + request.queueLatencyMillis() + ); return ApiResponse.onSuccess("채용 공고 worker 작업 재시도 상태를 반영했습니다."); } @@ -79,7 +86,14 @@ public ApiResponse failTask( @Valid @RequestBody JobPostingWorkerFailureRequest request ) { internalApiKeyValidator.validate(internalApiKey); - jobPostingWorkerBridgeService.failTask(taskId, request.failureReason(), request.errorMessage(), request.retryCount()); + jobPostingWorkerBridgeService.failTask( + taskId, + request.failureReason(), + request.errorMessage(), + request.retryCount(), + request.workerId(), + request.queueLatencyMillis() + ); return ApiResponse.onSuccess("채용 공고 worker 작업 실패 상태를 반영했습니다."); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java index 3305fef..3dc9a3c 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java @@ -4,10 +4,13 @@ import jakarta.validation.constraints.Min; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Size; public record JobPostingWorkerFailureRequest( @NotNull FailureReason failureReason, @NotBlank String errorMessage, - @Min(0) int retryCount + @Min(0) int retryCount, + @NotBlank @Size(max = 100) String workerId, + Long queueLatencyMillis ) { } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java index 345b49f..99e3ec1 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java @@ -4,10 +4,13 @@ import jakarta.validation.constraints.Min; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Size; public record JobPostingWorkerRetryRequest( @NotNull FailureReason failureReason, @NotBlank String errorMessage, - @Min(0) int retryCount + @Min(0) int retryCount, + @NotBlank @Size(max = 100) String workerId, + Long queueLatencyMillis ) { } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java index a5d811e..c2fc1c0 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java @@ -108,6 +108,13 @@ public void markSuccess(String resultPayload) { } public void markRetryScheduled(FailureReason failureReason, String errorMessage, int retryCount) { + if (isTerminal()) { + return; + } + if (retryCount >= maxRetryCount) { + markFailed(failureReason, errorMessage, retryCount); + return; + } this.status = TaskStatus.PENDING; this.message = "채용 공고 비동기 재시도를 대기 중입니다."; this.failureReason = failureReason; @@ -117,6 +124,9 @@ public void markRetryScheduled(FailureReason failureReason, String errorMessage, } public void markFailed(FailureReason failureReason, String errorMessage, int retryCount) { + if (isTerminal()) { + return; + } this.status = TaskStatus.FAILED; this.message = "채용 공고 비동기 처리에 실패했습니다."; this.failureReason = failureReason; @@ -125,6 +135,19 @@ public void markFailed(FailureReason failureReason, String errorMessage, int ret this.completedAt = LocalDateTime.now(); } + public void updateWorkerMetadata(String workerId, Long queueLatencyMillis) { + if (workerId != null && !workerId.isBlank()) { + this.workerId = workerId; + } + if (queueLatencyMillis != null) { + this.queueLatencyMillis = Math.max(0L, queueLatencyMillis); + } + } + + private boolean isTerminal() { + return status == TaskStatus.SUCCEEDED || status == TaskStatus.FAILED; + } + public enum FailureReason { RATE_LIMIT, QUEUE_TIMEOUT, diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/repository/JobPostingAsyncTaskRepository.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/repository/JobPostingAsyncTaskRepository.java index 4841c0c..5deefb5 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/repository/JobPostingAsyncTaskRepository.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/repository/JobPostingAsyncTaskRepository.java @@ -1,10 +1,15 @@ package com.jobdri.jobdri_api.domain.jobposting.repository; import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask; +import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask.TaskStatus; import org.springframework.data.jpa.repository.JpaRepository; +import java.util.Collection; +import java.util.List; import java.util.Optional; public interface JobPostingAsyncTaskRepository extends JpaRepository { Optional findByTaskIdAndUserId(String taskId, Long userId); + + List findByStatusIn(Collection statuses); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java new file mode 100644 index 0000000..9a1aebc --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java @@ -0,0 +1,46 @@ +package com.jobdri.jobdri_api.domain.jobposting.service; + +import com.jobdri.jobdri_api.domain.jobposting.dto.response.JobPostingAsyncStatusResponse; +import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask.TaskStatus; +import com.jobdri.jobdri_api.global.sse.SseSubscriptionRegistry; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.function.Supplier; + +@Service +@RequiredArgsConstructor +public class JobPostingAsyncSseService { + + private static final String EVENT_NAME = "job-posting-status"; + + private final SseSubscriptionRegistry sseSubscriptionRegistry; + + public SseEmitter subscribe(String taskId, Supplier initialStatusSupplier) { + return sseSubscriptionRegistry.subscribe( + channelKey(taskId), + EVENT_NAME, + initialStatusSupplier, + this::isTerminal + ); + } + + public void publish(JobPostingAsyncStatusResponse statusResponse) { + sseSubscriptionRegistry.publish( + channelKey(statusResponse.getTaskId()), + EVENT_NAME, + statusResponse, + isTerminal(statusResponse) + ); + } + + private String channelKey(String taskId) { + return "job-posting:" + taskId; + } + + private boolean isTerminal(JobPostingAsyncStatusResponse statusResponse) { + TaskStatus status = TaskStatus.valueOf(statusResponse.getStatus()); + return status == TaskStatus.SUCCEEDED || status == TaskStatus.FAILED; + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java index 4a4fd79..266b8fc 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java @@ -16,10 +16,13 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; import java.time.Duration; import java.time.Instant; import java.time.LocalDateTime; +import java.util.EnumSet; @Service @RequiredArgsConstructor @@ -27,6 +30,7 @@ public class JobPostingAsyncTaskService { private final JobPostingAsyncTaskRepository jobPostingAsyncTaskRepository; private final ObjectMapper objectMapper; + private final JobPostingAsyncSseService jobPostingAsyncSseService; @Value("${app.worker.job-posting.max-retry-count:3}") private int maxRetryCount; @@ -54,6 +58,7 @@ public void markRunning(String taskId, String workerId, int retryCount, Instant return; } task.markRunning(workerId, retryCount, submittedAt); + publishAfterCommit(toStatusResponse(task)); } @Transactional @@ -69,6 +74,7 @@ public JobPostingIngestResponse markSuccess(String taskId, JobPostingIngestRespo ); } task.markSuccess(serializeResult(result)); + publishAfterCommit(toStatusResponse(task)); return result; } @@ -79,6 +85,7 @@ public void markRetryScheduled(String taskId, FailureReason failureReason, Strin return; } task.markRetryScheduled(failureReason, errorMessage, retryCount); + publishAfterCommit(toStatusResponse(task)); } @Transactional @@ -88,6 +95,12 @@ public void markFailed(String taskId, FailureReason failureReason, String errorM return; } task.markFailed(failureReason, errorMessage, retryCount); + publishAfterCommit(toStatusResponse(task)); + } + + @Transactional + public void updateWorkerMetadata(String taskId, String workerId, Long queueLatencyMillis) { + getTaskState(taskId).updateWorkerMetadata(workerId, queueLatencyMillis); } @Transactional @@ -112,6 +125,20 @@ public JobPostingAsyncStatusResponse getTask(User user, String taskId) { return toStatusResponse(taskState); } + @Transactional + public int sweepTimedOutTasks() { + int expiredCount = 0; + for (JobPostingAsyncTask task : jobPostingAsyncTaskRepository.findByStatusIn(EnumSet.of(TaskStatus.PENDING, TaskStatus.RUNNING))) { + TaskStatus beforeStatus = task.getStatus(); + expireTimedOutTaskIfNeeded(task); + if (beforeStatus != task.getStatus() && task.getStatus() == TaskStatus.FAILED) { + publishAfterCommit(toStatusResponse(task)); + expiredCount++; + } + } + return expiredCount; + } + private JobPostingAsyncStatusResponse toStatusResponse(JobPostingAsyncTask taskState) { return JobPostingAsyncStatusResponse.builder() .taskId(taskState.getTaskId()) @@ -202,4 +229,17 @@ private JobPostingIngestResponse deserializeResult(String resultPayload) { ); } } + + private void publishAfterCommit(JobPostingAsyncStatusResponse statusResponse) { + if (!TransactionSynchronizationManager.isSynchronizationActive()) { + jobPostingAsyncSseService.publish(statusResponse); + return; + } + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void afterCommit() { + jobPostingAsyncSseService.publish(statusResponse); + } + }); + } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java index 768bf97..72497ed 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java @@ -39,11 +39,37 @@ public JobPostingIngestResponse completeTask(String taskId, JobPostingIngestResp return jobPostingAsyncTaskService.markSuccess(taskId, result); } - public void markRetry(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { + @Transactional + public void markRetry( + String taskId, + FailureReason failureReason, + String errorMessage, + int retryCount, + String workerId, + Long queueLatencyMillis + ) { + JobPostingAsyncTask task = getTask(taskId); + if (task.getStatus() == TaskStatus.SUCCEEDED || task.getStatus() == TaskStatus.FAILED) { + return; + } + jobPostingAsyncTaskService.updateWorkerMetadata(taskId, workerId, queueLatencyMillis); jobPostingAsyncTaskService.markRetryScheduled(taskId, failureReason, errorMessage, retryCount); } - public void failTask(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { + @Transactional + public void failTask( + String taskId, + FailureReason failureReason, + String errorMessage, + int retryCount, + String workerId, + Long queueLatencyMillis + ) { + JobPostingAsyncTask task = getTask(taskId); + if (task.getStatus() == TaskStatus.SUCCEEDED || task.getStatus() == TaskStatus.FAILED) { + return; + } + jobPostingAsyncTaskService.updateWorkerMetadata(taskId, workerId, queueLatencyMillis); jobPostingAsyncTaskService.markFailed(taskId, failureReason, errorMessage, retryCount); } @@ -104,6 +130,14 @@ public JobPostingIngestResponse finalizeAndComplete( return jobPostingAsyncTaskService.markSuccess(taskId, result); } + private JobPostingAsyncTask getTask(String taskId) { + return jobPostingAsyncTaskRepository.findById(taskId) + .orElseThrow(() -> new GeneralException( + GeneralErrorCode.JOB_POSTING_ASYNC_TASK_NOT_FOUND, + "해당 비동기 작업을 찾을 수 없습니다. taskId=" + taskId + )); + } + private String fallbackCompanyName(String companyName) { if (companyName == null || companyName.isBlank()) { return "미분류 회사"; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/Payment.java b/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/Payment.java index 62b99f7..de03f3b 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/Payment.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/Payment.java @@ -81,9 +81,8 @@ public void fail() { this.status = PaymentStatus.FAILED; } - public void resetToPending() { - this.paymentKey = null; - this.status = PaymentStatus.PENDING; + public void markUnknown() { + this.status = PaymentStatus.UNKNOWN; } public boolean belongsTo(Long userId) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/PaymentStatus.java b/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/PaymentStatus.java index c274f83..03fbb0e 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/PaymentStatus.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/PaymentStatus.java @@ -6,6 +6,7 @@ public enum PaymentStatus { PENDING, PROCESSING, + UNKNOWN, FAILED, COMPLETED } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentService.java b/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentService.java index 47de7c1..96662a4 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentService.java @@ -83,7 +83,7 @@ public PaymentConfirmResponse confirm(User user, PaymentConfirmRequest request) validateTossResponse(request, tossResponse); } catch (GeneralException e) { if (e.getCode() == GeneralErrorCode.EXTERNAL_SERVICE_TIMEOUT) { - paymentTransactionService.rollbackConfirmationToPending( + paymentTransactionService.markConfirmationUnknown( validatedUser.getId(), request.orderId(), request.paymentKey() diff --git a/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java b/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java index 4ed0e4c..64c4edc 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java @@ -32,6 +32,13 @@ public PaymentConfirmationStart startConfirmation(Long userId, PaymentConfirmReq } return new PaymentConfirmationStart(payment, true); } + if (payment.getStatus() == PaymentStatus.UNKNOWN) { + if (!payment.hasPaymentKey(request.paymentKey())) { + throw new GeneralException(GeneralErrorCode.PAYMENT_ALREADY_PROCESSED, "이미 처리된 결제입니다."); + } + payment.markProcessing(request.paymentKey()); + return new PaymentConfirmationStart(payment, false); + } if (payment.getStatus() == PaymentStatus.PROCESSING || payment.getStatus() == PaymentStatus.FAILED) { throw new GeneralException(GeneralErrorCode.PAYMENT_ALREADY_PROCESSED, "이미 처리된 결제입니다."); } @@ -77,13 +84,13 @@ public void failConfirmation(Long userId, String orderId, String paymentKey) { } @Transactional - public void rollbackConfirmationToPending(Long userId, String orderId, String paymentKey) { + public void markConfirmationUnknown(Long userId, String orderId, String paymentKey) { Payment payment = getOwnedPaymentForUpdate(userId, orderId); if (!payment.hasPaymentKey(paymentKey)) { return; } if (payment.getStatus() == PaymentStatus.PROCESSING) { - payment.resetToPending(); + payment.markUnknown(); } } diff --git a/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java b/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java new file mode 100644 index 0000000..40cc55d --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java @@ -0,0 +1,52 @@ +package com.jobdri.jobdri_api.global.scheduling; + +import com.jobdri.jobdri_api.domain.analysis.service.AnalysisAsyncSweepService; +import com.jobdri.jobdri_api.domain.jobposting.service.JobPostingAsyncTaskService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +@Slf4j +@Component +@RequiredArgsConstructor +public class AsyncTaskSweepScheduler { + + private final JobPostingAsyncTaskService jobPostingAsyncTaskService; + private final AnalysisAsyncSweepService analysisAsyncSweepService; + + @Scheduled( + fixedDelayString = "${app.async-task-sweep.fixed-delay-millis:60000}", + initialDelayString = "${app.async-task-sweep.initial-delay-millis:30000}" + ) + public void sweepTimedOutTasks() { + int expiredJobPostingTasks = sweepJobPostingTasks(); + int expiredAnalysisTasks = sweepAnalysisTasks(); + + if (expiredJobPostingTasks > 0 || expiredAnalysisTasks > 0) { + log.info( + "Async task sweep expired tasks. jobPostingExpired={} analysisExpired={}", + expiredJobPostingTasks, + expiredAnalysisTasks + ); + } + } + + private int sweepJobPostingTasks() { + try { + return jobPostingAsyncTaskService.sweepTimedOutTasks(); + } catch (RuntimeException e) { + log.error("Job posting async task sweep failed.", e); + return 0; + } + } + + private int sweepAnalysisTasks() { + try { + return analysisAsyncSweepService.sweepTimedOutTasks(); + } catch (RuntimeException e) { + log.error("Analysis async task sweep failed.", e); + return 0; + } + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java new file mode 100644 index 0000000..3f5f6dc --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java @@ -0,0 +1,171 @@ +package com.jobdri.jobdri_api.global.sse; + +import jakarta.annotation.PreDestroy; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; +import java.util.function.Supplier; + +@Component +public class SseSubscriptionRegistry { + + private final Map> emitters = new ConcurrentHashMap<>(); + private final Map channelMonitors = new ConcurrentHashMap<>(); + private final long timeoutMillis; + private final ScheduledExecutorService heartbeatExecutor; + private final ExecutorService heartbeatDispatchExecutor; + + public SseSubscriptionRegistry( + @Value("${app.sse.timeout-millis:1800000}") long timeoutMillis, + @Value("${app.sse.heartbeat-interval-millis:15000}") long heartbeatIntervalMillis, + @Value("${app.sse.heartbeat-dispatch-thread-count:4}") int heartbeatDispatchThreadCount + ) { + this.timeoutMillis = timeoutMillis; + this.heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "sse-heartbeat"); + thread.setDaemon(true); + return thread; + }); + this.heartbeatDispatchExecutor = Executors.newFixedThreadPool( + Math.max(1, heartbeatDispatchThreadCount), + runnable -> { + Thread thread = new Thread(runnable, "sse-heartbeat-dispatch"); + thread.setDaemon(true); + return thread; + } + ); + if (heartbeatIntervalMillis > 0) { + this.heartbeatExecutor.scheduleAtFixedRate( + this::sendHeartbeatSafely, + heartbeatIntervalMillis, + heartbeatIntervalMillis, + TimeUnit.MILLISECONDS + ); + } + } + + public SseEmitter subscribe(String channelKey, String eventName, Object initialPayload, boolean completeAfterInitial) { + return subscribe(channelKey, eventName, () -> initialPayload, ignored -> completeAfterInitial); + } + + public SseEmitter subscribe( + String channelKey, + String eventName, + Supplier initialPayloadSupplier, + Predicate completeAfterInitialPredicate + ) { + SseEmitter emitter = new SseEmitter(timeoutMillis); + synchronized (channelMonitor(channelKey)) { + emitters.computeIfAbsent(channelKey, ignored -> new CopyOnWriteArrayList<>()).add(emitter); + + emitter.onCompletion(() -> remove(channelKey, emitter)); + emitter.onTimeout(() -> { + emitter.complete(); + remove(channelKey, emitter); + }); + emitter.onError(ignored -> remove(channelKey, emitter)); + + T initialPayload = initialPayloadSupplier.get(); + boolean completeAfterInitial = completeAfterInitialPredicate.test(initialPayload); + + try { + emitter.send(SseEmitter.event().name("connected").data("connected")); + emitter.send(SseEmitter.event().name(eventName).data(initialPayload)); + if (completeAfterInitial) { + emitter.complete(); + remove(channelKey, emitter); + } + } catch (Exception e) { + emitter.completeWithError(e); + remove(channelKey, emitter); + } + } + + return emitter; + } + + public void publish(String channelKey, String eventName, Object payload, boolean completeAfterPublish) { + synchronized (channelMonitor(channelKey)) { + List subscribers = emitters.get(channelKey); + if (subscribers == null || subscribers.isEmpty()) { + return; + } + + for (SseEmitter emitter : subscribers) { + try { + emitter.send(SseEmitter.event().name(eventName).data(payload)); + if (completeAfterPublish) { + emitter.complete(); + remove(channelKey, emitter); + } + } catch (Exception e) { + emitter.completeWithError(e); + remove(channelKey, emitter); + } + } + } + } + + @PreDestroy + void shutdown() { + heartbeatExecutor.shutdownNow(); + heartbeatDispatchExecutor.shutdownNow(); + } + + private void sendHeartbeatSafely() { + try { + sendHeartbeat(); + } catch (Exception ignored) { + // Heartbeat failures should not affect the rest of the application. + } + } + + private void sendHeartbeat() { + for (Map.Entry> entry : emitters.entrySet()) { + String channelKey = entry.getKey(); + for (SseEmitter emitter : entry.getValue()) { + heartbeatDispatchExecutor.execute(() -> sendHeartbeat(channelKey, emitter)); + } + } + } + + private void sendHeartbeat(String channelKey, SseEmitter emitter) { + synchronized (channelMonitor(channelKey)) { + if (!emitters.getOrDefault(channelKey, new CopyOnWriteArrayList<>()).contains(emitter)) { + return; + } + try { + emitter.send(SseEmitter.event().name("heartbeat").data("ping")); + } catch (Exception e) { + emitter.completeWithError(e); + remove(channelKey, emitter); + } + } + } + + private void remove(String channelKey, SseEmitter emitter) { + List subscribers = emitters.get(channelKey); + if (subscribers == null) { + return; + } + subscribers.remove(emitter); + if (subscribers.isEmpty()) { + emitters.remove(channelKey); + channelMonitors.remove(channelKey); + } + } + + private Object channelMonitor(String channelKey) { + return channelMonitors.computeIfAbsent(channelKey, ignored -> new Object()); + } +} diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeServiceTest.java index 443b93d..a7c69be 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeServiceTest.java @@ -60,7 +60,7 @@ void submitReturnsExistingTaskWhenCreateConflicts() { assertThat(response.taskId()).isEqualTo(existingTask.getTaskId()); assertThat(response.status()).isEqualTo("PENDING"); - verify(analysisService, never()).reserveAnalysisCredit(eq(user), anyString()); + verify(analysisService, never()).deductAnalysisCredit(eq(user), anyString()); verify(analysisAsyncProcessor, never()).process(eq(existingTask.getTaskId()), eq(1L), eq(10L), eq(3)); } @@ -98,7 +98,7 @@ void submitDoesNotReserveCreditBeforeWorkerStarts() { assertThat(response.taskId()).isEqualTo(createdTask.getTaskId()); assertThat(response.status()).isEqualTo("PENDING"); verify(analysisAsyncProcessor, times(1)).process(createdTask.getTaskId(), 1L, 10L, 3); - verify(analysisService, never()).reserveAnalysisCredit(eq(user), anyString()); + verify(analysisService, never()).deductAnalysisCredit(eq(user), anyString()); } @Test diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeServiceTest.java index 84b6857..1656527 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeServiceTest.java @@ -79,7 +79,7 @@ void getContextReservesCreditBeforePreparingExecution() { analysisWorkerBridgeService.getContext(task.getTaskId(), 1L, 10L); - verify(analysisService).reserveAnalysisCredit(user, "analysisTaskId=" + task.getTaskId()); + verify(analysisService).deductAnalysisCredit(user, "analysisTaskId=" + task.getTaskId()); verify(analysisAsyncTaskService).markCreditReserved(task.getTaskId(), "analysisTaskId=" + task.getTaskId()); verify(analysisService).prepareAnalysisExecution(user, 10L); } @@ -117,7 +117,7 @@ void getContextDoesNotReserveCreditTwice() { analysisWorkerBridgeService.getContext(task.getTaskId(), 1L, 10L); - verify(analysisService, never()).reserveAnalysisCredit(eq(user), anyString()); + verify(analysisService, never()).deductAnalysisCredit(eq(user), anyString()); verify(analysisAsyncTaskService, never()).markCreditReserved(eq(task.getTaskId()), anyString()); verify(analysisService).prepareAnalysisExecution(user, 10L); } diff --git a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java index 844754d..523ac62 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java @@ -19,6 +19,8 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -28,11 +30,18 @@ class JobPostingAsyncTaskServiceTest { @Mock private JobPostingAsyncTaskRepository jobPostingAsyncTaskRepository; + @Mock + private JobPostingAsyncSseService jobPostingAsyncSseService; + private JobPostingAsyncTaskService jobPostingAsyncTaskService; @BeforeEach void setUp() { - jobPostingAsyncTaskService = new JobPostingAsyncTaskService(jobPostingAsyncTaskRepository, new ObjectMapper()); + jobPostingAsyncTaskService = new JobPostingAsyncTaskService( + jobPostingAsyncTaskRepository, + new ObjectMapper(), + jobPostingAsyncSseService + ); } @Test @@ -108,4 +117,60 @@ void getTaskMarksRunningTimeout() { assertThat(response.getStatus()).isEqualTo("FAILED"); assertThat(response.getFailureReason()).isEqualTo(FailureReason.WORKER_TIMEOUT.name()); } + + @Test + @DisplayName("재시도 횟수가 최대값에 도달하면 FAILED로 전이한다") + void markRetryScheduledFailsWhenMaxRetryReached() { + JobPostingAsyncTask task = JobPostingAsyncTask.pending(7L, 3); + + when(jobPostingAsyncTaskRepository.findById(task.getTaskId())).thenReturn(Optional.of(task)); + + jobPostingAsyncTaskService.markRetryScheduled( + task.getTaskId(), + FailureReason.INTERNAL_ERROR, + "retry exhausted", + 3 + ); + + assertThat(task.getStatus()).isEqualTo(JobPostingAsyncTask.TaskStatus.FAILED); + assertThat(task.getFailureReason()).isEqualTo(FailureReason.INTERNAL_ERROR); + assertThat(task.getRetryCount()).isEqualTo(3); + verify(jobPostingAsyncSseService).publish(any()); + } + + @Test + @DisplayName("worker 메타데이터를 별도로 갱신할 수 있다") + void updateWorkerMetadataUpdatesWorkerFields() { + JobPostingAsyncTask task = JobPostingAsyncTask.pending(7L, 3); + + when(jobPostingAsyncTaskRepository.findById(task.getTaskId())).thenReturn(Optional.of(task)); + + jobPostingAsyncTaskService.updateWorkerMetadata(task.getTaskId(), "worker-2", 1234L); + + assertThat(task.getWorkerId()).isEqualTo("worker-2"); + assertThat(task.getQueueLatencyMillis()).isEqualTo(1234L); + } + + @Test + @DisplayName("이미 종료된 task에 대한 재시도 예약은 무시된다") + void markRetryScheduledDoesNothingWhenTaskIsTerminal() { + JobPostingAsyncTask task = JobPostingAsyncTask.pending(7L, 3); + task.markFailed(FailureReason.INTERNAL_ERROR, "failed", 1); + String originalError = task.getError(); + int originalRetryCount = task.getRetryCount(); + + when(jobPostingAsyncTaskRepository.findById(task.getTaskId())).thenReturn(Optional.of(task)); + + jobPostingAsyncTaskService.markRetryScheduled( + task.getTaskId(), + FailureReason.QUEUE_TIMEOUT, + "should be ignored", + 2 + ); + + assertThat(task.getFailureReason()).isEqualTo(FailureReason.INTERNAL_ERROR); + assertThat(task.getError()).isEqualTo(originalError); + assertThat(task.getRetryCount()).isEqualTo(originalRetryCount); + verify(jobPostingAsyncSseService, never()).publish(any()); + } } diff --git a/src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java index bb0e9fa..21238a8 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java @@ -184,7 +184,7 @@ void confirmMarksPaymentAsFailedWhenTossConfirmFails() { } @Test - @DisplayName("토스 승인 타임아웃이면 결제를 PENDING으로 되돌리고 재시도를 허용한다") + @DisplayName("토스 승인 타임아웃이면 결제를 UNKNOWN으로 남기고 재시도를 허용한다") void confirmAllowsRetryWhenTossConfirmTimesOut() { User user = saveUser("payment-confirm-timeout@example.com"); PaymentPrepareResponse prepared = paymentService.prepare(user, new PaymentPrepareRequest("ONE_TIME")); @@ -208,7 +208,7 @@ void confirmAllowsRetryWhenTossConfirmTimesOut() { .isEqualTo(GeneralErrorCode.EXTERNAL_SERVICE_TIMEOUT); Payment timedOutPayment = paymentRepository.findByOrderId(prepared.orderId()).orElseThrow(); - assertThat(timedOutPayment.getStatus()).isEqualTo(PaymentStatus.PENDING); + assertThat(timedOutPayment.getStatus()).isEqualTo(PaymentStatus.UNKNOWN); PaymentConfirmResponse retryResponse = paymentService.confirm(user, request);