[volume-10] Collect, Stack, Zip

apps/commerce-api/src/main/java/com/loopers/application/api/product/ProductV1Api.java

@@ -66,10 +66,11 @@ public ApiResponse<GetProductDetailResponse> getProductDetail(
     @GetMapping("/rankings")
     public ApiResponse<GetProductRankingsResponse> getProductRankings(
             @RequestParam @DateTimeFormat(pattern = "yyyy-MM-dd") LocalDate date,
+            @RequestParam(required = false, defaultValue = "DAILY") String type,
             @RequestParam(required = false, defaultValue = "0") int pageNo,
             @RequestParam(required = false, defaultValue = "10") int pageSize
     ) {
-        ProductRankingList ranking = getProductRankingService.getRanking(new GetProductRankingQuery(date, pageNo, pageSize));
+        ProductRankingList ranking = getProductRankingService.getRanking(new GetProductRankingQuery(date, type, pageNo, pageSize));
 
         return ApiResponse.success(GetProductRankingsResponse.from(ranking));
     }

apps/commerce-api/src/main/java/com/loopers/application/api/product/ProductV1ApiSpec.java

@@ -41,5 +41,5 @@ ApiResponse<GetProductListResponse> getProductList(
             summary = "상품 랭킹 조회",
             description = "상품 랭킹을 조회합니다."
     )
-    ApiResponse<GetProductRankingsResponse> getProductRankings(LocalDate date, int pageNo, int pageSize);
+    ApiResponse<GetProductRankingsResponse> getProductRankings(LocalDate date, String type, int pageNo, int pageSize);
 }

apps/commerce-batch/build.gradle.kts

@@ -0,0 +1,26 @@
+dependencies {
+    implementation(project(":core:infra:database:mysql:mysql-config"))
+    implementation(project(":supports:jackson"))
+    implementation(project(":supports:logging"))
+    implementation(project(":supports:monitoring"))
+
+    //service
+    implementation(project(":core:service"))
+
+    //domain
+    implementation(project(":core:domain"))
+
+    // web
+    implementation("org.springframework.boot:spring-boot-starter-web")
+
+    //batch
+    implementation("org.springframework.boot:spring-boot-starter-batch")
+
+    implementation("org.springframework:spring-tx")
+    implementation("org.springframework.boot:spring-boot-starter-actuator")
+
+    // test-fixtures
+    testImplementation(project(":core:infra:database:mysql:mysql-config"))
+    testImplementation(testFixtures(project(":core:domain")))
+    testImplementation(testFixtures(project(":core:infra:database:mysql")))
+}

apps/commerce-batch/src/main/java/com/loopers/CommerceBatchApplication.java

@@ -0,0 +1,14 @@
+package com.loopers;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+@EnableScheduling
+@SpringBootApplication
+public class CommerceBatchApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(CommerceBatchApplication.class, args);
+    }
+}

apps/commerce-batch/src/main/java/com/loopers/application/batch/product/MonthlyProductMetricBatchPartitioner.java

@@ -0,0 +1,55 @@
+package com.loopers.application.batch.product;
+
+import com.loopers.core.domain.product.repository.DailyProductMetricRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.core.configuration.annotation.StepScope;
+import org.springframework.batch.core.partition.support.Partitioner;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDate;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@Component
+@StepScope
+@RequiredArgsConstructor
+public class MonthlyProductMetricBatchPartitioner implements Partitioner {
+
+    private final DailyProductMetricRepository dailyProductMetricRepository;
+
+    @Value("#{jobParameters['startDate']}")
+    private String startDateParam;
+
+    @Value("#{jobParameters['endDate']}")
+    private String endDateParam;
+
+    @Override
+    public Map<String, ExecutionContext> partition(int gridSize) {
+        LocalDate startDate = LocalDate.parse(startDateParam);
+        LocalDate endDate = LocalDate.parse(endDateParam);
+        Long totalCount = dailyProductMetricRepository.countDistinctProductIdsBy(startDate, endDate);
+
+        if (totalCount == 0) {
+            return Collections.emptyMap();
+        }
+
+        long targetSize = (totalCount / gridSize) + 1;
+        Map<String, ExecutionContext> partitions = new HashMap<>();
+
+        for (int i = 0; i < gridSize; i++) {
+            ExecutionContext context = new ExecutionContext();
+
+            context.putLong("partitionOffset", i * targetSize);
+            context.putLong("partitionLimit", targetSize);
+            context.putString("startDate", startDateParam);
+            context.putString("endDate", endDateParam);
+
+            partitions.put("partition" + i, context);
+        }
+
+        return partitions;
+    }
+}

apps/commerce-batch/src/main/java/com/loopers/application/batch/product/MonthlyProductMetricBatchReader.java

@@ -0,0 +1,39 @@
+package com.loopers.application.batch.product;
+
+import com.loopers.core.domain.product.repository.DailyProductMetricRepository;
+import com.loopers.core.domain.product.vo.ProductMetricAggregation;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStreamReader;
+
+import java.time.LocalDate;
+import java.util.Iterator;
+import java.util.Objects;
+
+@RequiredArgsConstructor
+public class MonthlyProductMetricBatchReader implements ItemStreamReader<ProductMetricAggregation> {
+
+    private final DailyProductMetricRepository dailyProductMetricRepository;
+    private Iterator<ProductMetricAggregation> iterator;
+
+    @Override
+    public void open(@NonNull ExecutionContext executionContext) {
+        LocalDate startDate = LocalDate.parse(executionContext.getString("startDate"));
+        LocalDate endDate = LocalDate.parse(executionContext.getString("endDate"));
+        long partitionOffset = executionContext.getLong("partitionOffset");
+        long partitionLimit = executionContext.getLong("partitionLimit");
+
+        this.iterator = dailyProductMetricRepository.findAggregatedBy(startDate, endDate, partitionOffset, partitionLimit)
+                .iterator();
+    }
+
+    @Override
+    public ProductMetricAggregation read() {
+        if (Objects.isNull(iterator) || !iterator.hasNext()) {
+            return null;
+        }
+
+        return iterator.next();
+    }
+}

apps/commerce-batch/src/main/java/com/loopers/application/batch/product/MonthlyProductMetricBatchWriter.java

@@ -0,0 +1,40 @@
+package com.loopers.application.batch.product;
+
+import com.loopers.core.domain.product.MonthlyProductMetric;
+import com.loopers.core.domain.product.repository.MonthlyProductMetricRepository;
+import com.loopers.core.domain.product.vo.ProductMetricAggregation;
+import com.loopers.core.domain.product.vo.YearMonth;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.core.configuration.annotation.StepScope;
+import org.springframework.batch.item.Chunk;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStreamWriter;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDate;
+import java.util.List;
+
+@StepScope
+@Component
+@RequiredArgsConstructor
+public class MonthlyProductMetricBatchWriter implements ItemStreamWriter<ProductMetricAggregation> {
+
+    private final MonthlyProductMetricRepository repository;
+    private YearMonth yearMonth;
+
+    @Override
+    public void open(@NonNull ExecutionContext executionContext) {
+        LocalDate startDate = LocalDate.parse(executionContext.getString("startDate"));
+        this.yearMonth = YearMonth.from(startDate);
+    }
+
+    @Override
+    public void write(@NonNull Chunk<? extends ProductMetricAggregation> chunk) {
+        List<MonthlyProductMetric> monthlyMetrics = chunk.getItems().stream()
+                .map(aggregation -> aggregation.to(yearMonth))
+                .toList();
+
+        repository.bulkUpsert(monthlyMetrics);
+    }
+}

apps/commerce-batch/src/main/java/com/loopers/application/batch/product/MonthlyProductMetricScheduler.java

@@ -0,0 +1,39 @@
+package com.loopers.application.batch.product;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.JobParametersInvalidException;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
+import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
+import org.springframework.batch.core.repository.JobRestartException;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDate;
+import java.time.YearMonth;
+
+@Component
+@RequiredArgsConstructor
+public class MonthlyProductMetricScheduler {
+
+    private final JobLauncher jobLauncher;
+    private final Job monthlyProductMetricJob;
+
+    @Scheduled(cron = "0 0 2 1 * ?")
+    public void run() throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, JobParametersInvalidException, JobRestartException {
+        YearMonth lastMonth = YearMonth.now().minusMonths(1);
+        LocalDate startDate = lastMonth.atDay(1);
+        LocalDate endDate = lastMonth.atEndOfMonth();
+
+        JobParameters jobParameters = new JobParametersBuilder()
+                .addString("startDate", startDate.toString())
+                .addString("endDate", endDate.toString())
+                .addLong("timestamp", System.currentTimeMillis())
+                .toJobParameters();
+
+        jobLauncher.run(monthlyProductMetricJob, jobParameters);
+    }
+}

apps/commerce-batch/src/main/java/com/loopers/application/batch/product/WeeklyProductMetricBatchPartitioner.java

@@ -0,0 +1,55 @@
+package com.loopers.application.batch.product;
+
+import com.loopers.core.domain.product.repository.DailyProductMetricRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.core.configuration.annotation.StepScope;
+import org.springframework.batch.core.partition.support.Partitioner;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDate;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@Component
+@StepScope
+@RequiredArgsConstructor
+public class WeeklyProductMetricBatchPartitioner implements Partitioner {
+
+    private final DailyProductMetricRepository dailyProductMetricRepository;
+
+    @Value("#{jobParameters['startDate']}")
+    private String startDateParam;
+
+    @Value("#{jobParameters['endDate']}")
+    private String endDateParam;
+
+    @Override
+    public Map<String, ExecutionContext> partition(int gridSize) {
+        LocalDate startDate = LocalDate.parse(startDateParam);
+        LocalDate endDate = LocalDate.parse(endDateParam);
+        Long totalCount = dailyProductMetricRepository.countDistinctProductIdsBy(startDate, endDate);
+
+        if (totalCount == 0) {
+            return Collections.emptyMap();
+        }
+
+        long targetSize = (totalCount / gridSize) + 1;
+        Map<String, ExecutionContext> partitions = new HashMap<>();
+
+        for (int i = 0; i < gridSize; i++) {
+            ExecutionContext context = new ExecutionContext();
+
+            context.putLong("partitionOffset", i * targetSize);
+            context.putLong("partitionLimit", targetSize);
+            context.putString("startDate", startDateParam);
+            context.putString("endDate", endDateParam);
+
+            partitions.put("partition" + i, context);
+        }
+
+        return partitions;
+    }
+}

apps/commerce-batch/src/main/java/com/loopers/application/batch/product/WeeklyProductMetricBatchReader.java

@@ -0,0 +1,41 @@
+package com.loopers.application.batch.product;
+
+import com.loopers.core.domain.product.repository.DailyProductMetricRepository;
+import com.loopers.core.domain.product.vo.ProductMetricAggregation;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStreamReader;
+import org.springframework.lang.Nullable;
+
+import java.time.LocalDate;
+import java.util.Iterator;
+import java.util.Objects;
+
+@RequiredArgsConstructor
+public class WeeklyProductMetricBatchReader implements ItemStreamReader<ProductMetricAggregation> {
+
+    private final DailyProductMetricRepository dailyProductMetricRepository;
+    private Iterator<ProductMetricAggregation> iterator;
+
+    @Override
+    public void open(@NonNull ExecutionContext executionContext) {
+        LocalDate startDate = LocalDate.parse(executionContext.getString("startDate"));
+        LocalDate endDate = LocalDate.parse(executionContext.getString("endDate"));
+        long partitionOffset = executionContext.getLong("partitionOffset");
+        long partitionLimit = executionContext.getLong("partitionLimit");
+
+        this.iterator = dailyProductMetricRepository.findAggregatedBy(startDate, endDate, partitionOffset, partitionLimit)
+                .iterator();
+    }
+
+    @Nullable
+    @Override
+    public ProductMetricAggregation read() {
+        if (Objects.isNull(iterator) || !iterator.hasNext()) {
+            return null;
+        }
+
+        return iterator.next();
+    }
+}

apps/commerce-batch/src/main/java/com/loopers/application/batch/product/WeeklyProductMetricBatchWriter.java

@@ -0,0 +1,40 @@
+package com.loopers.application.batch.product;
+
+import com.loopers.core.domain.common.vo.YearMonthWeek;
+import com.loopers.core.domain.product.WeeklyProductMetric;
+import com.loopers.core.domain.product.repository.WeeklyProductMetricRepository;
+import com.loopers.core.domain.product.vo.ProductMetricAggregation;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.core.configuration.annotation.StepScope;
+import org.springframework.batch.item.Chunk;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStreamWriter;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDate;
+import java.util.List;
+
+@StepScope
+@Component
+@RequiredArgsConstructor
+public class WeeklyProductMetricBatchWriter implements ItemStreamWriter<ProductMetricAggregation> {
+
+    private final WeeklyProductMetricRepository repository;
+    private YearMonthWeek yearMonthWeek;
+
+    @Override
+    public void open(@NonNull ExecutionContext executionContext) {
+        LocalDate startDate = LocalDate.parse(executionContext.getString("startDate"));
+        this.yearMonthWeek = YearMonthWeek.from(startDate);
+    }
+
+    @Override
+    public void write(@NonNull Chunk<? extends ProductMetricAggregation> chunk) {
+        List<WeeklyProductMetric> weeklyMetrics = chunk.getItems().stream()
+                .map(aggregation -> aggregation.to(yearMonthWeek))
+                .toList();
+
+        repository.bulkUpsert(weeklyMetrics);
+    }
+}