Merge pull request #17 from Octa-Cloud/feat/mong-91/회원-탈퇴로-인한-서비스-전파-개발

Feat/mong 91/회원 탈퇴로 인한 서비스 전파 개발

Author: KoungQ

build.gradle

@@ -33,6 +33,11 @@ dependencies {
     // Kafka
     implementation 'org.springframework.kafka:spring-kafka'
 
+    //resilience4j
+    implementation platform("io.github.resilience4j:resilience4j-bom:2.2.0") // 버전 한줄로 관리
+    implementation "io.github.resilience4j:resilience4j-spring-boot3"       // ★ 필수 (Registry/AutoConfig)
+    implementation "org.springframework.boot:spring-boot-starter-aop"   // 필수
+
     // Swagger
     implementation 'org.springdoc:springdoc-openapi-starter-webmvc-ui:2.6.0'
 

src/main/java/com/project/user/domain/application/usecase/DeleteAccountUseCase.java

@@ -1,27 +1,45 @@
 package com.project.user.domain.application.usecase;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.project.user.domain.application.dto.request.UserDeletionRequest;
 import com.project.user.domain.domain.entity.User;
 import com.project.user.domain.domain.service.UserService;
 import com.project.user.global.exception.RestApiException;
 import com.project.user.global.security.TokenProvider;
+import com.project.user.domain.infra.kafka.ResilientSender;
 import jakarta.transaction.Transactional;
 import lombok.RequiredArgsConstructor;
 import org.mindrot.jbcrypt.BCrypt;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Service;
 
+import java.time.Duration;
+import java.util.UUID;
+
+import static com.project.user.global.exception.code.status.AuthErrorStatus.*;
 import static com.project.user.global.exception.code.status.AuthErrorStatus.INVALID_ID_TOKEN;
 import static com.project.user.global.exception.code.status.AuthErrorStatus.INVALID_PASSWORD;
-
+import lombok.extern.slf4j.Slf4j;
+@Slf4j
 @Service
 @Transactional
 @RequiredArgsConstructor
 public class DeleteAccountUseCase {
 
     private final TokenProvider tokenProvider;
     private final UserService userService;
-
+    private final ObjectMapper om;
+    private final ResilientSender sender;
+    /**
+     * 유저 삭제 흐름의 시작점.
+     * 1) 액세스 토큰에서 userNo 추출 및 사용자 검증
+     * 2) 토큰 TTL(초) 계산 → 오케스트레이터가 블랙리스트 TTL로 사용
+     * 3) user.deletion.start 주제에 사가 시작 이벤트를 "동기 전송(get)"으로 발행
+     *    - acks=all + .get() 으로 브로커 수신을 확인(전송 멱등/안정성 강화)
+     * 컨트롤러는 202 Accepted로 비동기 진행 알림이 자연스럽다.
+     */
     public void execute(String accessToken, UserDeletionRequest request) {
+        log.debug("accessToken userId={}", tokenProvider.getId(accessToken));
         Long userNo = tokenProvider.getId(accessToken)
                 .orElseThrow(() -> new RestApiException(INVALID_ID_TOKEN));
 
@@ -31,8 +49,28 @@ public void execute(String accessToken, UserDeletionRequest request) {
             throw new RestApiException(INVALID_PASSWORD);
         }
 
-        // todo: Kafka - blacklist 전파
+        Duration expiration = tokenProvider.getRemainingDuration(accessToken)
+                .orElseThrow(() -> new RestApiException(INVALID_ACCESS_TOKEN));
+        long ttlSec = Math.max(0, expiration.getSeconds());
 
-        userService.deleteUser(userNo);
+        var start = new StartEvent(
+                UUID.randomUUID().toString(), // ★ sagaId == correlationId
+                userNo,
+                accessToken,
+                ttlSec
+        );
+
+        try {
+            String payload = om.writeValueAsString(start);
+            // key=userNo 로 파티셔닝 → 동일 유저 이벤트 순서 보장
+            //kafka.send("user.start-delete.command", String.valueOf(userNo), payload).get();
+            sender.sendSync("user.start-delete.command", String.valueOf(userNo), payload);
+        } catch (Exception e) {
+            // 프로듀스 실패 → 트랜잭션 롤백(서비스 계층 예외로 래핑)
+            throw new RuntimeException("failed to publish user.start-delete.command", e);
+        }
     }
+
+    // 오케스트레이터가 소비하는 시작 이벤트 포맷
+    public record StartEvent(String sagaId, Long userNo, String accessToken, long accessTokenTtlSec) {}
 }

src/main/java/com/project/user/domain/domain/entity/ProcessedEvent.java

@@ -0,0 +1,45 @@
+package com.project.user.domain.domain.entity;
+
+import jakarta.persistence.*;
+import lombok.*;
+import java.time.Instant;
+
+@Entity
+@Table(name="processed_event")
+@Getter @Setter @NoArgsConstructor @AllArgsConstructor @Builder
+public class ProcessedEvent {
+    /**
+     * eventId 를 PK로 사용 → 동일 이벤트 재처리 방지(멱등).
+     * 상태 전이가 명확: IN_PROGRESS → SUCCESS | ERROR(+attempts++).
+     */
+    @Id
+    @Column(length=36)
+    private String eventId;
+
+    @Column(nullable=false, length=16)
+    private String type;   // FINAL_DELETE
+
+    @Column(nullable=false, length=16)
+    private String status; // IN_PROGRESS | SUCCESS | ERROR
+
+    @Builder.Default
+    @Column(nullable=false)
+    private Integer attempts = 1;
+
+    private String lastError;
+
+    @Builder.Default
+    @Column(nullable=false)
+    private Instant updatedAt = Instant.now();
+
+    // DB 기본값 보장(INSERT/UPDATE 시각)
+    @PrePersist
+    public void prePersist() {
+        if (attempts == null) attempts = 1;
+        if (updatedAt == null) updatedAt = Instant.now();
+    }
+    @PreUpdate
+    public void preUpdate() {
+        this.updatedAt = Instant.now();
+    }
+}

src/main/java/com/project/user/domain/domain/repository/ProcessedEventRepository.java

@@ -0,0 +1,6 @@
+package com.project.user.domain.domain.repository;
+
+import com.project.user.domain.domain.entity.ProcessedEvent;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+public interface ProcessedEventRepository extends JpaRepository<ProcessedEvent, String> {}

src/main/java/com/project/user/domain/domain/service/UserService.java

@@ -1,5 +1,7 @@
 package com.project.user.domain.domain.service;
 
+import com.project.user.domain.application.dto.request.ChangeNicknameRequest;
+import com.project.user.domain.application.dto.request.ChangePasswordRequest;
 import com.project.user.domain.application.dto.request.SignUpRequest;
 import com.project.user.domain.domain.entity.User;
 import com.project.user.domain.domain.repository.UserRepository;
@@ -9,6 +11,9 @@
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.util.Objects;
+
+import static com.project.user.global.exception.code.status.GlobalErrorStatus.PASSWORD_NOT_MATCH;
 import static com.project.user.global.exception.code.status.GlobalErrorStatus._NOT_FOUND;
 
 @Service
@@ -43,8 +48,23 @@ public User findById(Long userNo) {
                 .orElseThrow(() -> new RestApiException(_NOT_FOUND));
     }
 
+    public User updateNickname(Long userNo, ChangeNicknameRequest request) {
+        User user = findById(userNo);
+        user.changeNickname(request.nickname());
+        return user;
+    }
+
+    public void updatePassword(Long userNo, ChangePasswordRequest request) {
+        //비밀번호 == 확인용 비밀번호 검사, 문자열 값 자체 비교
+        if(!Objects.equals(request.password(), request.checkPassword())){
+            throw new RestApiException(PASSWORD_NOT_MATCH);
+        }
+        User user = findById(userNo);
+        user.changePassword(BCrypt.hashpw(request.password(), BCrypt.gensalt(12)));
+    }
     @Transactional
     public void deleteUser(Long userNo) {
-        userRepository.deleteById(userNo);
+        // 멱등: 이미 삭제돼 있어도 예외 없이 통과
+        userRepository.findById(userNo).ifPresent(userRepository::delete);
     }
 }

src/main/java/com/project/user/domain/infra/kafka/DltLogger.java

@@ -0,0 +1,43 @@
+package com.project.user.domain.infra.kafka;
+// 공통 유틸(선택): 헤더를 예쁘게 로깅하고, 카운터 증가
+
+import io.micrometer.core.instrument.MeterRegistry;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class DltLogger {
+
+    private final MeterRegistry meter; // micrometer
+
+    public void logAndCount(
+            String dltTopic,
+            ConsumerRecord<String, String> rec,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_TOPIC, required = false)   String oTopic,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_PARTITION, required = false) Integer oPart,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_OFFSET, required = false)   Long oOffset,
+            @Header(name = KafkaHeaders.DLT_EXCEPTION_FQCN, required = false)    String exClass,
+            @Header(name = KafkaHeaders.DLT_EXCEPTION_MESSAGE, required = false) String exMsg
+    ) {
+        // 메트릭: dlt 카운트
+        meter.counter("kafka.dlt.count",
+                        "dltTopic", dltTopic,
+                        "origTopic", String.valueOf(oTopic))
+                .increment();
+
+        log.error("[DLT] topic={}, origTopic={}, origPartition={}, origOffset={}, key={}, ts={}, exClass={}, exMsg={}, payload={}",
+                dltTopic,
+                oTopic, oPart, oOffset,
+                rec.key(),
+                rec.timestamp(),
+                exClass, exMsg,
+                rec.value()
+        );
+    }
+}

src/main/java/com/project/user/domain/infra/kafka/InvalidPayloadException.java

@@ -0,0 +1,7 @@
+package com.project.user.domain.infra.kafka;
+
+public class InvalidPayloadException extends RuntimeException {
+    // ★ 파싱/검증 불가(독성)인 경우 던져서 @RetryableTopic exclude 로 즉시 DLT 이동
+    public InvalidPayloadException(String message) { super(message); }
+    public InvalidPayloadException(String message, Throwable cause) { super(message, cause); }
+}

src/main/java/com/project/user/domain/infra/kafka/ResilientSender.java

@@ -0,0 +1,28 @@
+package com.project.user.domain.infra.kafka;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+import io.github.resilience4j.retry.annotation.Retry;
+import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
+
+/**
+ * 카프카 전송 래퍼:
+ * - .get() 호출로 브로커 acks=all 까지 동기 확인(실패 시 예외 발생)
+ * - Resilience4j @Retry/@CircuitBreaker 로 일시 장애 완충
+ *   (Fallback 은 재던져서 Outbox 백오프로 연결)
+ */
+@Component @RequiredArgsConstructor
+public class ResilientSender {
+    private final KafkaTemplate<String,String> kafka;
+
+    @Retry(name="kafkaSend")
+    @CircuitBreaker(name="kafkaSend", fallbackMethod = "sendFallback")
+    public void sendSync(String topic, String key, String payload) throws Exception {
+        kafka.send(topic, key, payload).get(); // 동기 전송
+    }
+    @SuppressWarnings("unused")
+    public void sendFallback(String topic, String key, String payload, Throwable t) throws Exception {
+        throw (t instanceof Exception) ? (Exception)t : new RuntimeException(t);
+    }
+}

src/main/java/com/project/user/domain/infra/kafka/UserDltListeners.java

@@ -0,0 +1,27 @@
+package com.project.user.domain.infra.kafka;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class UserDltListeners {
+
+    private final DltLogger dlt;
+
+    @KafkaListener(topics = "user.final-delete.command.dlt", groupId = "user-service")
+    public void onFinalDeleteCmdDlt(
+            ConsumerRecord<String,String> rec,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_TOPIC, required = false)   String oTopic,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_PARTITION, required = false) Integer oPart,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_OFFSET, required = false)   Long oOffset,
+            @Header(name = KafkaHeaders.DLT_EXCEPTION_FQCN, required = false)    String exClass,
+            @Header(name = KafkaHeaders.DLT_EXCEPTION_MESSAGE, required = false) String exMsg
+    ){
+        dlt.logAndCount("user.final-delete.command.dlt", rec, oTopic, oPart, oOffset, exClass, exMsg);
+    }
+}