UPLUS-126 RDB 기반 사용량 배치 시스템 보완

src/main/java/com/project/controller/BatchController.java

@@ -0,0 +1,29 @@
+package com.project.controller;
+
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.project.rdb.batch.orchestrator.UsageOrchestrator;
+
+import lombok.RequiredArgsConstructor;
+
+@RestController
+@RequestMapping("/api/batch")
+@RequiredArgsConstructor
+public class BatchController {
+
+    private final UsageOrchestrator usageOrchestrator;
+
+    @PostMapping("/usage/run")
+    public ResponseEntity<String> runUsagePipeline() {
+        try {
+            usageOrchestrator.run();
+            return ResponseEntity.ok("✅ usage batch pipeline started");
+        } catch (Exception e) {
+            return ResponseEntity.internalServerError()
+                    .body("❌ batch execution failed: " + e.getMessage());
+        }
+    }
+}

src/main/java/com/project/global/exception/code/domain/GlobalErrorCode.java

@@ -24,7 +24,8 @@ public enum GlobalErrorCode implements BaseErrorCode {
             HttpStatus.BAD_REQUEST, "COMMON_011", "UsageNotification Produce 과정에서 에러가 발생하였습니다"),
     USAGE_OUTBOX_WRITER_FAILED(
             HttpStatus.BAD_REQUEST, "COMMON_012", "UsageOutbox Writer 배치 시스템이 실패하였습니다"),
-    PLAN_NOT_VALID(HttpStatus.BAD_REQUEST, "COMMON_013", "존재하지 않는 요금제입니다");
+    PLAN_NOT_VALID(HttpStatus.BAD_REQUEST, "COMMON_013", "존재하지 않는 요금제입니다"),
+    BATCH_NOT_FINISHED(HttpStatus.BAD_REQUEST, "COMMON_014", "이전 배치 작업이 끝나지 않았습니다");
 
     private final HttpStatus httpStatus;
     private final String customCode;

src/main/java/com/project/rdb/batch/model/entity/UsageNotificationOutbox.java

@@ -38,7 +38,7 @@ public class UsageNotificationOutbox {
     private int threshold;
 
     @Column(nullable = false)
-    private int percent;
+    private double percent;
 
     @Column(name = "total_used_mb", nullable = false)
     private Long totalUsedMb;

src/main/java/com/project/rdb/batch/notificationsend/config/NotificationSendJobConfig.java

@@ -12,10 +12,10 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.transaction.PlatformTransactionManager;
 
-import com.project.rdb.batch.model.dto.UsageNotificationEvent;
-import com.project.rdb.batch.model.dto.UsageNotificationOutboxRow;
+import com.project.rdb.batch.notificationsend.dto.NotificationMessage;
 import com.project.rdb.batch.notificationsend.processor.NotificationSendProcessor;
 import com.project.rdb.batch.notificationsend.writer.NotificationSendWriter;
+import com.project.rdb.batch.usagenotification.dto.UsageNotificationOutboxRow;
 
 import lombok.RequiredArgsConstructor;
 
@@ -44,7 +44,7 @@ public Job notificationSendJob() {
     @Bean
     public Step notificationSendStep() {
         return new StepBuilder("notificationSendStep", jobRepository)
-                .<UsageNotificationOutboxRow, UsageNotificationEvent>chunk(CHUNK_SIZE, txManager)
+                .<UsageNotificationOutboxRow, NotificationMessage>chunk(CHUNK_SIZE, txManager)
                 .reader(outboxReader)
                 .processor(notificationSendProcessor)
                 .writer(notificationSendWriter)

src/main/java/com/project/rdb/batch/notificationsend/dto/NotificationMessage.java

@@ -0,0 +1,11 @@
+package com.project.rdb.batch.notificationsend.dto;
+
+import java.util.Map;
+import java.util.UUID;
+
+public record NotificationMessage(
+        UUID eventId,
+        Long id,
+        Long templateGroupId,
+        Map<String, Object> subscriptionInfo,
+        Map<String, Object> variables) {}

src/main/java/com/project/rdb/batch/notificationsend/dto/NotificationSendTask.java

@@ -0,0 +1,8 @@
+package com.project.rdb.batch.notificationsend.dto;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.springframework.kafka.support.SendResult;
+
+public record NotificationSendTask(
+        NotificationMessage event, CompletableFuture<SendResult<String, String>> future) {}

src/main/java/com/project/rdb/batch/notificationsend/processor/NotificationSendProcessor.java

@@ -1,32 +1,40 @@
 package com.project.rdb.batch.notificationsend.processor;
 
+import java.util.Map;
 import java.util.UUID;
 
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.stereotype.Component;
 
-import com.project.rdb.batch.model.dto.UsageNotificationEvent;
-import com.project.rdb.batch.model.dto.UsageNotificationOutboxRow;
+import com.project.rdb.batch.notificationsend.dto.NotificationMessage;
+import com.project.rdb.batch.usagenotification.dto.UsageNotificationOutboxRow;
 
 import lombok.extern.slf4j.Slf4j;
 
 @Component
 @Slf4j
 public class NotificationSendProcessor
-        implements ItemProcessor<UsageNotificationOutboxRow, UsageNotificationEvent> {
+        implements ItemProcessor<UsageNotificationOutboxRow, NotificationMessage> {
 
     @Override
-    public UsageNotificationEvent process(UsageNotificationOutboxRow item) {
+    public NotificationMessage process(UsageNotificationOutboxRow item) {
 
-        return new UsageNotificationEvent(
+        return new NotificationMessage(
                 UUID.randomUUID(),
                 item.id(),
-                item.subId(),
-                item.period(),
-                item.unit(),
-                item.threshold(),
-                item.percent(),
-                item.totalUsedMb(),
-                item.allotmentMb());
+                101L,
+                Map.of(
+                        "subId", item.subId(),
+                        "phoneNumber", item.phoneNumber(),
+                        "email", item.email()),
+                Map.of(
+                        "period", item.period(),
+                        "threshold", item.threshold(),
+                        "percent", item.percent(),
+                        "totalUsedMb", item.totalUsedMb(),
+                        "allotmentMb", item.allotmentMb(),
+                        "phoneNumber", item.phoneNumber(),
+                        "email", item.email(),
+                        "createdAt", item.createdAt()));
     }
 }

src/main/java/com/project/rdb/batch/notificationsend/reader/NotificationSendReaderConfig.java

@@ -7,7 +7,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-import com.project.rdb.batch.model.dto.UsageNotificationOutboxRow;
+import com.project.rdb.batch.usagenotification.dto.UsageNotificationOutboxRow;
 
 @Configuration
 public class NotificationSendReaderConfig {
@@ -19,17 +19,24 @@ public JdbcCursorItemReader<UsageNotificationOutboxRow> notificationSendOutboxRe
         String sql =
                 """
                     SELECT
-                        id,
-                        sub_id,
-                        period,
-                        unit,
-                        threshold,
-                        percent,
-                        total_used_mb,
-                        allotment_mb
-                    FROM usage_notification_outbox
-                    WHERE status = 'PENDING'
-                    ORDER BY id
+                        o.id,
+                        o.sub_id,
+                        o.period,
+                        o.plan_name,
+                        o.threshold,
+                        o.percent,
+                        o.total_used_mb,
+                        o.allotment_mb,
+                        s.phone_number,
+                        c.email_enc,
+                        o.created_at
+                    FROM usage_notification_outbox o
+                    JOIN subscription s
+                      ON s.sub_id = o.sub_id
+                    JOIN customer c
+                      ON c.customer_id = s.customer_id
+                    WHERE o.status = 'PENDING'
+                    ORDER BY o.id
                     """;
 
         return new JdbcCursorItemReaderBuilder<UsageNotificationOutboxRow>()
@@ -43,11 +50,14 @@ public JdbcCursorItemReader<UsageNotificationOutboxRow> notificationSendOutboxRe
                                         rs.getLong("id"),
                                         rs.getLong("sub_id"),
                                         rs.getString("period"),
-                                        rs.getString("unit"),
+                                        rs.getString("plan_name"),
                                         rs.getInt("threshold"),
-                                        rs.getInt("percent"),
+                                        rs.getDouble("percent"),
                                         rs.getLong("total_used_mb"),
-                                        rs.getLong("allotment_mb")))
+                                        rs.getLong("allotment_mb"),
+                                        rs.getString("phone_number"),
+                                        rs.getString("email_enc"),
+                                        rs.getTimestamp("created_at").toLocalDateTime()))
                 .build();
     }
 }

src/main/java/com/project/rdb/batch/notificationsend/writer/NotificationSendWriter.java

@@ -13,26 +13,26 @@
 import org.springframework.stereotype.Component;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.project.rdb.batch.model.dto.NotificationSendTask;
-import com.project.rdb.batch.model.dto.UsageNotificationEvent;
 import com.project.rdb.batch.model.repository.UsageNotificationOutboxRepository;
+import com.project.rdb.batch.notificationsend.dto.NotificationMessage;
+import com.project.rdb.batch.notificationsend.dto.NotificationSendTask;
 
 import lombok.RequiredArgsConstructor;
 
 @Component
 @RequiredArgsConstructor
-public class NotificationSendWriter implements ItemWriter<UsageNotificationEvent> {
+public class NotificationSendWriter implements ItemWriter<NotificationMessage> {
 
     private final KafkaTemplate<String, String> kafkaTemplate;
     private final UsageNotificationOutboxRepository repository;
     private final ObjectMapper objectMapper;
     private static final int MAX_FAILURE_REASON_LENGTH = 255;
 
     @Override
-    public void write(Chunk<? extends UsageNotificationEvent> items) {
+    public void write(Chunk<? extends NotificationMessage> items) {
 
         // usageNotificationOutboxId 추출
-        List<Long> ids = items.getItems().stream().map(UsageNotificationEvent::id).toList();
+        List<Long> ids = items.getItems().stream().map(NotificationMessage::id).toList();
 
         // PROCESSING Status 먼저 DB 반영
         repository.markProcessing(ids);
@@ -42,12 +42,13 @@ public void write(Chunk<? extends UsageNotificationEvent> items) {
 
         Map<Long, String> failedReasons = new HashMap<>();
 
-        for (UsageNotificationEvent event : items) {
+        for (NotificationMessage event : items) {
             try {
                 String payload = objectMapper.writeValueAsString(event);
+                String key = String.valueOf(event.subscriptionInfo().get("subId"));
 
                 CompletableFuture<SendResult<String, String>> future =
-                        kafkaTemplate.send("notification-usage", event.subId().toString(), payload);
+                        kafkaTemplate.send("noti-tp", key, payload);
 
                 tasks.add(new NotificationSendTask(event, future));
             } catch (Exception e) {

src/main/java/com/project/rdb/batch/orchestrator/BatchTimeWindowService.java

@@ -0,0 +1,41 @@
+package com.project.rdb.batch.orchestrator;
+
+import java.time.LocalDateTime;
+import java.util.Map;
+
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import lombok.RequiredArgsConstructor;
+
+@Service
+@RequiredArgsConstructor
+public class BatchTimeWindowService {
+
+    private final NamedParameterJdbcTemplate jdbc;
+
+    public LocalDateTime resolve(String jobName) {
+        return jdbc.queryForObject(
+                """
+                        SELECT last_processed_at
+                        FROM batch_watermark
+                        WHERE job_name = :jobName
+                        """,
+                Map.of("jobName", jobName),
+                LocalDateTime.class);
+    }
+
+    @Transactional
+    public void update(String jobName, LocalDateTime newFrom) {
+        jdbc.update(
+                """
+                UPDATE batch_watermark
+                SET last_processed_at = :newFrom
+                WHERE job_name = :jobName
+                """,
+                Map.of(
+                        "jobName", jobName,
+                        "newFrom", newFrom));
+    }
+}