Asynchronous APIs

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

동기식 통신 흐름:
클라이언트 → 요청 → 서버
클라이언트 (대기…) 서버 (처리 중…)
클라이언트 ← 응답 ← 서버

비동기식 통신 흐름:
클라이언트 → 요청 → 서버
클라이언트 ← 확인 응답 ← 서버
클라이언트 (다른 작업 수행) 서버 (백그라운드 처리…)
클라이언트 ← 결과 알림 ← 서버 (처리 완료)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

// 작업 제출 엔드포인트
@PostMapping("/api/jobs")
public ResponseEntity<JobResponse> submitJob(@RequestBody JobRequest request) {
    // 작업 ID 생성 및 백그라운드 작업 시작
    String jobId = jobService.createJob(request);
    
    // 즉시 응답 반환
    return ResponseEntity
        .accepted()
        .body(new JobResponse(jobId, "PENDING", "/api/jobs/" + jobId));
}

// 작업 상태 확인 엔드포인트 (폴링용)
@GetMapping("/api/jobs/{jobId}")
public ResponseEntity<JobResponse> getJobStatus(@PathVariable String jobId) {
    Job job = jobService.getJob(jobId);
    
    if (job == null) {
        return ResponseEntity.notFound().build();
    }
    
    JobResponse response = new JobResponse(
        job.getId(),
        job.getStatus().toString(),
        "/api/jobs/" + jobId
    );
    
    // 작업이 완료된 경우 결과 포함
    if (job.getStatus() == JobStatus.COMPLETED) {
        response.setResult(job.getResult());
    }
    
    return ResponseEntity.ok(response);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

// 작업 제출 엔드포인트 (웹훅 URL 포함)
@PostMapping("/api/async-tasks")
public ResponseEntity<TaskResponse> submitTask(
        @RequestBody TaskRequest request,
        @RequestParam String webhookUrl) {
    
    // 작업 ID 생성 및 요청 정보 저장
    String taskId = UUID.randomUUID().toString();
    asyncTaskService.saveTask(taskId, request, webhookUrl);
    
    // 비동기 작업 실행
    CompletableFuture.runAsync(() -> {
        try {
            // 작업 실행
            TaskResult result = asyncTaskService.executeTask(taskId, request);
            
            // 웹훅 호출
            webhookService.sendWebhook(webhookUrl, taskId, result);
        } catch (Exception e) {
            // 오류 처리 및 웹훅으로 오류 전송
            webhookService.sendErrorWebhook(webhookUrl, taskId, e.getMessage());
        }
    });
    
    // 즉시 응답
    return ResponseEntity
        .accepted()
        .body(new TaskResponse(taskId, "PROCESSING", "/api/async-tasks/" + taskId));
}

// 웹훅 서비스
@Service
public class WebhookService {
    
    private final RestTemplate restTemplate;
    
    public void sendWebhook(String webhookUrl, String taskId, TaskResult result) {
        WebhookPayload payload = new WebhookPayload(
            taskId,
            "COMPLETED",
            result
        );
        
        try {
            restTemplate.postForEntity(webhookUrl, payload, Void.class);
            log.info("Webhook sent for task: {}", taskId);
        } catch (Exception e) {
            log.error("Failed to send webhook for task {}: {}", taskId, e.getMessage());
            // 재시도 로직 구현 가능
        }
    }
    
    public void sendErrorWebhook(String webhookUrl, String taskId, String errorMessage) {
        WebhookPayload payload = new WebhookPayload(
            taskId,
            "FAILED",
            Map.of("error", errorMessage)
        );
        
        try {
            restTemplate.postForEntity(webhookUrl, payload, Void.class);
            log.info("Error webhook sent for task: {}", taskId);
        } catch (Exception e) {
            log.error("Failed to send error webhook for task {}: {}", taskId, e.getMessage());
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

@GetMapping("/api/jobs/{jobId}/longpoll")
public DeferredResult<ResponseEntity<JobResult>> longPollJobResult(
        @PathVariable String jobId,
        @RequestParam(defaultValue = "30000") long timeout) {
    
    // DeferredResult 생성 (타임아웃 설정)
    DeferredResult<ResponseEntity<JobResult>> deferredResult = new DeferredResult<>(timeout);
    
    // 타임아웃 처리
    deferredResult.onTimeout(() -> {
        deferredResult.setResult(
            ResponseEntity.ok(new JobResult(jobId, "PROCESSING", null))
        );
    });
    
    // 현재 작업 상태 확인
    Job job = jobService.getJob(jobId);
    
    if (job == null) {
        deferredResult.setResult(ResponseEntity.notFound().build());
    } else if (job.isCompleted() || job.isFailed()) {
        // 이미 완료된 경우 즉시 결과 반환
        deferredResult.setResult(
            ResponseEntity.ok(new JobResult(jobId, job.getStatus().toString(), job.getResult()))
        );
    } else {
        // 작업이 아직 진행 중이면 완료 리스너 등록
        jobCompletionService.addCompletionListener(jobId, job -> {
            deferredResult.setResult(
                ResponseEntity.ok(new JobResult(jobId, job.getStatus().toString(), job.getResult()))
            );
        });
    }
    
    return deferredResult;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

@GetMapping(value = "/api/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter subscribeToEvents() {
    // SSE 이미터 생성 (타임아웃 설정)
    SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
    
    // 클라이언트 등록
    eventService.addClient(emitter);
    
    // 연결 종료 시 정리
    emitter.onCompletion(() -> eventService.removeClient(emitter));
    emitter.onTimeout(() -> eventService.removeClient(emitter));
    emitter.onError(e -> eventService.removeClient(emitter));
    
    // 초기 연결 확인 이벤트 전송
    try {
        emitter.send(SseEmitter.event()
            .name("CONNECTED")
            .data("Connection established")
            .id(UUID.randomUUID().toString())
            .build());
    } catch (IOException e) {
        emitter.completeWithError(e);
    }
    
    return emitter;
}

// 이벤트 발생 시 모든 클라이언트에게 전송
@Service
public class EventService {
    
    private final List<SseEmitter> clients = new CopyOnWriteArrayList<>();
    
    public void addClient(SseEmitter emitter) {
        clients.add(emitter);
    }
    
    public void removeClient(SseEmitter emitter) {
        clients.remove(emitter);
    }
    
    public void broadcastEvent(String eventName, Object data) {
        List<SseEmitter> deadEmitters = new ArrayList<>();
        
        clients.forEach(emitter -> {
            try {
                emitter.send(SseEmitter.event()
                    .name(eventName)
                    .data(data)
                    .id(UUID.randomUUID().toString())
                    .build());
            } catch (Exception e) {
                deadEmitters.add(emitter);
            }
        });
        
        // 실패한 이미터 제거
        clients.removeAll(deadEmitters);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/topic");  // 구독 경로
        config.setApplicationDestinationPrefixes("/app");  // 메시지 전송 경로
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/websocket")
            .setAllowedOrigins("*")
            .withSockJS();  // 폴백 옵션
    }
}

@Controller
public class TaskWebSocketController {

    @MessageMapping("/tasks/submit")
    @SendTo("/topic/task-updates")
    public TaskResponse submitTask(TaskRequest request) {
        // 작업 ID 생성
        String taskId = UUID.randomUUID().toString();
        
        // 백그라운드 작업 시작
        CompletableFuture.runAsync(() -> {
            try {
                // 작업 실행 중 상태 업데이트
                simpMessagingTemplate.convertAndSend(
                    "/topic/task-updates/" + taskId,
                    new TaskUpdate(taskId, "PROCESSING", "Task in progress…")
                );
                
                // 작업 실행
                TaskResult result = taskService.executeTask(request);
                
                // 완료 알림
                simpMessagingTemplate.convertAndSend(
                    "/topic/task-updates/" + taskId,
                    new TaskUpdate(taskId, "COMPLETED", result)
                );
            } catch (Exception e) {
                // 오류 알림
                simpMessagingTemplate.convertAndSend(
                    "/topic/task-updates/" + taskId,
                    new TaskUpdate(taskId, "FAILED", e.getMessage())
                );
            }
        });
        
        // 초기 응답
        return new TaskResponse(taskId, "SUBMITTED", "/topic/task-updates/" + taskId);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

// API 컨트롤러
@RestController
@RequestMapping("/api/jobs")
public class JobController {

    private final JobService jobService;
    private final RabbitTemplate rabbitTemplate;
    
    @PostMapping
    public ResponseEntity<JobResponse> submitJob(@RequestBody JobRequest request) {
        // 작업 생성 및 초기 상태 저장
        String jobId = UUID.randomUUID().toString();
        jobService.saveJob(new Job(jobId, request, JobStatus.PENDING));
        
        // 메시지 큐로 작업 전송
        rabbitTemplate.convertAndSend(
            "job-exchange",
            "job.submit",
            new JobMessage(jobId, request)
        );
        
        // 즉시 응답
        return ResponseEntity
            .accepted()
            .body(new JobResponse(jobId, "PENDING", "/api/jobs/" + jobId));
    }
    
    @GetMapping("/{jobId}")
    public ResponseEntity<JobResponse> getJobStatus(@PathVariable String jobId) {
        Job job = jobService.getJob(jobId);
        
        if (job == null) {
            return ResponseEntity.notFound().build();
        }
        
        JobResponse response = new JobResponse(
            job.getId(),
            job.getStatus().toString(),
            "/api/jobs/" + jobId
        );
        
        if (job.getStatus() == JobStatus.COMPLETED || job.getStatus() == JobStatus.FAILED) {
            response.setResult(job.getResult());
        }
        
        return ResponseEntity.ok(response);
    }
}

// 메시지 소비자
@Component
public class JobProcessor {

    private final JobService jobService;
    
    @RabbitListener(queues = "job-queue")
    public void processJob(JobMessage message) {
        String jobId = message.getJobId();
        JobRequest request = message.getRequest();
        
        try {
            // 작업 상태 업데이트
            jobService.updateJobStatus(jobId, JobStatus.PROCESSING);
            
            // 작업 실행
            JobResult result = executeJob(request);
            
            // 성공 결과 저장
            jobService.completeJob(jobId, result);
        } catch (Exception e) {
            // 실패 결과 저장
            jobService.failJob(jobId, e.getMessage());
        }
    }
    
    private JobResult executeJob(JobRequest request) {
        // 실제 작업 처리 로직
        // …
        return new JobResult("Some result data");
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

// task_service.proto
syntax = "proto3";

package task;

service TaskService {
  // 서버 스트리밍 RPC
  rpc SubmitTask(TaskRequest) returns (stream TaskUpdate);
}

message TaskRequest {
  string name = 1;
  map<string, string> parameters = 2;
}

message TaskUpdate {
  string task_id = 1;
  string status = 2;
  float progress = 3;
  string message = 4;
  bytes result = 5;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

// gRPC 서비스 구현
public class TaskServiceImpl extends TaskServiceGrpc.TaskServiceImplBase {
    
    @Override
    public void submitTask(TaskRequest request, StreamObserver<TaskUpdate> responseObserver) {
        // 작업 ID 생성
        String taskId = UUID.randomUUID().toString();
        
        // 초기 업데이트 전송
        responseObserver.onNext(TaskUpdate.newBuilder()
            .setTaskId(taskId)
            .setStatus("STARTED")
            .setProgress(0.0f)
            .setMessage("Task started")
            .build());
        
        // 백그라운드에서 작업 처리
        executorService.submit(() -> {
            try {
                // 25% 진행 상황 업데이트
                responseObserver.onNext(TaskUpdate.newBuilder()
                    .setTaskId(taskId)
                    .setStatus("PROCESSING")
                    .setProgress(0.25f)
                    .setMessage("Processing started")
                    .build());
                
                Thread.sleep(1000); // 작업 시뮬레이션
                
                // 50% 진행 상황 업데이트
                responseObserver.onNext(TaskUpdate.newBuilder()
                    .setTaskId(taskId)
                    .setStatus("PROCESSING")
                    .setProgress(0.5f)
                    .setMessage("Processing halfway done")
                    .build());
                
                Thread.sleep(1000); // 작업 시뮬레이션
                
                // 75% 진행 상황 업데이트
                responseObserver.onNext(TaskUpdate.newBuilder()
                    .setTaskId(taskId)
                    .setStatus("PROCESSING")
                    .setProgress(0.75f)
                    .setMessage("Processing almost complete")
                    .build());
                
                Thread.sleep(1000); // 작업 시뮬레이션
                
                // 작업 결과 생성
                ByteString result = processTask(request);
                
                // 완료 업데이트 전송
                responseObserver.onNext(TaskUpdate.newBuilder()
                    .setTaskId(taskId)
                    .setStatus("COMPLETED")
                    .setProgress(1.0f)
                    .setMessage("Task completed successfully")
                    .setResult(result)
                    .build());
                
                // 스트림 완료
                responseObserver.onCompleted();
            } catch (Exception e) {
                // 오류 업데이트 전송
                responseObserver.onNext(TaskUpdate.newBuilder()
                    .setTaskId(taskId)
                    .setStatus("FAILED")
                    .setMessage("Error: " + e.getMessage())
                    .build());
                
                // 스트림 완료
                responseObserver.onCompleted();
            }
        });
    }
    
    private ByteString processTask(TaskRequest request) {
        // 실제 작업 처리 로직
        // …
        return ByteString.copyFromUtf8("Task result data");
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// 작업 모델 예시
public class AsyncJob {
    private final String id;
    private JobStatus status;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
    private LocalDateTime completedAt;
    private Object result;
    private String errorMessage;
    
    // 작업 상태 열거형
    public enum JobStatus {
        PENDING,      // 대기 중
        PROCESSING,   // 처리 중
        COMPLETED,    // 완료됨
        FAILED,       // 실패함
        CANCELLED     // 취소됨
    }
    
    // 생성자, 게터, 세터 등
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// 진행 상황 업데이트 엔드포인트
@GetMapping("/api/jobs/{jobId}/progress")
public ResponseEntity<JobProgress> getJobProgress(@PathVariable String jobId) {
    Job job = jobService.getJob(jobId);
    
    if (job == null) {
        return ResponseEntity.notFound().build();
    }
    
    JobProgress progress = new JobProgress(
        job.getId(),
        job.getStatus().toString(),
        job.getProgressPercentage(),
        job.getCurrentStep(),
        job.getTotalSteps(),
        job.getEstimatedTimeRemaining()
    );
    
    return ResponseEntity.ok(progress);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// 멱등성 키를 사용한 요청 처리
@PostMapping("/api/payments")
public ResponseEntity<PaymentResponse> processPayment(
        @RequestBody PaymentRequest request,
        @RequestHeader(value = "Idempotency-Key", required = true) String idempotencyKey) {
    
    // 멱등성 키로 기존 요청 확인
    Payment existingPayment = paymentService.findByIdempotencyKey(idempotencyKey);
    
    if (existingPayment != null) {
        // 기존 응답 반환
        return ResponseEntity
            .status(existingPayment.getStatus() == PaymentStatus.COMPLETED ? 
                    HttpStatus.OK : HttpStatus.ACCEPTED)
            .body(mapToResponse(existingPayment));
    }
    
    // 새 요청 처리
    String paymentId = UUID.randomUUID().toString();
    
    Payment payment = new Payment(
        paymentId,
        request,
        PaymentStatus.PENDING,
        idempotencyKey
    );
    paymentService.save(payment);
    
    // 비동기 처리 시작
    paymentProcessor.processAsync(paymentId);
    
    return ResponseEntity
        .accepted()
        .body(new PaymentResponse(paymentId, "PENDING", "/api/payments/" + paymentId));
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// 작업 제한 시간 설정
@PostMapping("/api/data-processing")
public ResponseEntity<JobResponse> startDataProcessing(
        @RequestBody ProcessingRequest request,
        @RequestParam(defaultValue = "3600") int timeoutSeconds) {
    
    // 작업 생성
    String jobId = UUID.randomUUID().toString();
    LocalDateTime expiresAt = LocalDateTime.now().plusSeconds(timeoutSeconds);
    
    Job job = new Job(jobId, request, JobStatus.PENDING, expiresAt);
    jobRepository.save(job);
    
    // 비동기 처리 시작 (타임아웃 포함)
    scheduledExecutorService.schedule(() -> {
        Job currentJob = jobRepository.findById(jobId).orElse(null);
        if (currentJob != null && 
            (currentJob.getStatus() == JobStatus.PENDING || 
             currentJob.getStatus() == JobStatus.PROCESSING)) {
            
            jobRepository.updateStatus(jobId, JobStatus.FAILED, "Operation timed out");
        }
    }, timeoutSeconds, TimeUnit.SECONDS);
    
    dataProcessingService.processAsync(jobId);
    
    return ResponseEntity
        .accepted()
        .body(new JobResponse(jobId, "PENDING", expiresAt));
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

// 재시도 정책이 있는 메시지 소비자
@Component
public class JobConsumer {

    private final JobService jobService;
    
    @RabbitListener(queues = "job-queue")
    @Retryable(
        value = { TemporaryFailureException.class },
        maxAttempts = 3,
        backoff = @Backoff(delay = 1000, multiplier = 2)
    )
    public void processJob(JobMessage message) {
        String jobId = message.getJobId();
        
        try {
            // 작업 상태 업데이트
            jobService.updateStatus(jobId, JobStatus.PROCESSING);
            
            // 작업 처리
            JobResult result = executeJob(message.getRequest());
            
            // 성공 상태 업데이트
            jobService.completeJob(jobId, result);
        } catch (TemporaryFailureException e) {
            // 일시적 실패 - 재시도됨
            log.warn("Temporary failure processing job {}, will retry: {}", jobId, e.getMessage());
            throw e;  // 재시도를 위해 예외 다시 throw
        } catch (Exception e) {
            // 영구적 실패
            log.error("Permanent failure processing job {}: {}", jobId, e.getMessage());
            jobService.failJob(jobId, e.getMessage());
        }
    }
    
    @Recover
    public void recover(TemporaryFailureException e, JobMessage message) {
        // 최대 재시도 후에도 실패한 경우 처리
        String jobId = message.getJobId();
        log.error("All retries failed for job {}: {}", jobId, e.getMessage());
        jobService.failJob(jobId, "Failed after multiple retries: " + e.getMessage());
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// 작업 취소 엔드포인트
@DeleteMapping("/api/jobs/{jobId}")
public ResponseEntity<JobResponse> cancelJob(@PathVariable String jobId) {
    Job job = jobService.getJob(jobId);
    
    if (job == null) {
        return ResponseEntity.notFound().build();
    }
    
    // 이미 완료되거나 실패한 작업은 취소할 수 없음
    if (job.getStatus() == JobStatus.COMPLETED || 
        job.getStatus() == JobStatus.FAILED ||
        job.getStatus() == JobStatus.CANCELLED) {
        
        return ResponseEntity
            .badRequest()
            .body(new JobResponse(
                job.getId(), 
                job.getStatus().toString(),
                "Cannot cancel job with status: " + job.getStatus()
            ));
    }
    
    // 작업 취소 처리
    jobService.cancelJob(jobId);
    
    return ResponseEntity.ok(new JobResponse(jobId, "CANCELLED", null));
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

// 작업 저장소 인터페이스
public interface JobRepository {
    String createJob(JobRequest request);
    Job findById(String jobId);
    void updateStatus(String jobId, JobStatus status);
    void updateProgress(String jobId, float progress, String message);
    void completeJob(String jobId, Object result);
    void failJob(String jobId, String errorMessage);
    void cancelJob(String jobId);
    List<Job> findExpiredJobs(LocalDateTime expirationTime);
    List<Job> findJobsByStatus(JobStatus status);
}

// Redis를 사용한 작업 저장소 구현
@Repository
public class RedisJobRepository implements JobRepository {
    
    private final StringRedisTemplate redisTemplate;
    private final ObjectMapper objectMapper;
    
    // Hash 작업을 위한 Redis 키 접두사
    private static final String JOB_KEY_PREFIX = "job:";
    
    @Override
    public String createJob(JobRequest request) {
        String jobId = UUID.randomUUID().toString();
        Job job = new Job(
            jobId,
            request,
            JobStatus.PENDING,
            LocalDateTime.now(),
            null,
            null,
            0.0f,
            null,
            null
        );
        
        saveJob(job);
        return jobId;
    }
    
    @Override
    public Job findById(String jobId) {
        String key = JOB_KEY_PREFIX + jobId;
        Map<Object, Object> entries = redisTemplate.opsForHash().entries(key);
        
        if (entries.isEmpty()) {
            return null;
        }
        
        try {
            // Redis 해시에서 Job 객체로 변환
            Job job = new Job();
            job.setId(jobId);
            job.setStatus(JobStatus.valueOf((String) entries.get("status")));
            job.setRequest(objectMapper.readValue(
                (String) entries.get("request"), JobRequest.class));
            
            if (entries.containsKey("progress")) {
                job.setProgress(Float.parseFloat((String) entries.get("progress")));
            }
            
            if (entries.containsKey("message")) {
                job.setMessage((String) entries.get("message"));
            }
            
            if (entries.containsKey("result")) {
                job.setResult(objectMapper.readValue(
                    (String) entries.get("result"), Object.class));
            }
            
            if (entries.containsKey("errorMessage")) {
                job.setErrorMessage((String) entries.get("errorMessage"));
            }
            
            if (entries.containsKey("createdAt")) {
                job.setCreatedAt(LocalDateTime.parse((String) entries.get("createdAt")));
            }
            
            if (entries.containsKey("updatedAt")) {
                job.setUpdatedAt(LocalDateTime.parse((String) entries.get("updatedAt")));
            }
            
            if (entries.containsKey("completedAt")) {
                job.setCompletedAt(LocalDateTime.parse((String) entries.get("completedAt")));
            }
            
            return job;
        } catch (Exception e) {
            throw new RuntimeException("Error deserializing job", e);
        }
    }
    
    // 기타 메서드 구현…
    
    private void saveJob(Job job) {
        String key = JOB_KEY_PREFIX + job.getId();
        Map<String, String> hash = new HashMap<>();
        
        try {
            hash.put("status", job.getStatus().name());
            hash.put("request", objectMapper.writeValueAsString(job.getRequest()));
            hash.put("createdAt", job.getCreatedAt().toString());
            
            if (job.getUpdatedAt() != null) {
                hash.put("updatedAt", job.getUpdatedAt().toString());
            }
            
            if (job.getProgress() > 0) {
                hash.put("progress", String.valueOf(job.getProgress()));
            }
            
            if (job.getMessage() != null) {
                hash.put("message", job.getMessage());
            }
            
            if (job.getResult() != null) {
                hash.put("result", objectMapper.writeValueAsString(job.getResult()));
            }
            
            if (job.getErrorMessage() != null) {
                hash.put("errorMessage", job.getErrorMessage());
            }
            
            if (job.getCompletedAt() != null) {
                hash.put("completedAt", job.getCompletedAt().toString());
            }
            
            redisTemplate.opsForHash().putAll(key, hash);
            
            // 만료 시간 설정 (예: 7일)
            redisTemplate.expire(key, 7, TimeUnit.DAYS);
        } catch (Exception e) {
            throw new RuntimeException("Error serializing job", e);
        }
    }
}