Feat/#109 polling to sse

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) {

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<AnalysisAsyncStatusResponse> 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<AnalysisResponse> getAnalysis(

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<AnalysisAsyncTask, String> {
@@ -15,4 +16,6 @@ Optional<AnalysisAsyncTask> findFirstByUserIdAndMockApplyIdAndStatusInOrderByCre
             Long mockApplyId,
             Collection<TaskStatus> statuses
     );
+
+    List<AnalysisAsyncTask> findByStatusIn(Collection<TaskStatus> statuses);
 }

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;

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;

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";
 

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;

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<AnalysisAsyncStatusResponse> 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;
+    }
+}

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) {
+    }
+}

src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java

@@ -12,16 +12,20 @@
 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;
 import java.util.Optional;
 
 @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<AnalysisAsyncTask> 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);
+            }
+        });
+    }
 }

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,

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<String> BANNED_PHRASES = List.of(
             "추가하세요",

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
 ) {