동적 워크플로우 기반 배치 시스템 초기 설정

apps/user-service/build.gradle

@@ -40,6 +40,9 @@ dependencies {
 	// MyBatis
 	implementation 'org.mybatis.spring.boot:mybatis-spring-boot-starter:3.0.5'
 
+	// batch
+	implementation 'org.springframework.boot:spring-boot-starter-batch'
+
 	// Log4j2 - 모든 모듈을 2.22.1로 통일
 	implementation 'org.springframework.boot:spring-boot-starter-log4j2'
 	implementation 'org.apache.logging.log4j:log4j-core:2.22.1'

apps/user-service/src/main/java/com/gltkorea/icebang/UserServiceApplication.java

@@ -1,9 +1,13 @@
 package com.gltkorea.icebang;
 
 import org.mybatis.spring.annotation.MapperScan;
+import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
+@EnableScheduling
+@EnableBatchProcessing
 @SpringBootApplication
 @MapperScan("com.gltkorea.icebang.mapper")
 public class UserServiceApplication {

apps/user-service/src/main/java/com/gltkorea/icebang/config/scheduler/SchedulerConfig.java

@@ -0,0 +1,28 @@
+package com.gltkorea.icebang.config.scheduler;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+/** 동적 스케줄링을 위한 TaskScheduler Bean을 설정하는 클래스 */
+@Configuration
+public class SchedulerConfig {
+
+  @Bean
+  public TaskScheduler taskScheduler() {
+    // ThreadPool 기반의 TaskScheduler를 생성합니다.
+    ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+
+    // 스케줄러가 사용할 스레드 풀의 크기를 설정합니다.
+    // 동시에 실행될 수 있는 스케줄 작업의 최대 개수입니다.
+    scheduler.setPoolSize(10);
+
+    // 스레드 이름의 접두사를 설정하여 로그 추적을 용이하게 합니다.
+    scheduler.setThreadNamePrefix("dynamic-scheduler-");
+
+    // 스케줄러를 초기화합니다.
+    scheduler.initialize();
+    return scheduler;
+  }
+}

apps/user-service/src/main/java/com/gltkorea/icebang/domain/batch/job/BlogContentJobConfig.java

@@ -0,0 +1,51 @@
+package com.gltkorea.icebang.domain.batch.job;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.job.builder.JobBuilder;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.builder.StepBuilder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.transaction.PlatformTransactionManager;
+
+import com.gltkorea.icebang.domain.batch.tasklet.ContentGenerationTasklet;
+import com.gltkorea.icebang.domain.batch.tasklet.KeywordExtractionTasklet;
+
+import lombok.RequiredArgsConstructor;
+
+@Configuration
+@RequiredArgsConstructor
+public class BlogContentJobConfig {
+
+  // 변경점 1: Factory 대신 실제 Tasklet만 필드로 주입받습니다.
+  private final KeywordExtractionTasklet keywordExtractionTasklet;
+  private final ContentGenerationTasklet contentGenerationTasklet;
+
+  @Bean
+  public Job blogContentJob(
+      JobRepository jobRepository, Step keywordExtractionStep, Step contentGenerationStep) {
+    return new JobBuilder("blogContentJob", jobRepository) // 변경점 2: JobBuilder를 직접 생성합니다.
+        .start(keywordExtractionStep)
+        .next(contentGenerationStep)
+        .build();
+  }
+
+  @Bean
+  public Step keywordExtractionStep(
+      JobRepository jobRepository, PlatformTransactionManager transactionManager) {
+    return new StepBuilder("keywordExtractionStep", jobRepository) // 변경점 3: StepBuilder를 직접 생성합니다.
+        .tasklet(
+            keywordExtractionTasklet,
+            transactionManager) // 변경점 4: tasklet에 transactionManager를 함께 전달합니다.
+        .build();
+  }
+
+  @Bean
+  public Step contentGenerationStep(
+      JobRepository jobRepository, PlatformTransactionManager transactionManager) {
+    return new StepBuilder("contentGenerationStep", jobRepository)
+        .tasklet(contentGenerationTasklet, transactionManager)
+        .build();
+  }
+}

apps/user-service/src/main/java/com/gltkorea/icebang/domain/batch/tasklet/ContentGenerationTasklet.java

@@ -0,0 +1,49 @@
+package com.gltkorea.icebang.domain.batch.tasklet;
+
+import java.util.List;
+
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.repeat.RepeatStatus;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class ContentGenerationTasklet implements Tasklet {
+
+  // private final ContentService contentService; // 비즈니스 로직을 담은 서비스
+  // private final FastApiClient fastApiClient; // FastAPI 통신을 위한 클라이언트
+
+  @Override
+  public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext)
+      throws Exception {
+    log.info(">>>> [Step 2] ContentGenerationTasklet executed.");
+
+    // --- 핵심: JobExecutionContext에서 이전 Step의 결과물 가져오기 ---
+    ExecutionContext jobExecutionContext =
+        chunkContext.getStepContext().getStepExecution().getJobExecution().getExecutionContext();
+
+    // KeywordExtractionTasklet이 저장한 "extractedKeywordIds" Key로 데이터 조회
+    List<Long> keywordIds = (List<Long>) jobExecutionContext.get("extractedKeywordIds");
+
+    if (keywordIds == null || keywordIds.isEmpty()) {
+      log.warn(">>>> No keyword IDs found from previous step. Skipping content generation.");
+      return RepeatStatus.FINISHED;
+    }
+
+    log.info(">>>> Received Keyword IDs for content generation: {}", keywordIds);
+
+    // TODO: 1. 전달받은 키워드 ID 목록으로 DB에서 상세 정보 조회
+    // TODO: 2. 각 키워드/상품 정보에 대해 외부 AI 서비스(FastAPI/LangChain)를 호출하여 콘텐츠 생성을 요청
+    // TODO: 3. 생성된 콘텐츠를 DB에 저장
+
+    log.info(">>>> [Step 2] ContentGenerationTasklet finished.");
+    return RepeatStatus.FINISHED;
+  }
+}

apps/user-service/src/main/java/com/gltkorea/icebang/domain/batch/tasklet/KeywordExtractionTasklet.java

@@ -0,0 +1,47 @@
+package com.gltkorea.icebang.domain.batch.tasklet;
+
+import java.util.List;
+
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.repeat.RepeatStatus;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class KeywordExtractionTasklet implements Tasklet {
+
+  // private final TrendKeywordService trendKeywordService; // 비즈니스 로직을 담은 서비스
+
+  @Override
+  public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext)
+      throws Exception {
+    log.info(">>>> [Step 1] KeywordExtractionTasklet executed.");
+
+    // TODO: 1. DB에서 카테고리 정보 조회
+    // TODO: 2. 외부 API 또는 내부 로직을 통해 트렌드 키워드 추출
+    // TODO: 3. 추출된 키워드를 DB에 저장
+
+    // --- 핵심: 다음 Step에 전달할 데이터 생성 ---
+    // 예시: 새로 생성된 키워드 ID 목록을 가져왔다고 가정
+    List<Long> extractedKeywordIds = List.of(1L, 2L, 3L); // 실제로는 DB 저장 후 반환된 ID 목록
+    log.info(">>>> Extracted Keyword IDs: {}", extractedKeywordIds);
+
+    // --- 핵심: JobExecutionContext에 결과물 저장 ---
+    // JobExecution 전체에서 공유되는 컨텍스트를 가져옵니다.
+    ExecutionContext jobExecutionContext =
+        chunkContext.getStepContext().getStepExecution().getJobExecution().getExecutionContext();
+
+    // "extractedKeywordIds" 라는 Key로 데이터 저장
+    jobExecutionContext.put("extractedKeywordIds", extractedKeywordIds);
+
+    log.info(">>>> [Step 1] KeywordExtractionTasklet finished.");
+    return RepeatStatus.FINISHED;
+  }
+}

apps/user-service/src/main/java/com/gltkorea/icebang/domain/schedule/model/Schedule.java

@@ -0,0 +1,14 @@
+package com.gltkorea.icebang.domain.schedule.model;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+public class Schedule {
+  private Long scheduleId;
+  private Long workflowId;
+  private String cronExpression;
+  private boolean isActive;
+  // ... 기타 필요한 컬럼
+}

apps/user-service/src/main/java/com/gltkorea/icebang/domain/schedule/runner/SchedulerInitializer.java

@@ -0,0 +1,31 @@
+package com.gltkorea.icebang.domain.schedule.runner;
+
+import java.util.List;
+
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+import com.gltkorea.icebang.domain.schedule.model.Schedule;
+import com.gltkorea.icebang.domain.schedule.service.DynamicSchedulerService;
+import com.gltkorea.icebang.mapper.ScheduleMapper;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class SchedulerInitializer implements ApplicationRunner {
+
+  private final ScheduleMapper scheduleMapper;
+  private final DynamicSchedulerService dynamicSchedulerService;
+
+  @Override
+  public void run(ApplicationArguments args) {
+    log.info(">>>> Initializing schedules from database...");
+    List<Schedule> activeSchedules = scheduleMapper.findAllByIsActive(true);
+    activeSchedules.forEach(dynamicSchedulerService::register);
+    log.info(">>>> {} active schedules have been registered.", activeSchedules.size());
+  }
+}

apps/user-service/src/main/java/com/gltkorea/icebang/domain/schedule/service/DynamicSchedulerService.java

@@ -0,0 +1,66 @@
+package com.gltkorea.icebang.domain.schedule.service;
+
+import java.time.LocalDateTime;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.context.ApplicationContext;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.support.CronTrigger;
+import org.springframework.stereotype.Service;
+
+import com.gltkorea.icebang.domain.schedule.model.Schedule;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DynamicSchedulerService {
+
+  private final TaskScheduler taskScheduler;
+  private final JobLauncher jobLauncher;
+  private final ApplicationContext applicationContext;
+  private final Map<Long, ScheduledFuture<?>> scheduledTasks = new ConcurrentHashMap<>();
+
+  public void register(Schedule schedule) {
+    // TODO: schedule.getWorkflowId()를 기반으로 실행할 Job의 이름을 DB에서 조회
+    String jobName = "blogContentJob"; // 예시
+    Job jobToRun = applicationContext.getBean(jobName, Job.class);
+
+    Runnable runnable =
+        () -> {
+          try {
+            JobParametersBuilder paramsBuilder = new JobParametersBuilder();
+            paramsBuilder.addString("runAt", LocalDateTime.now().toString());
+            paramsBuilder.addLong("scheduleId", schedule.getScheduleId());
+            jobLauncher.run(jobToRun, paramsBuilder.toJobParameters());
+          } catch (Exception e) {
+            log.error(
+                "Failed to run scheduled job for scheduleId: {}", schedule.getScheduleId(), e);
+          }
+        };
+
+    CronTrigger trigger = new CronTrigger(schedule.getCronExpression());
+    ScheduledFuture<?> future = taskScheduler.schedule(runnable, trigger);
+    scheduledTasks.put(schedule.getScheduleId(), future);
+    log.info(
+        ">>>> Schedule registered: id={}, cron={}",
+        schedule.getScheduleId(),
+        schedule.getCronExpression());
+  }
+
+  public void remove(Long scheduleId) {
+    ScheduledFuture<?> future = scheduledTasks.get(scheduleId);
+    if (future != null) {
+      future.cancel(true);
+      scheduledTasks.remove(scheduleId);
+      log.info(">>>> Schedule removed: id={}", scheduleId);
+    }
+  }
+}

apps/user-service/src/main/java/com/gltkorea/icebang/mapper/ScheduleMapper.java

@@ -0,0 +1,12 @@
+package com.gltkorea.icebang.mapper;
+
+import java.util.List;
+
+import org.apache.ibatis.annotations.Mapper;
+
+import com.gltkorea.icebang.domain.schedule.model.Schedule;
+
+@Mapper
+public interface ScheduleMapper {
+  List<Schedule> findAllByIsActive(boolean isActive);
+}

apps/user-service/src/main/resources/mybatis/mapper/ScheduleMapper.xml

@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+
+<mapper namespace="com.gltkorea.icebang.mapper.ScheduleMapper">
+
+    <select id="findAllByIsActive" resultType="com.gltkorea.icebang.domain.schedule.model.Schedule"> SELECT
+                                                                                                         schedule_id AS scheduleId,
+                                                                                                         workflow_id AS workflowId,
+                                                                                                         cron_expression AS cronExpression,
+                                                                                                         is_active AS isActive
+                                                                                                     FROM
+                                                                                                         schedule
+                                                                                                     WHERE
+                                                                                                         is_active = #{isActive}
+    </select>
+
+</mapper>