[Refactor] LLM 호출 동시성 및 중복 작업 제어 개선 (#113)

ops/db/migrations/20260704_analysis_async_tasks.sql

@@ -2,6 +2,8 @@ CREATE TABLE IF NOT EXISTS analysis_async_tasks (
     task_id VARCHAR(36) PRIMARY KEY,
     user_id BIGINT NOT NULL,
     mock_apply_id BIGINT NOT NULL,
+    credit_reference_id VARCHAR(100),
+    credit_status VARCHAR(20) NOT NULL DEFAULT 'NONE',
     status VARCHAR(20) NOT NULL,
     message VARCHAR(255) NOT NULL,
     error VARCHAR(2000),

ops/db/migrations/20260704_analysis_async_tasks_active_unique.sql

@@ -0,0 +1,3 @@
+CREATE UNIQUE INDEX IF NOT EXISTS uk_analysis_async_tasks_active_user_mock_apply
+    ON analysis_async_tasks (user_id, mock_apply_id)
+    WHERE status IN ('PENDING', 'RUNNING');

ops/db/migrations/20260704_analysis_async_tasks_credit_status.sql

@@ -0,0 +1,5 @@
+ALTER TABLE analysis_async_tasks
+    ADD COLUMN IF NOT EXISTS credit_reference_id VARCHAR(100);
+
+ALTER TABLE analysis_async_tasks
+    ADD COLUMN IF NOT EXISTS credit_status VARCHAR(20) NOT NULL DEFAULT 'NONE';

src/main/java/com/jobdri/jobdri_api/domain/analysis/entity/AnalysisAsyncTask.java

@@ -30,6 +30,13 @@ public class AnalysisAsyncTask extends CreatedAtEntity {
     @Column(name = "mock_apply_id", nullable = false)
     private Long mockApplyId;
 
+    @Column(name = "credit_reference_id", length = 100)
+    private String creditReferenceId;
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "credit_status", nullable = false, length = 20)
+    private CreditStatus creditStatus;
+
     @Enumerated(EnumType.STRING)
     @Column(nullable = false, length = 20)
     private TaskStatus status;
@@ -51,11 +58,25 @@ public static AnalysisAsyncTask pending(Long userId, Long mockApplyId) {
         task.taskId = UUID.randomUUID().toString();
         task.userId = userId;
         task.mockApplyId = mockApplyId;
+        task.creditStatus = CreditStatus.NONE;
         task.status = TaskStatus.PENDING;
         task.message = "자소서 분석 비동기 작업이 접수되었습니다.";
         return task;
     }
 
+    public void markCreditReserved(String creditReferenceId) {
+        this.creditReferenceId = creditReferenceId;
+        this.creditStatus = CreditStatus.RESERVED;
+    }
+
+    public void markCreditConfirmed() {
+        this.creditStatus = CreditStatus.CONFIRMED;
+    }
+
+    public void markCreditReleased() {
+        this.creditStatus = CreditStatus.RELEASED;
+    }
+
     public void markRunning() {
         this.status = TaskStatus.RUNNING;
         this.message = "자소서 분석을 진행 중입니다.";
@@ -82,4 +103,11 @@ public enum TaskStatus {
         SUCCEEDED,
         FAILED
     }
+
+    public enum CreditStatus {
+        NONE,
+        RESERVED,
+        CONFIRMED,
+        RELEASED
+    }
 }

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

@@ -7,6 +7,7 @@
 import com.jobdri.jobdri_api.domain.corpus.service.CorpusRetrievalService.RetrievedJobPostingReference;
 import com.jobdri.jobdri_api.domain.corpus.service.CorpusRetrievalService.RetrievedQuestionReference;
 import com.jobdri.jobdri_api.domain.jobposting.entity.JobPosting;
+import com.jobdri.jobdri_api.global.config.LlmConcurrencyLimiter;
 import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode;
 import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException;
 import com.openai.client.OpenAIClient;
@@ -29,10 +30,15 @@ public class AnalysisAiClient {
 
     private final OpenAIClient openAIClient;
     private final CorpusRetrievalService corpusRetrievalService;
+    private final LlmConcurrencyLimiter llmConcurrencyLimiter;
 
     @Value("${openai.model.cover-letter-analysis:gpt-4o-mini}")
     private String analysisModel;
 
+    public AnalysisLlmResponse analyze(AnalysisExecutionPayload payload) {
+        return analyze(payload.jobPosting(), payload.answeredQuestions());
+    }
+
     public AnalysisLlmResponse analyze(JobPosting jobPosting, List<Question> questions) {
         RetrievalContext referenceContext = emptyContext();
         try {
@@ -49,7 +55,10 @@ public AnalysisLlmResponse analyze(JobPosting jobPosting, List<Question> questio
                 .build();
 
         try {
-            StructuredResponse<AnalysisLlmResponse> response = openAIClient.responses().create(params);
+            StructuredResponse<AnalysisLlmResponse> response = llmConcurrencyLimiter.execute(
+                    "cover-letter-analysis",
+                    () -> openAIClient.responses().create(params)
+            );
             return extractStructuredContent(response);
         } catch (GeneralException e) {
             throw e;

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

@@ -2,16 +2,21 @@
 
 import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncStatusResponse;
 import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncSubmitResponse;
+import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask;
 import com.jobdri.jobdri_api.domain.user.entity.User;
 import com.jobdri.jobdri_api.domain.user.service.UserService;
 import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode;
 import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException;
 import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.stereotype.Service;
 
+import java.util.Locale;
+
 @Service
 @RequiredArgsConstructor
 public class AnalysisAsyncFacadeService {
+    private static final String ACTIVE_TASK_UNIQUE_CONSTRAINT = "uk_analysis_async_tasks_active_user_mock_apply";
 
     private final AnalysisAsyncTaskService analysisAsyncTaskService;
     private final AnalysisAsyncProcessor analysisAsyncProcessor;
@@ -23,11 +28,7 @@ public AnalysisAsyncSubmitResponse submit(User user, Long mockApplyId) {
         analysisService.validateAnalysisRequest(validatedUser, mockApplyId);
 
         return analysisAsyncTaskService.findActiveTask(validatedUser.getId(), mockApplyId)
-                .map(existingTask -> new AnalysisAsyncSubmitResponse(
-                        existingTask.getTaskId(),
-                        existingTask.getStatus().name(),
-                        "이미 진행 중인 자소서 분석 작업이 있습니다."
-                ))
+                .map(this::toInProgressResponse)
                 .orElseGet(() -> createAndProcessTask(validatedUser, mockApplyId));
     }
 
@@ -55,11 +56,18 @@ public AnalysisAsyncStatusResponse getTask(User user, Long mockApplyId, String t
     }
 
     private AnalysisAsyncSubmitResponse createAndProcessTask(User user, Long mockApplyId) {
-        String taskId = analysisAsyncTaskService.createPendingTask(user.getId(), mockApplyId);
+        PendingTaskResult pendingTaskResult = createPendingTask(user, mockApplyId);
+        if (!pendingTaskResult.created()) {
+            return toInProgressResponse(pendingTaskResult.task());
+        }
+
+        AnalysisAsyncTask task = pendingTaskResult.task();
+        String taskId = task.getTaskId();
         String creditReferenceId = "analysisTaskId=" + taskId;
 
         try {
-            analysisService.chargeAnalysisCredit(user, creditReferenceId);
+            analysisService.reserveAnalysisCredit(user, creditReferenceId);
+            analysisAsyncTaskService.markCreditReserved(taskId, creditReferenceId);
             analysisAsyncProcessor.process(taskId, user.getId(), mockApplyId, creditReferenceId);
             return new AnalysisAsyncSubmitResponse(
                     taskId,
@@ -71,4 +79,50 @@ private AnalysisAsyncSubmitResponse createAndProcessTask(User user, Long mockApp
             throw e;
         }
     }
+
+    private PendingTaskResult createPendingTask(User user, Long mockApplyId) {
+        try {
+            return new PendingTaskResult(
+                    analysisAsyncTaskService.createPendingTask(user.getId(), mockApplyId),
+                    true
+            );
+        } catch (DataIntegrityViolationException e) {
+            if (!isActiveTaskUniqueConflict(e)) {
+                throw e;
+            }
+            AnalysisAsyncTask existingTask = analysisAsyncTaskService.findActiveTask(user.getId(), mockApplyId)
+                    .orElseThrow(() -> e);
+            return new PendingTaskResult(existingTask, false);
+        }
+    }
+
+    private AnalysisAsyncSubmitResponse toInProgressResponse(AnalysisAsyncTask task) {
+        return new AnalysisAsyncSubmitResponse(
+                task.getTaskId(),
+                task.getStatus().name(),
+                "이미 진행 중인 자소서 분석 작업이 있습니다."
+        );
+    }
+
+    private boolean isActiveTaskUniqueConflict(DataIntegrityViolationException exception) {
+        Throwable cause = exception;
+        while (cause != null) {
+            if (cause instanceof org.hibernate.exception.ConstraintViolationException constraintViolation
+                    && containsConstraintName(constraintViolation.getConstraintName())) {
+                return true;
+            }
+            if (containsConstraintName(cause.getMessage())) {
+                return true;
+            }
+            cause = cause.getCause();
+        }
+        return false;
+    }
+
+    private boolean containsConstraintName(String value) {
+        return value != null && value.toLowerCase(Locale.ROOT).contains(ACTIVE_TASK_UNIQUE_CONSTRAINT);
+    }
+
+    private record PendingTaskResult(AnalysisAsyncTask task, boolean created) {
+    }
 }

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

@@ -22,12 +22,17 @@ public void process(String taskId, Long userId, Long mockApplyId, String creditR
 
         try {
             User user = userService.getUser(userId);
-            analysisService.runAnalysis(user, mockApplyId);
+            AnalysisExecutionPayload payload = analysisService.prepareAnalysisExecution(user, mockApplyId);
+            var llmResponse = analysisService.executeAnalysis(payload);
+            analysisService.finalizeAnalysis(user, mockApplyId, payload, llmResponse);
+            analysisService.confirmAnalysisCredit(user, creditReferenceId);
+            analysisAsyncTaskService.markCreditConfirmed(taskId);
             analysisAsyncTaskService.markSuccess(taskId);
         } catch (Exception e) {
             log.error("자소서 분석 비동기 처리 실패: taskId={}, mockApplyId={}", taskId, mockApplyId, e);
             try {
-                analysisService.refundAnalysisCredit(userService.getUser(userId), creditReferenceId);
+                analysisService.releaseAnalysisCredit(userService.getUser(userId), creditReferenceId);
+                analysisAsyncTaskService.markCreditReleased(taskId);
             } catch (Exception refundException) {
                 log.error("자소서 분석 실패 환불 처리 실패: taskId={}, userId={}", taskId, userId, refundException);
             }

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

@@ -2,6 +2,7 @@
 
 import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncStatusResponse;
 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.TaskStatus;
 import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository;
 import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode;
@@ -20,9 +21,8 @@ public class AnalysisAsyncTaskService {
     private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository;
 
     @Transactional
-    public String createPendingTask(Long userId, Long mockApplyId) {
-        AnalysisAsyncTask task = analysisAsyncTaskRepository.save(AnalysisAsyncTask.pending(userId, mockApplyId));
-        return task.getTaskId();
+    public AnalysisAsyncTask createPendingTask(Long userId, Long mockApplyId) {
+        return analysisAsyncTaskRepository.saveAndFlush(AnalysisAsyncTask.pending(userId, mockApplyId));
     }
 
     @Transactional
@@ -54,6 +54,26 @@ public void markFailed(String taskId, String errorMessage) {
         getTask(taskId).markFailed(errorMessage);
     }
 
+    @Transactional
+    public void markCreditReserved(String taskId, String creditReferenceId) {
+        getTask(taskId).markCreditReserved(creditReferenceId);
+    }
+
+    @Transactional
+    public void markCreditConfirmed(String taskId) {
+        getTask(taskId).markCreditConfirmed();
+    }
+
+    @Transactional
+    public void markCreditReleased(String taskId) {
+        getTask(taskId).markCreditReleased();
+    }
+
+    @Transactional(readOnly = true)
+    public CreditStatus getCreditStatus(String taskId) {
+        return getTask(taskId).getCreditStatus();
+    }
+
     @Transactional(readOnly = true)
     public AnalysisAsyncStatusResponse getTaskStatus(Long userId, String taskId) {
         AnalysisAsyncTask task = analysisAsyncTaskRepository.findByTaskIdAndUserId(taskId, userId)

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

@@ -0,0 +1,15 @@
+package com.jobdri.jobdri_api.domain.analysis.service;
+
+import com.jobdri.jobdri_api.domain.analysis.entity.Question;
+import com.jobdri.jobdri_api.domain.jobposting.entity.JobPosting;
+
+import java.util.List;
+
+public record AnalysisExecutionPayload(
+        Long userId,
+        Long mockApplyId,
+        JobPosting jobPosting,
+        List<Question> questions,
+        List<Question> answeredQuestions
+) {
+}