[volume-10] Collect, Stack, Zip

apps/commerce-api/build.gradle.kts

@@ -1,6 +1,6 @@
 dependencies {
     implementation(project(":apps:commerce-core"))
-    
+
     // add-ons
     implementation(project(":modules:jpa"))
     implementation(project(":modules:redis"))
@@ -19,6 +19,9 @@ dependencies {
         exclude(group = "org.yaml", module = "snakeyaml")
     }
 
+    // Spring Batch
+    implementation("org.springframework.boot:spring-boot-starter-batch")
+
     // Resilience4j
     implementation("io.github.resilience4j:resilience4j-spring-boot3:${project.properties["resilience4jVersion"]}")
     implementation("org.springframework.boot:spring-boot-starter-aop")
@@ -29,6 +32,9 @@ dependencies {
     annotationProcessor("jakarta.persistence:jakarta.persistence-api")
     annotationProcessor("jakarta.annotation:jakarta.annotation-api")
 
+    // Spring Batch Test (테스트용)
+    testImplementation("org.springframework.batch:spring-batch-test")
+
     // test-fixtures
     testImplementation(testFixtures(project(":modules:jpa")))
     testImplementation(testFixtures(project(":modules:redis")))

apps/commerce-api/src/main/java/com/loopers/application/ranking/RankingAggregationInfo.java

@@ -0,0 +1,68 @@
+package com.loopers.application.ranking;
+
+import lombok.Getter;
+import org.springframework.batch.core.JobExecution;
+
+/**
+ * 랭킹 집계 배치 실행 결과 정보
+ */
+@Getter
+public class RankingAggregationInfo {
+
+    private final Long jobExecutionId;
+    private final String jobName;
+    private final String status;
+    private final String exitStatus;
+    private final String period; // "weekly" or "monthly"
+    private final String message;
+
+    private RankingAggregationInfo(
+            Long jobExecutionId,
+            String jobName,
+            String status,
+            String exitStatus,
+            String period,
+            String message
+    ) {
+        this.jobExecutionId = jobExecutionId;
+        this.jobName = jobName;
+        this.status = status;
+        this.exitStatus = exitStatus;
+        this.period = period;
+        this.message = message;
+    }
+
+    public static RankingAggregationInfo from(JobExecution jobExecution, String period) {
+        return new RankingAggregationInfo(
+                jobExecution.getId(),
+                jobExecution.getJobInstance().getJobName(),
+                jobExecution.getStatus().name(),
+                jobExecution.getExitStatus() != null ? jobExecution.getExitStatus().getExitCode() : null,
+                period,
+                jobExecution.getExitStatus() != null ? jobExecution.getExitStatus().getExitDescription() : null
+        );
+    }
+
+    public static RankingAggregationInfo combined(
+            RankingAggregationInfo weekly,
+            RankingAggregationInfo monthly
+    ) {
+        boolean bothSuccess = "COMPLETED".equals(weekly.status) && "COMPLETED".equals(monthly.status);
+        String combinedStatus = bothSuccess ? "COMPLETED" : "FAILED";
+        String combinedMessage = String.format(
+                "Weekly: %s, Monthly: %s",
+                weekly.status,
+                monthly.status
+        );
+
+        return new RankingAggregationInfo(
+                null, // combined에는 jobExecutionId 없음
+                "weeklyAndMonthlyRankingJob",
+                combinedStatus,
+                bothSuccess ? "COMPLETED" : "FAILED",
+                "both",
+                combinedMessage
+        );
+    }
+}
+

apps/commerce-api/src/main/java/com/loopers/application/ranking/RankingAggregationScheduler.java

@@ -0,0 +1,47 @@
+package com.loopers.application.ranking;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.time.ZonedDateTime;
+
+/**
+ * 주간, 월간 랭킹 스케줄러
+ * <p>
+ * 매일 자정에 일간 집계 완료 후 주간, 월간 랭킹을 갱신합니다.
+ * 스프링 배치를 사용하여 tb_product_metrics_daily의 일간 데이터를 집계하고,
+ * 주간 및 월간 랭킹을 계산합니다.
+ * 이 작업은 Chunk 단위로 처리하여 대량의 데이터를 효율적으로 처리합니다.
+ * spring batch read process write 패턴을 따릅니다.
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class RankingAggregationScheduler {
+
+    private final RankingAggregationService rankingAggregationService;
+
+    /**
+     * 매일 자정에 실행 (일간 집계 완료 후)
+     * 주간, 월간 랭킹 갱신
+     */
+    @Scheduled(cron = "0 0 0 * * *") // 매일 자정
+    public void aggregateWeeklyAndMonthlyRankings() {
+        try {
+            log.info("스케줄러: 주간 및 월간 랭킹 집계 배치 실행 시작");
+
+            // 어제 날짜 기준으로 집계 (오늘 자정에 실행되므로 어제 데이터까지 집계)
+            ZonedDateTime targetDate = ZonedDateTime.now().minusDays(1);
+
+            rankingAggregationService.executeWeeklyAndMonthlyRanking(targetDate);
+
+            log.info("스케줄러: 주간 및 월간 랭킹 집계 배치 실행 완료");
+
+        } catch (Exception e) {
+            log.error("스케줄러: 주간 및 월간 랭킹 집계 배치 실행 실패", e);
+            // 스케줄러는 예외를 던지지 않고 로그만 남김 (다음 실행에 영향 주지 않음)
+        }
+    }
+}

apps/commerce-api/src/main/java/com/loopers/application/ranking/RankingAggregationService.java

@@ -0,0 +1,172 @@
+package com.loopers.application.ranking;
+
+import com.loopers.support.error.CoreException;
+import com.loopers.support.error.ErrorType;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+
+import java.time.DayOfWeek;
+import java.time.ZonedDateTime;
+import java.time.temporal.TemporalAdjusters;
+
+/**
+ * 랭킹 집계 배치 서비스
+ * <p>
+ * 주간/월간 랭킹 집계 배치를 실행하는 서비스를 제공합니다.
+ */
+@Slf4j
+@Component
+public class RankingAggregationService {
+
+    private final JobLauncher jobLauncher;
+    private final Job rankingJob;
+
+    public RankingAggregationService(
+            JobLauncher jobLauncher,
+            @Qualifier("rankingJob") Job rankingJob
+    ) {
+        this.jobLauncher = jobLauncher;
+        this.rankingJob = rankingJob;
+    }
+
+    /**
+     * 주간 랭킹 집계 배치 실행
+     *
+     * @param targetDate 집계 대상 날짜 (이 날짜가 속한 주간을 집계)
+     * @return JobExecution 실행 결과
+     * @note Spring Batch는 자체 트랜잭션을 관리하므로 @Transactional을 사용하지 않습니다.
+     */
+    public RankingAggregationInfo executeWeeklyRanking(ZonedDateTime targetDate) {
+        if (targetDate == null) {
+            throw new CoreException(ErrorType.BAD_REQUEST, "집계 대상 날짜는 필수입니다.");
+        }
+
+        log.info("주간 랭킹 집계 배치 실행 시작: targetDate={}", targetDate);
+
+        try {
+            // 주간 시작일/종료일 계산
+            ZonedDateTime weekStart = targetDate.with(TemporalAdjusters.previousOrSame(DayOfWeek.MONDAY))
+                    .toLocalDate()
+                    .atStartOfDay(targetDate.getZone());
+            ZonedDateTime weekEnd = targetDate.with(TemporalAdjusters.nextOrSame(DayOfWeek.SUNDAY))
+                    .toLocalDate()
+                    .atTime(23, 59, 59)
+                    .atZone(targetDate.getZone());
+            ZonedDateTime rankingDate = weekEnd;
+
+            // Job 파라미터 생성
+            JobParameters jobParameters = new JobParametersBuilder()
+                    .addString("periodType", "weekly")
+                    .addString("rankingDate", rankingDate.toString())
+                    .addString("startDate", weekStart.toLocalDate().toString())
+                    .addString("endDate", weekEnd.toLocalDate().toString())
+                    .addLong("timestamp", System.currentTimeMillis())
+                    .toJobParameters();
+
+            // 배치 실행
+            JobExecution jobExecution = jobLauncher.run(rankingJob, jobParameters);
+
+            log.info("주간 랭킹 집계 배치 실행 완료: jobExecutionId={}, status={}, exitStatus={}",
+                    jobExecution.getId(),
+                    jobExecution.getStatus(),
+                    jobExecution.getExitStatus());
+
+            return RankingAggregationInfo.from(jobExecution, "weekly");
+
+        } catch (Exception e) {
+            log.error("주간 랭킹 집계 배치 실행 실패: targetDate={}", targetDate, e);
+            throw new CoreException(ErrorType.INTERNAL_ERROR,
+                    "주간 랭킹 집계 배치 실행에 실패했습니다: " + e.getMessage());
+        }
+    }
+
+    /**
+     * 월간 랭킹 집계 배치 실행
+     *
+     * @param targetDate 집계 대상 날짜 (이 날짜가 속한 월간을 집계)
+     * @return JobExecution 실행 결과
+     * @note Spring Batch는 자체 트랜잭션을 관리하므로 @Transactional을 사용하지 않습니다.
+     */
+    public RankingAggregationInfo executeMonthlyRanking(ZonedDateTime targetDate) {
+        if (targetDate == null) {
+            throw new CoreException(ErrorType.BAD_REQUEST, "집계 대상 날짜는 필수입니다.");
+        }
+
+        log.info("월간 랭킹 집계 배치 실행 시작: targetDate={}", targetDate);
+
+        try {
+            // 월간 시작일/종료일 계산
+            ZonedDateTime monthStart = targetDate.with(TemporalAdjusters.firstDayOfMonth())
+                    .toLocalDate()
+                    .atStartOfDay(targetDate.getZone());
+            ZonedDateTime monthEnd = targetDate.with(TemporalAdjusters.lastDayOfMonth())
+                    .toLocalDate()
+                    .atTime(23, 59, 59)
+                    .atZone(targetDate.getZone());
+            ZonedDateTime rankingDate = monthEnd;
+
+            // Job 파라미터 생성
+            // monthly는 Reader에서 최근 30일로 자동 계산하므로 startDate/endDate 생략
+            JobParameters jobParameters = new JobParametersBuilder()
+                    .addString("periodType", "monthly")
+                    .addString("rankingDate", rankingDate.toString())
+                    // startDate, endDate는 Reader에서 자동 계산 (최근 30일)
+                    .addLong("timestamp", System.currentTimeMillis())
+                    .toJobParameters();
+
+            // 배치 실행
+            JobExecution jobExecution = jobLauncher.run(rankingJob, jobParameters);
+
+            log.info("월간 랭킹 집계 배치 실행 완료: jobExecutionId={}, status={}, exitStatus={}",
+                    jobExecution.getId(),
+                    jobExecution.getStatus(),
+                    jobExecution.getExitStatus());
+
+            return RankingAggregationInfo.from(jobExecution, "monthly");
+
+        } catch (Exception e) {
+            log.error("월간 랭킹 집계 배치 실행 실패: targetDate={}", targetDate, e);
+            throw new CoreException(ErrorType.INTERNAL_ERROR,
+                    "월간 랭킹 집계 배치 실행에 실패했습니다: " + e.getMessage());
+        }
+    }
+
+    /**
+     * 주간 및 월간 랭킹 집계 배치 실행
+     *
+     * @param targetDate 집계 대상 날짜
+     * @return 배치 실행 결과
+     * @note Spring Batch는 자체 트랜잭션을 관리하므로 @Transactional을 사용하지 않습니다.
+     */
+    public RankingAggregationInfo executeWeeklyAndMonthlyRanking(ZonedDateTime targetDate) {
+        if (targetDate == null) {
+            throw new CoreException(ErrorType.BAD_REQUEST, "집계 대상 날짜는 필수입니다.");
+        }
+
+        log.info("주간 및 월간 랭킹 집계 배치 실행 시작: targetDate={}", targetDate);
+
+        try {
+            // 주간 랭킹 집계
+            RankingAggregationInfo weeklyResult = executeWeeklyRanking(targetDate);
+
+            // 월간 랭킹 집계
+            RankingAggregationInfo monthlyResult = executeMonthlyRanking(targetDate);
+
+            log.info("주간 및 월간 랭킹 집계 배치 실행 완료");
+
+            return RankingAggregationInfo.combined(weeklyResult, monthlyResult);
+
+        } catch (Exception e) {
+            log.error("주간 및 월간 랭킹 집계 배치 실행 실패: targetDate={}", targetDate, e);
+            throw new CoreException(ErrorType.INTERNAL_ERROR,
+                    "주간 및 월간 랭킹 집계 배치 실행에 실패했습니다: " + e.getMessage());
+        }
+    }
+}
+