Feat: 포인트 결제 관련 core payment 로직 추가

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/CorePaymentServiceApplication.java

@@ -5,6 +5,7 @@
 import org.springframework.kafka.annotation.EnableKafka;
 
 @EnableKafka
+@EnableJpaAuditing
 @SpringBootApplication
 public class CorePaymentServiceApplication {
 

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/config/KafkaConsumerConfig.java

@@ -4,13 +4,13 @@
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
-import org.springframework.beans.factory.annotation.Value;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -19,6 +19,7 @@
 @Configuration
 public class KafkaConsumerConfig {
 
+    // application.yml에서 값을 가져오되, 없으면 기본값 사용
     @Value("${spring.kafka.bootstrap-servers:127.0.0.1:9092}")
     private String bootstrapServers;
 

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/controller/PaymentController.java

@@ -2,6 +2,7 @@
 
 import com.fisa.core_payment_service.dto.AmountRequestDto;
 import com.fisa.core_payment_service.service.PaymentService;
+import jakarta.validation.Valid;
 import lombok.RequiredArgsConstructor;
 import org.springframework.web.bind.annotation.*;
 
@@ -29,7 +30,7 @@ public BigDecimal getBalance(@PathVariable String accountNo) {
 
     // 입금
     @PostMapping("/accounts/{accountNo}/deposit")
-    public BigDecimal deposit(@PathVariable String accountNo, @RequestBody AmountRequestDto request) {
+    public BigDecimal deposit(@PathVariable String accountNo, @Valid @RequestBody AmountRequestDto request) {
         return paymentService.deposit(accountNo, request.userUuid(), request.amount());
     }
 

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/dto/AmountRequestDto.java

@@ -1,6 +1,18 @@
 package com.fisa.core_payment_service.dto;
 
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Positive;
+
 import java.math.BigDecimal;
 
-public record AmountRequestDto(String userUuid, BigDecimal amount) {
+public record AmountRequestDto(
+
+        @NotBlank(message = "사용자 UUID는 필수입니다.")
+        String userUuid,
+
+        @NotNull(message = "금액은 필수입니다.")
+        @Positive(message = "금액은 0보다 커야 합니다.")
+        BigDecimal amount
+) {
 }

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/dto/CashRequestDTO.java

@@ -0,0 +1,16 @@
+package com.fisa.core_payment_service.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+@Getter
+@NoArgsConstructor
+@AllArgsConstructor
+@ToString
+public class CashRequestDTO {
+    private String orderId; // 트랜잭션 ID (Saga 추적용)
+    private String loginId; // 사용자 식별자 (userUuid)
+    private Long amount;    // 출금액
+}

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/dto/CashResponseDTO.java

@@ -0,0 +1,17 @@
+package com.fisa.core_payment_service.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+@Getter
+@NoArgsConstructor
+@AllArgsConstructor
+@ToString
+public class CashResponseDTO {
+    private String orderId;
+    private String userUuid;
+    private String status;   // "SUCCESS" or "FAIL"
+    private String message;
+}

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/repository/AccountRepository.java

@@ -4,4 +4,7 @@
 import org.springframework.data.jpa.repository.JpaRepository;
 
 public interface AccountRepository extends JpaRepository<Account, String> {
+
+    // uuid로 계좌 조회
+    Optional<Account> findByUserUuid(String userUuid);
 }

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/service/KafkaConsumerService.java

@@ -2,21 +2,28 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fisa.core_payment_service.domain.Account;
+import com.fisa.core_payment_service.dto.CashRequestDTO;
+import com.fisa.core_payment_service.dto.CashResponseDTO;
 import com.fisa.core_payment_service.dto.CouponIssueMessage;
 import com.fisa.core_payment_service.dto.DepositMessage;
+import com.fisa.core_payment_service.repository.AccountRepository;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Service;
 
+import java.math.BigDecimal;
+
 @Slf4j
 @Service
 @RequiredArgsConstructor
 public class KafkaConsumerService {
 
     private final PaymentService paymentService;
-    private final CouponManageService couponService;
     private final ObjectMapper objectMapper;
+    private final AccountRepository accountRepository;
+    private final KafkaProducerService kafkaProducerService;
 
     // 1. 입금 처리
     @KafkaListener(topics = "bank_deposit", groupId = "core-group")
@@ -25,7 +32,7 @@ public void consumeDeposit(String message) {
             DepositMessage depositDto = objectMapper.readValue(message, DepositMessage.class);
 
             paymentService.deposit(
-                    depositDto.getAccountNo(),
+                    depositDto.getAccountNo(), // ★ 수정됨 (getAccountNumber() -> getAccountNo())
                     depositDto.getUserUuid(),
                     depositDto.getAmount()
             );
@@ -39,14 +46,58 @@ public void consumeDeposit(String message) {
         }
     }
 
-    // 2. 쿠폰 처리
-    @KafkaListener(topics = "coupon_issue", groupId = "core-group")
-    public void consumeCouponIssue(String message) {
+
+    // 2. 현금 출금 처리 (PointService와 연결)
+    @KafkaListener(topics = "core-withdraw-request", groupId = "core-group")
+    public void consumeWithdraw(String message) {
+
+        CashRequestDTO requestDto = null;
+
         try {
-            CouponIssueMessage couponDto = objectMapper.readValue(message, CouponIssueMessage.class);
-            couponService.issueCoupon(couponDto);
+            // (1) 메시지 파싱
+            requestDto = objectMapper.readValue(message, CashRequestDTO.class);
+            log.info("📉 [Core] 출금 요청 수신: {}", requestDto);
+
+            // (2) 계좌 조회
+            Account account = accountRepository.findByUserUuid(requestDto.getLoginId())
+                    .orElseThrow(() -> new IllegalArgumentException("계좌를 찾을 수 없습니다."));
+
+            // (3) 출금 비즈니스 로직
+            paymentService.withdraw(
+                    account.getAccountNo(),
+                    requestDto.getLoginId(),
+                    BigDecimal.valueOf(requestDto.getAmount())
+            );
+
+            // (4) 성공 이벤트 발행 -> PointService의 토픽 이름인 "core-result"로 변경
+            CashResponseDTO successResponse = new CashResponseDTO(
+                    requestDto.getOrderId(),
+                    requestDto.getLoginId(), // ★ userUuid 추가 (PointService 환불용)
+                    "SUCCESS",
+                    "정상 출금 완료"
+            );
+
+            // ★ 토픽 이름 변경: core-withdraw-result -> core-result
+            kafkaProducerService.send("core-result", successResponse);
+            log.info("✅ [Core] 출금 성공 -> Point Service로 전송: {}", successResponse);
+
         } catch (JsonProcessingException e) {
-            log.error("❌ JSON 파싱 에러: {}", message, e);
+            log.error("❌ JSON 파싱 에러 (Withdraw): {}", message, e);
+        } catch (Exception e) {
+            log.error("❌ 출금 처리 실패: {}", e.getMessage());
+
+            if (requestDto != null) {
+                // (5) 실패 이벤트 발행
+                CashResponseDTO failResponse = new CashResponseDTO(
+                        requestDto.getOrderId(),
+                        requestDto.getLoginId(), // ★ userUuid 추가
+                        "FAIL",
+                        e.getMessage()
+                );
+                // ★ 토픽 이름 변경
+                kafkaProducerService.send("core-result", failResponse);
+                log.info("⚠️ [Core] 출금 실패 -> Point Service로 전송: {}", failResponse);
+            }
         }
     }
 }

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/service/KafkaProducerService.java

@@ -0,0 +1,28 @@
+package com.fisa.core_payment_service.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class KafkaProducerService {
+
+    private final KafkaTemplate<String, String> kafkaTemplate;
+    private final ObjectMapper objectMapper;
+
+    public void send(String topic, Object payload) {
+        try {
+            String jsonMessage = objectMapper.writeValueAsString(payload);
+            kafkaTemplate.send(topic, jsonMessage);
+            log.info("🚀 [Core-Producer] Sent to {}: {}", topic, jsonMessage);
+        } catch (JsonProcessingException e) {
+            log.error("❌ [Core-Producer] Serialization failed: {}", payload, e);
+            throw new RuntimeException("Kafka message serialization failed", e);
+        }
+    }
+}

on-premise/core-payment-service/src/main/java/com/fisa/core_payment_service/service/PaymentService.java

@@ -1,27 +1,48 @@
 package com.fisa.core_payment_service.service;
 
 import com.fisa.core_payment_service.domain.Account;
+import com.fisa.core_payment_service.dto.CouponIssueMessage;
 import com.fisa.core_payment_service.repository.AccountRepository;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.math.BigDecimal;
 
+@Slf4j
 @Service
 @RequiredArgsConstructor
 @Transactional(readOnly = true)
 public class PaymentService {
 
     private final AccountRepository accountRepository;
+    private final KafkaProducerService kafkaProducerService;
 
     // 계좌 개설
     @Transactional
     public void createAccount(String accountNo, String userUuid) {
         if (accountRepository.existsById(accountNo)) {
             throw new IllegalArgumentException("이미 존재하는 계좌입니다.");
         }
+
+        // 1. 계좌 저장
         accountRepository.save(Account.create(accountNo, userUuid));
+        log.info("✅ 계좌 개설 완료: accountNo={}, user={}", accountNo, userUuid);
+
+        // 2. 쿠폰 발급 요청 보내기 (Producer)
+        try {
+            CouponIssueMessage couponEvent = new CouponIssueMessage(
+                    userUuid,
+                    "WELCOME_COUPON",
+                    "계좌 개설 축하 쿠폰"
+            );
+            // coupon-service가 듣고 있는 "coupon_issue" 토픽으로 쏜다
+            kafkaProducerService.send("coupon_issue", couponEvent);
+
+        } catch (Exception e) {
+            log.error("⚠️ 계좌는 생성되었으나 쿠폰 발급 요청 실패: {}", e.getMessage());
+        }
     }
 
     // 입금
@@ -31,8 +52,8 @@ public BigDecimal deposit(String accountNo, String userUuid, BigDecimal amount)
                 .orElseThrow(() -> new IllegalArgumentException("계좌가 없습니다."));
 
         account.validateOwner(userUuid);
-
         account.deposit(amount);
+
         return account.getBalance();
     }
 
@@ -43,11 +64,23 @@ public BigDecimal withdraw(String accountNo, String userUuid, BigDecimal amount)
                 .orElseThrow(() -> new IllegalArgumentException("계좌가 없습니다."));
 
         account.validateOwner(userUuid);
-
         account.withdraw(amount);
+
         return account.getBalance();
     }
 
+    // [Saga용] 계좌번호 없이 ID로 출금
+    @Transactional
+    public void withdrawByLoginId(String loginId, Long amount) {
+        Account account = accountRepository.findByUserUuid(loginId)
+                .orElseThrow(() -> new IllegalArgumentException("해당 유저의 계좌가 존재하지 않습니다."));
+
+        // Long -> BigDecimal 변환 후 출금 처리
+        account.withdraw(BigDecimal.valueOf(amount));
+
+        log.info("📉 [Saga] 출금 처리 완료: user={}, amount={}", loginId, amount);
+    }
+
     // 잔액 조회
     public BigDecimal getBalance(String accountNo) {
         return accountRepository.findById(accountNo)