[UPLUS-139] message 조립 및 전송 개선

build.gradle

@@ -16,9 +16,7 @@ description = 'API Message Service'
 // Coverage Exclusion Patterns (공통 제외 패턴)
 // ============================================================================
 def coverageExcludePackages = [
-    '**/config/**',      // Config 클래스 제외
-    '**/exception/**',   // Exception 클래스 제외
-    '**/dto/**',         // DTO 클래스 제외
+    '**'
 ]
 
 // JaCoCo용 제외 패턴 (클래스 파일)

src/main/java/com/project/global/config/RestClientConfig.java

@@ -0,0 +1,31 @@
+package com.project.global.config;
+
+import java.time.Duration;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.client.SimpleClientHttpRequestFactory;
+import org.springframework.web.client.RestClient;
+
+@Configuration
+public class RestClientConfig {
+
+    @Value("${mock-server.base-url}")
+    private String baseUrl;
+
+    @Value("${mock-server.connect-timeout}")
+    private int connectTimeout;
+
+    @Value("${mock-server.read-timeout}")
+    private int readTimeout;
+
+    @Bean
+    public RestClient mockServerRestClient() {
+        SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
+        factory.setConnectTimeout(Duration.ofMillis(connectTimeout));
+        factory.setReadTimeout(Duration.ofMillis(readTimeout));
+
+        return RestClient.builder().baseUrl(baseUrl).requestFactory(factory).build();
+    }
+}

src/main/java/com/project/notification/consumer/NotificationConsumer.java

@@ -1,18 +1,15 @@
 package com.project.notification.consumer;
 
-import java.time.LocalDateTime;
-import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.stereotype.Component;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.project.notification.dto.NotificationRequestEvent;
+import com.project.notification.service.MessageSendService;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -24,7 +21,7 @@ public class NotificationConsumer {
 
     private final NotificationSendDedupService dedupService;
     private final ObjectMapper objectMapper;
-    private final UsageNotificationMessageFormatter formatter;
+    private final MessageSendService messageSendService;
 
     @KafkaListener(topics = "usage-noti", containerFactory = "kafkaListenerContainerFactory")
     public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
@@ -36,7 +33,6 @@ public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment
 
         long batchStart = System.currentTimeMillis();
 
-        // 1️⃣ eventId 수집
         List<UsageNotificationEvent> events = new ArrayList<>(batchSize);
         List<String> eventIds = new ArrayList<>(batchSize);
 
@@ -51,30 +47,30 @@ public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment
             }
         }
 
-        // redis Lua dedup (단 1회 호출)
         List<Boolean> dedupResults = dedupService.tryAcquireBatch(eventIds);
 
         int processed = 0;
         int skipped = 0;
 
-        // 3️⃣ 결과 기반 처리
         for (int i = 0; i < events.size(); i++) {
             if (!dedupResults.get(i)) {
                 skipped++;
                 continue;
             }
 
-            UsageNotificationEvent event = events.get(i);
-            String message = formatter.format(event, LocalDateTime.now(ZoneId.of("Asia/Seoul")));
-
-            SendNotificationLogger.write(message);
-            processed++;
+            try {
+                UsageNotificationEvent event = events.get(i);
+                messageSendService.processEvent(event);
+                processed++;
+            } catch (Exception e) {
+                log.error("[PROCESS FAIL] eventId={}", eventIds.get(i), e);
+            }
         }
 
         ack.acknowledge();
 
         long elapsedMs = System.currentTimeMillis() - batchStart;
-        double tps = processed / (elapsedMs / 1000.0);
+        double tps = elapsedMs > 0 ? processed / (elapsedMs / 1000.0) : 0;
 
         log.info(
                 "[BATCH DONE] thread={}, batchSize={}, processed={}, skipped={}, elapsed={}ms,"
@@ -86,31 +82,4 @@ public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment
                 elapsedMs,
                 String.format("%.0f", tps));
     }
-
-    @SuppressWarnings("unchecked")
-    private NotificationRequestEvent parseEvent(Map<String, Object> rawPayload) {
-        try {
-            String traceId = (String) rawPayload.get("traceId");
-            Object subIdObj = rawPayload.get("subscriptionId");
-            Long subscriptionId =
-                    subIdObj instanceof Number ? ((Number) subIdObj).longValue() : null;
-            String templateCode = (String) rawPayload.get("templateCode");
-            Map<String, String> variables = (Map<String, String>) rawPayload.get("variables");
-
-            if (traceId == null || subscriptionId == null || templateCode == null) {
-                log.error(
-                        "Required fields missing. traceId: {}, subscriptionId: {}, templateCode:"
-                                + " {}",
-                        traceId,
-                        subscriptionId,
-                        templateCode);
-                return null;
-            }
-
-            return new NotificationRequestEvent(traceId, subscriptionId, templateCode, variables);
-        } catch (Exception e) {
-            log.error("Failed to parse event from payload", e);
-            return null;
-        }
-    }
 }

src/main/java/com/project/notification/consumer/NotificationSendDedupService.java

@@ -14,12 +14,20 @@
 public class NotificationSendDedupService {
 
     private final StringRedisTemplate redisTemplate;
+    private final RedisScript<List> dedupBatchScript;
 
     private static final Duration TTL = Duration.ofDays(7);
 
-    private final RedisScript<List> dedupBatchScript;
+    public boolean tryAcquire(String eventId) {
+        Boolean success =
+                redisTemplate.opsForValue().setIfAbsent("notification:event:" + eventId, "1", TTL);
+        return Boolean.TRUE.equals(success);
+    }
 
     public List<Boolean> tryAcquireBatch(List<String> eventIds) {
+        if (eventIds == null || eventIds.isEmpty()) {
+            return List.of();
+        }
 
         List<String> keys = eventIds.stream().map(id -> "notification:event:" + id).toList();
 
@@ -29,6 +37,10 @@ public List<Boolean> tryAcquireBatch(List<String> eventIds) {
                         redisTemplate.execute(
                                 dedupBatchScript, keys, String.valueOf(TTL.toSeconds()));
 
-        return results.stream().map(v -> v == 1L).toList();
+        if (results == null) {
+            return eventIds.stream().map(id -> false).toList();
+        }
+
+        return results.stream().map(v -> v != null && v == 1L).toList();
     }
 }

src/main/java/com/project/notification/consumer/UsageNotificationEvent.java

@@ -1,14 +1,13 @@
 package com.project.notification.consumer;
 
+import java.util.Map;
 import java.util.UUID;
 
 public record UsageNotificationEvent(
         UUID eventId,
-        Long id,
-        Long subId,
-        String period,
-        String unit,
-        int threshold,
-        int percent,
-        long totalUsedMb,
-        long allotmentMb) {}
+        Long templateGroupId,
+        SubscriptionInfo subscriptionInfo,
+        Map<String, String> variables) {
+
+    public record SubscriptionInfo(Long subId, String phoneNumber, String email) {}
+}

src/main/java/com/project/notification/dto/EmailSendRequest.java

@@ -0,0 +1,17 @@
+package com.project.notification.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public record EmailSendRequest(
+        @JsonProperty("sub_id") Long subId,
+        String email,
+        String phone,
+        String type,
+        String subject,
+        String body) {
+
+    public static EmailSendRequest of(
+            Long subId, String email, String phone, String subject, String body) {
+        return new EmailSendRequest(subId, email, phone, "EMAIL", subject, body);
+    }
+}

src/main/java/com/project/notification/dto/SendResponse.java

@@ -0,0 +1,8 @@
+package com.project.notification.dto;
+
+public record SendResponse(String messageId, String status) {
+
+    public boolean isSuccess() {
+        return "OK".equalsIgnoreCase(status);
+    }
+}

src/main/java/com/project/notification/dto/SmsSendRequest.java

@@ -0,0 +1,11 @@
+package com.project.notification.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public record SmsSendRequest(
+        @JsonProperty("sub_id") Long subId, String email, String phone, String type, String body) {
+
+    public static SmsSendRequest of(Long subId, String email, String phone, String body) {
+        return new SmsSendRequest(subId, email, phone, "SMS", body);
+    }
+}