1.7.3

src/main/java/site/holliverse/admin/config/SolapiConfig.java

@@ -1,6 +1,5 @@
 package site.holliverse.admin.config;
 
-import io.micrometer.core.instrument.MeterRegistry;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
@@ -9,21 +8,18 @@
 import org.springframework.http.client.SimpleClientHttpRequestFactory;
 import org.springframework.web.client.RestTemplate;
 import site.holliverse.admin.integration.sms.SolapiSmsClient;
-import site.holliverse.shared.monitoring.http.ObservedRestTemplateInterceptor;
 
 @Profile("admin")
 @Configuration
 @EnableConfigurationProperties(SolapiProperties.class)
 public class SolapiConfig {
 
     @Bean
-    public RestTemplate solapiRestTemplate(SolapiProperties properties, MeterRegistry meterRegistry) {
+    public RestTemplate solapiRestTemplate(SolapiProperties properties) {
         SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
         factory.setConnectTimeout(properties.connectTimeoutMs());
         factory.setReadTimeout(properties.readTimeoutMs());
-        RestTemplate restTemplate = new RestTemplate(factory);
-        restTemplate.getInterceptors().add(new ObservedRestTemplateInterceptor(meterRegistry, "solapi"));
-        return restTemplate;
+        return new RestTemplate(factory);
     }
 
     @Bean

src/main/java/site/holliverse/customer/application/usecase/log/UserLogService.java

@@ -2,54 +2,33 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.github.f4b6a3.tsid.Tsid;
-import io.micrometer.core.instrument.Counter;
-import io.micrometer.core.instrument.DistributionSummary;
-import io.micrometer.core.instrument.MeterRegistry;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Profile;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
-import site.holliverse.customer.error.CustomerErrorCode;
-import site.holliverse.customer.error.CustomerException;
+
+import com.github.f4b6a3.tsid.Tsid;
 import site.holliverse.customer.integration.external.AdminLogFeaturesClient;
 import site.holliverse.customer.web.dto.log.UserLogRequest;
+import site.holliverse.customer.error.CustomerErrorCode;
+import site.holliverse.customer.error.CustomerException;
 import site.holliverse.shared.monitoring.CustomerMetrics;
 
 import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 
 @Slf4j
 @Service
 @Profile("customer")
+@RequiredArgsConstructor
 public class UserLogService {
 
     private final KafkaTemplate<String, String> kafkaTemplate;
     private final ObjectMapper objectMapper;
     private final AdminLogFeaturesClient adminLogFeaturesClient;
     private final CustomerMetrics customerMetrics;
-    private final MeterRegistry meterRegistry;
-    private final DistributionSummary batchSizeSummary;
-    private final Map<String, Counter> requestCounters = new ConcurrentHashMap<>();
-    private final Map<String, Counter> resultCounters = new ConcurrentHashMap<>();
-
-    public UserLogService(KafkaTemplate<String, String> kafkaTemplate,
-                          ObjectMapper objectMapper,
-                          AdminLogFeaturesClient adminLogFeaturesClient,
-                          CustomerMetrics customerMetrics,
-                          MeterRegistry meterRegistry) {
-        this.kafkaTemplate = kafkaTemplate;
-        this.objectMapper = objectMapper;
-        this.adminLogFeaturesClient = adminLogFeaturesClient;
-        this.customerMetrics = customerMetrics;
-        this.meterRegistry = meterRegistry;
-        this.batchSizeSummary = DistributionSummary.builder("holliverse.userlog.batch.size")
-                .description("Batch size of user log submissions")
-                .register(meterRegistry);
-    }
 
     @Value("${app.topic.client-events}")
     private String topic;
@@ -60,22 +39,26 @@ public void publishBatch(Long memberId, List<UserLogRequest> requests) {
             return;
         }
         customerMetrics.recordUserLogBatchSize(requests.size());
-        batchSizeSummary.record(requests.size());
-        requestCounter("batch").increment();
         for (UserLogRequest request : requests) {
             doPublish(memberId, request);
         }
 
+        // 이벤트 전달
         requests.forEach(request -> sendAdminTarget(memberId, request));
     }
 
     @Async("userLogTaskExecutor")
     public void publish(Long memberId, UserLogRequest request) {
-        requestCounter("single").increment();
         doPublish(memberId, request);
+
+        // 이벤트 전달
         sendAdminTarget(memberId, request);
     }
 
+    /**
+     * 단일 로그를 Kafka로 전송. {@link #publish}, {@link #publishBatch}에서만 호출.
+     * self-invocation 시 @Async가 적용되지 않으므로 공통 로직을 private로 분리함.
+     */
     private void doPublish(Long memberId, UserLogRequest request) {
         UserLogEventName eventName = UserLogEventName.from(request.eventName());
 
@@ -84,7 +67,6 @@ private void doPublish(Long memberId, UserLogRequest request) {
             eventId = decodeTsidToLong(request.tsid());
         } catch (IllegalArgumentException e) {
             customerMetrics.recordUserLogPublish(request.eventName(), "invalid_tsid");
-            resultCounter("invalid_tsid").increment();
             throw new CustomerException(CustomerErrorCode.INVALID_USER_LOG_EVENT_ID);
         }
 
@@ -102,7 +84,6 @@ private void doPublish(Long memberId, UserLogRequest request) {
             json = objectMapper.writeValueAsString(payload);
         } catch (JsonProcessingException e) {
             customerMetrics.recordUserLogPublish(eventName.value(), "serialization_error");
-            resultCounter("serialization_error").increment();
             log.warn("[UserLog] 직렬화 실패 memberId={}", memberId, e);
             return;
         }
@@ -111,24 +92,29 @@ private void doPublish(Long memberId, UserLogRequest request) {
                 .whenComplete((result, ex) -> {
                     if (ex != null) {
                         customerMetrics.recordUserLogPublish(eventName.value(), "kafka_error");
-                        resultCounter("kafka_error").increment();
                         log.warn("[UserLog] Kafka 전송 실패 memberId={} eventName={}",
                                 memberId, eventName.value(), ex);
                         return;
                     }
                     customerMetrics.recordUserLogPublish(eventName.value(), "success");
-                    resultCounter("kafka_success").increment();
                 });
     }
 
+    /**
+     * 프론트에서 전달한 TSID 문자열을 tsid-creator 라이브러리로 디코딩한다.
+     */
     private static long decodeTsidToLong(String tsid) {
         if (tsid == null || tsid.isBlank()) {
             throw new IllegalArgumentException("TSID must not be null or blank");
         }
+        // Tsid.from(...) 내부에서 형식·길이·알파벳 검증을 수행한다.
         Tsid parsed = Tsid.from(tsid);
         return parsed.toLong();
     }
 
+    /**
+     * Admin 이벤트 변환.
+     */
     private void sendAdminTarget(Long memberId, UserLogRequest request) {
         UserLogEventName eventName = UserLogEventName.from(request.eventName());
         if (!isAdminTarget(eventName)) {
@@ -142,25 +128,12 @@ private void sendAdminTarget(Long memberId, UserLogRequest request) {
         );
     }
 
+    /**
+     * 대상 이벤트.
+     */
     private boolean isAdminTarget(UserLogEventName eventName) {
         return eventName == UserLogEventName.CLICK_COMPARE
                 || eventName == UserLogEventName.CLICK_PENALTY
                 || eventName == UserLogEventName.CLICK_CHANGE;
     }
-
-    private Counter requestCounter(String mode) {
-        return requestCounters.computeIfAbsent(mode, ignored ->
-                Counter.builder("holliverse.userlog.requests")
-                        .description("User log request count by mode")
-                        .tag("mode", mode)
-                        .register(meterRegistry));
-    }
-
-    private Counter resultCounter(String result) {
-        return resultCounters.computeIfAbsent(result, ignored ->
-                Counter.builder("holliverse.userlog.publish")
-                        .description("User log publish result count")
-                        .tag("result", result)
-                        .register(meterRegistry));
-    }
 }

src/main/java/site/holliverse/customer/application/usecase/recommendation/RecommendationKafkaConsumeUseCase.java

@@ -0,0 +1,96 @@
+package site.holliverse.customer.application.usecase.recommendation;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import site.holliverse.customer.integration.kafka.dto.RecommendationMessagePayload;
+import site.holliverse.customer.persistence.entity.PersonaRecommendation;
+import site.holliverse.customer.persistence.entity.RecommendedProductItem;
+import site.holliverse.customer.persistence.repository.PersonaRecommendationRepository;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * recommendation-topic 메시지의 도메인 처리 담당.
+ * - persona_recommendation upsert
+ * - 대기 중인 Future 완료
+ */
+@Service
+@Profile("customer")
+@RequiredArgsConstructor
+public class RecommendationKafkaConsumeUseCase {
+
+    private final PersonaRecommendationRepository personaRecommendationRepository;
+    private final RecommendationPendingFutureRegistry pendingFutureRegistry;
+
+    @Transactional
+    public RecommendationKafkaConsumeResult execute(RecommendationMessagePayload message) {
+        List<RecommendedProductItem> products = message.recommendedProducts() == null
+                ? Collections.emptyList()
+                : message.recommendedProducts().stream()
+                .map(p -> new RecommendedProductItem(
+                        p.rank(),
+                        p.productId(),
+                        p.productName(),
+                        p.productType(),
+                        p.productPrice(),
+                        p.salePrice(),
+                        p.tags() != null ? p.tags() : Collections.emptyList(),
+                        p.reason()
+                ))
+                .toList();
+
+        String cachedText = message.cachedLlmRecommendation() != null ? message.cachedLlmRecommendation() : "";
+
+        PersonaRecommendation saved = personaRecommendationRepository
+                .findById(message.memberId())
+                .map(entity -> {
+                    entity.updateRecommendation(message.segment(), cachedText, products);
+                    return personaRecommendationRepository.save(entity);
+                })
+                .orElseGet(() -> personaRecommendationRepository.save(
+                        PersonaRecommendation.builder()
+                                .memberId(message.memberId())
+                                .segment(message.segment())
+                                .cachedLlmRecommendation(cachedText)
+                                .recommendedProducts(products)
+                                .build()));
+
+        String outcome = "stored_without_waiter";
+        CompletableFuture<RecommendationResult> future = pendingFutureRegistry.remove(message.memberId());
+        if (future != null) {
+            future.complete(RecommendationResult.fromEntity(saved, RecommendationResult.RecommendationSource.FASTAPI));
+            outcome = "completed_pending";
+        }
+
+        return new RecommendationKafkaConsumeResult(
+                message.memberId(),
+                message.segment().name(),
+                products.size(),
+                saved.getUpdatedAt(),
+                outcome
+        );
+    }
+
+    public void completeExceptionally(Long memberId, Exception e) {
+        if (memberId == null) {
+            return;
+        }
+        CompletableFuture<RecommendationResult> future = pendingFutureRegistry.remove(memberId);
+        if (future != null) {
+            future.completeExceptionally(e);
+        }
+    }
+
+    public record RecommendationKafkaConsumeResult(
+            Long memberId,
+            String segment,
+            int productCount,
+            Instant updatedAt,
+            String outcome
+    ) {}
+}

src/main/java/site/holliverse/customer/application/usecase/recommendation/RecommendationPendingFutureRegistry.java

@@ -1,11 +1,8 @@
 package site.holliverse.customer.application.usecase.recommendation;
 
-import io.micrometer.core.instrument.Gauge;
-import io.micrometer.core.instrument.MeterRegistry;
 import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Component;
 
-import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -19,12 +16,6 @@ public class RecommendationPendingFutureRegistry {
 
     private final ConcurrentHashMap<Long, CompletableFuture<RecommendationResult>> pending = new ConcurrentHashMap<>();
 
-    public RecommendationPendingFutureRegistry(MeterRegistry meterRegistry) {
-        Gauge.builder("holliverse.recommendation.pending.size", pending, Map::size)
-                .description("Current number of pending recommendation futures")
-                .register(meterRegistry);
-    }
-
     /**
      * 해당 회원에 대한 대기 Future를 반환하거나, 없으면 새로 생성해 등록 후 반환.
      */