[Feat] analysis async를 MQ worker 구조로 전환하고 worker 상태 메타데이터를 확장 (#110)

.env.example

@@ -38,11 +38,17 @@ TOSS_SECRET_KEY=change-me
 TOSS_BASE_URL=https://api.tosspayments.com
 
 JOB_POSTING_CLASSIFICATION_CONFIDENCE_THRESHOLD=0.65
+APP_WORKER_EXCHANGE=jobdri.worker.exchange
 APP_WORKER_INTERNAL_API_KEY=change-me-internal-worker-key
-APP_WORKER_JOB_POSTING_EXCHANGE=jobdri.worker.exchange
+APP_WORKER_JOB_POSTING_EXCHANGE=${APP_WORKER_EXCHANGE}
 APP_WORKER_JOB_POSTING_QUEUE=jobdri.job-posting.ingest
 APP_WORKER_JOB_POSTING_ROUTING_KEY=job-posting.ingest
 APP_WORKER_JOB_POSTING_DLQ=jobdri.job-posting.ingest.dlq
+APP_WORKER_ANALYSIS_EXCHANGE=${APP_WORKER_EXCHANGE}
+APP_WORKER_ANALYSIS_QUEUE=jobdri.analysis.execute
+APP_WORKER_ANALYSIS_ROUTING_KEY=analysis.execute
+APP_WORKER_ANALYSIS_DLQ=jobdri.analysis.execute.dlq
+APP_WORKER_ANALYSIS_MAX_RETRY_COUNT=3
 JOB_POSTING_ASYNC_CORE_POOL_SIZE=2
 JOB_POSTING_ASYNC_MAX_POOL_SIZE=4
 JOB_POSTING_ASYNC_QUEUE_CAPACITY=20

.env.production.example

@@ -42,11 +42,17 @@ TOSS_SECRET_KEY=change-me
 TOSS_BASE_URL=https://api.tosspayments.com
 
 JOB_POSTING_CLASSIFICATION_CONFIDENCE_THRESHOLD=0.65
+APP_WORKER_EXCHANGE=jobdri.worker.exchange
 APP_WORKER_INTERNAL_API_KEY=change-me-internal-worker-key
-APP_WORKER_JOB_POSTING_EXCHANGE=jobdri.worker.exchange
+APP_WORKER_JOB_POSTING_EXCHANGE=${APP_WORKER_EXCHANGE}
 APP_WORKER_JOB_POSTING_QUEUE=jobdri.job-posting.ingest
 APP_WORKER_JOB_POSTING_ROUTING_KEY=job-posting.ingest
 APP_WORKER_JOB_POSTING_DLQ=jobdri.job-posting.ingest.dlq
+APP_WORKER_ANALYSIS_EXCHANGE=${APP_WORKER_EXCHANGE}
+APP_WORKER_ANALYSIS_QUEUE=jobdri.analysis.execute
+APP_WORKER_ANALYSIS_ROUTING_KEY=analysis.execute
+APP_WORKER_ANALYSIS_DLQ=jobdri.analysis.execute.dlq
+APP_WORKER_ANALYSIS_MAX_RETRY_COUNT=3
 JOB_POSTING_ASYNC_CORE_POOL_SIZE=2
 JOB_POSTING_ASYNC_MAX_POOL_SIZE=4
 JOB_POSTING_ASYNC_QUEUE_CAPACITY=20

ops/db/migrations/20260704_analysis_async_tasks_worker_metadata.sql

@@ -0,0 +1,20 @@
+ALTER TABLE analysis_async_tasks
+    ADD COLUMN IF NOT EXISTS failure_reason VARCHAR(40);
+
+ALTER TABLE analysis_async_tasks
+    ADD COLUMN IF NOT EXISTS worker_id VARCHAR(100);
+
+ALTER TABLE analysis_async_tasks
+    ADD COLUMN IF NOT EXISTS retry_count INTEGER NOT NULL DEFAULT 0;
+
+ALTER TABLE analysis_async_tasks
+    ADD COLUMN IF NOT EXISTS max_retry_count INTEGER NOT NULL DEFAULT 3;
+
+ALTER TABLE analysis_async_tasks
+    ADD COLUMN IF NOT EXISTS submitted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP;
+
+ALTER TABLE analysis_async_tasks
+    ADD COLUMN IF NOT EXISTS last_attempt_at TIMESTAMP;
+
+ALTER TABLE analysis_async_tasks
+    ADD COLUMN IF NOT EXISTS queue_latency_millis BIGINT;

ops/db/migrations/20260704_job_posting_async_tasks_user_id.sql

@@ -0,0 +1,5 @@
+ALTER TABLE job_posting_async_tasks
+    ADD COLUMN IF NOT EXISTS user_id BIGINT;
+
+CREATE INDEX IF NOT EXISTS idx_job_posting_async_tasks_user_id_created_at
+    ON job_posting_async_tasks (user_id, created_at DESC);

ops/db/migrations/20260704_job_posting_async_tasks_worker_metadata.sql

@@ -0,0 +1,20 @@
+ALTER TABLE job_posting_async_tasks
+    ADD COLUMN IF NOT EXISTS failure_reason VARCHAR(40);
+
+ALTER TABLE job_posting_async_tasks
+    ADD COLUMN IF NOT EXISTS worker_id VARCHAR(100);
+
+ALTER TABLE job_posting_async_tasks
+    ADD COLUMN IF NOT EXISTS retry_count INTEGER NOT NULL DEFAULT 0;
+
+ALTER TABLE job_posting_async_tasks
+    ADD COLUMN IF NOT EXISTS max_retry_count INTEGER NOT NULL DEFAULT 3;
+
+ALTER TABLE job_posting_async_tasks
+    ADD COLUMN IF NOT EXISTS submitted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP;
+
+ALTER TABLE job_posting_async_tasks
+    ADD COLUMN IF NOT EXISTS last_attempt_at TIMESTAMP;
+
+ALTER TABLE job_posting_async_tasks
+    ADD COLUMN IF NOT EXISTS queue_latency_millis BIGINT;

src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisWorkerInternalController.java

@@ -0,0 +1,119 @@
+package com.jobdri.jobdri_api.domain.analysis.controller;
+
+import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncStatusResponse;
+import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisResponse;
+import com.jobdri.jobdri_api.domain.analysis.dto.worker.AnalysisWorkerCompleteRequest;
+import com.jobdri.jobdri_api.domain.analysis.dto.worker.AnalysisWorkerContextRequest;
+import com.jobdri.jobdri_api.domain.analysis.dto.worker.AnalysisWorkerContextResponse;
+import com.jobdri.jobdri_api.domain.analysis.dto.worker.AnalysisWorkerFailureRequest;
+import com.jobdri.jobdri_api.domain.analysis.dto.worker.AnalysisWorkerRetryRequest;
+import com.jobdri.jobdri_api.domain.analysis.dto.worker.AnalysisWorkerRunningRequest;
+import com.jobdri.jobdri_api.domain.analysis.service.AnalysisAsyncTaskService;
+import com.jobdri.jobdri_api.domain.analysis.service.AnalysisWorkerBridgeService;
+import com.jobdri.jobdri_api.global.apiPayload.ApiResponse;
+import com.jobdri.jobdri_api.global.security.InternalApiKeyValidator;
+import jakarta.validation.Valid;
+import lombok.RequiredArgsConstructor;
+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.RequestBody;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequiredArgsConstructor
+@RequestMapping("/api/internal/worker/analysis")
+public class AnalysisWorkerInternalController {
+
+    private static final String INTERNAL_API_KEY_HEADER = "X-Internal-Api-Key";
+
+    private final InternalApiKeyValidator internalApiKeyValidator;
+    private final AnalysisWorkerBridgeService analysisWorkerBridgeService;
+    private final AnalysisAsyncTaskService analysisAsyncTaskService;
+
+    @PostMapping("/tasks/{taskId}/running")
+    public ApiResponse<Void> markRunning(
+            @RequestHeader(INTERNAL_API_KEY_HEADER) String internalApiKey,
+            @PathVariable String taskId,
+            @Valid @RequestBody AnalysisWorkerRunningRequest request
+    ) {
+        internalApiKeyValidator.validate(internalApiKey);
+        analysisWorkerBridgeService.markRunning(taskId, request.workerId(), request.retryCount(), request.submittedAt());
+        return ApiResponse.onSuccess("자소서 분석 worker 작업 시작 상태를 반영했습니다.");
+    }
+
+    @PostMapping("/tasks/{taskId}/retry")
+    public ApiResponse<Void> markRetry(
+            @RequestHeader(INTERNAL_API_KEY_HEADER) String internalApiKey,
+            @PathVariable String taskId,
+            @Valid @RequestBody AnalysisWorkerRetryRequest request
+    ) {
+        internalApiKeyValidator.validate(internalApiKey);
+        analysisWorkerBridgeService.markRetry(
+                taskId,
+                request.failureReason(),
+                request.errorMessage(),
+                request.retryCount(),
+                request.workerId(),
+                request.queueLatencyMillis()
+        );
+        return ApiResponse.onSuccess("자소서 분석 worker 작업 재시도 상태를 반영했습니다.");
+    }
+
+    @PostMapping("/tasks/{taskId}/failed")
+    public ApiResponse<Void> failTask(
+            @RequestHeader(INTERNAL_API_KEY_HEADER) String internalApiKey,
+            @PathVariable String taskId,
+            @Valid @RequestBody AnalysisWorkerFailureRequest request
+    ) {
+        internalApiKeyValidator.validate(internalApiKey);
+        analysisWorkerBridgeService.failTask(
+                taskId,
+                request.failureReason(),
+                request.errorMessage(),
+                request.retryCount(),
+                request.workerId(),
+                request.queueLatencyMillis()
+        );
+        return ApiResponse.onSuccess("자소서 분석 worker 작업 실패 상태를 반영했습니다.");
+    }
+
+    @PostMapping("/context")
+    public ApiResponse<AnalysisWorkerContextResponse> getContext(
+            @RequestHeader(INTERNAL_API_KEY_HEADER) String internalApiKey,
+            @Valid @RequestBody AnalysisWorkerContextRequest request
+    ) {
+        internalApiKeyValidator.validate(internalApiKey);
+        return ApiResponse.onSuccess(
+                "자소서 분석 worker 컨텍스트 조회에 성공했습니다.",
+                analysisWorkerBridgeService.getContext(request.taskId(), request.userId(), request.mockApplyId())
+        );
+    }
+
+    @PostMapping("/tasks/{taskId}/complete")
+    public ApiResponse<AnalysisResponse> completeTask(
+            @RequestHeader(INTERNAL_API_KEY_HEADER) String internalApiKey,
+            @PathVariable String taskId,
+            @Valid @RequestBody AnalysisWorkerCompleteRequest request
+    ) {
+        internalApiKeyValidator.validate(internalApiKey);
+        return ApiResponse.onSuccess(
+                "자소서 분석 worker 작업 완료 상태를 반영했습니다.",
+                analysisWorkerBridgeService.completeTask(taskId, request)
+        );
+    }
+
+    @GetMapping("/tasks/{taskId}")
+    public ApiResponse<AnalysisAsyncStatusResponse> getTask(
+            @RequestHeader(INTERNAL_API_KEY_HEADER) String internalApiKey,
+            @PathVariable String taskId
+    ) {
+        internalApiKeyValidator.validate(internalApiKey);
+        return ApiResponse.onSuccess(
+                "자소서 분석 worker 작업 상태 조회에 성공했습니다.",
+                analysisAsyncTaskService.getTaskStatusByTaskId(taskId)
+        );
+    }
+}

src/main/java/com/jobdri/jobdri_api/domain/analysis/dto/response/AnalysisAsyncStatusResponse.java

@@ -11,7 +11,14 @@ public record AnalysisAsyncStatusResponse(
         String status,
         String message,
         String error,
+        String failureReason,
+        String workerId,
+        Integer retryCount,
+        Integer maxRetryCount,
+        Long queueLatencyMillis,
         LocalDateTime createdAt,
+        LocalDateTime submittedAt,
+        LocalDateTime lastAttemptAt,
         LocalDateTime startedAt,
         LocalDateTime completedAt,
         AnalysisResponse result

src/main/java/com/jobdri/jobdri_api/domain/analysis/dto/worker/AnalysisTaskMessage.java

@@ -0,0 +1,37 @@
+package com.jobdri.jobdri_api.domain.analysis.dto.worker;
+
+import lombok.Builder;
+
+import java.time.Instant;
+import java.util.UUID;
+
+@Builder
+public record AnalysisTaskMessage(
+        String messageId,
+        String taskType,
+        String taskId,
+        Long userId,
+        Long mockApplyId,
+        int retryCount,
+        int maxRetryCount,
+        Instant submittedAt
+) {
+
+    public static AnalysisTaskMessage of(
+            String taskId,
+            Long userId,
+            Long mockApplyId,
+            int maxRetryCount
+    ) {
+        return AnalysisTaskMessage.builder()
+                .messageId(UUID.randomUUID().toString())
+                .taskType("ANALYSIS")
+                .taskId(taskId)
+                .userId(userId)
+                .mockApplyId(mockApplyId)
+                .retryCount(0)
+                .maxRetryCount(Math.max(0, maxRetryCount))
+                .submittedAt(Instant.now())
+                .build();
+    }
+}

src/main/java/com/jobdri/jobdri_api/domain/analysis/dto/worker/AnalysisWorkerCompleteRequest.java

@@ -0,0 +1,15 @@
+package com.jobdri.jobdri_api.domain.analysis.dto.worker;
+
+import com.jobdri.jobdri_api.domain.analysis.dto.llm.AnalysisLlmResponse;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+
+public record AnalysisWorkerCompleteRequest(
+        @NotNull Long userId,
+        @NotNull Long mockApplyId,
+        @Valid @NotNull AnalysisLlmResponse llmResponse,
+        @NotBlank String workerId,
+        Long queueLatencyMillis
+) {
+}

src/main/java/com/jobdri/jobdri_api/domain/analysis/dto/worker/AnalysisWorkerContextRequest.java

@@ -0,0 +1,11 @@
+package com.jobdri.jobdri_api.domain.analysis.dto.worker;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+
+public record AnalysisWorkerContextRequest(
+        @NotBlank String taskId,
+        @NotNull Long userId,
+        @NotNull Long mockApplyId
+) {
+}

src/main/java/com/jobdri/jobdri_api/domain/analysis/dto/worker/AnalysisWorkerContextResponse.java

@@ -0,0 +1,25 @@
+package com.jobdri.jobdri_api.domain.analysis.dto.worker;
+
+import java.util.List;
+
+public record AnalysisWorkerContextResponse(
+        Long userId,
+        Long mockApplyId,
+        String companyName,
+        String jobTitle,
+        String task,
+        String requirements,
+        String preferredQualifications,
+        String bigClassificationName,
+        String middleClassificationName,
+        String detailClassificationName,
+        List<AnalysisWorkerQuestionItem> questions
+) {
+    public record AnalysisWorkerQuestionItem(
+            Long questionId,
+            String question,
+            String answer,
+            int charLimit
+    ) {
+    }
+}

src/main/java/com/jobdri/jobdri_api/domain/analysis/dto/worker/AnalysisWorkerFailureRequest.java

@@ -0,0 +1,15 @@
+package com.jobdri.jobdri_api.domain.analysis.dto.worker;
+
+import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.FailureReason;
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+
+public record AnalysisWorkerFailureRequest(
+        @NotBlank String errorMessage,
+        @NotNull FailureReason failureReason,
+        @Min(0) int retryCount,
+        @NotBlank String workerId,
+        Long queueLatencyMillis
+) {
+}

src/main/java/com/jobdri/jobdri_api/domain/analysis/dto/worker/AnalysisWorkerRetryRequest.java

@@ -0,0 +1,15 @@
+package com.jobdri.jobdri_api.domain.analysis.dto.worker;
+
+import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.FailureReason;
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+
+public record AnalysisWorkerRetryRequest(
+        @NotBlank String errorMessage,
+        @NotNull FailureReason failureReason,
+        @Min(0) int retryCount,
+        @NotBlank String workerId,
+        Long queueLatencyMillis
+) {
+}

src/main/java/com/jobdri/jobdri_api/domain/analysis/dto/worker/AnalysisWorkerRunningRequest.java

@@ -0,0 +1,13 @@
+package com.jobdri.jobdri_api.domain.analysis.dto.worker;
+
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.NotBlank;
+
+import java.time.Instant;
+
+public record AnalysisWorkerRunningRequest(
+        @NotBlank String workerId,
+        @Min(0) int retryCount,
+        Instant submittedAt
+) {
+}