[DABOM-447] usage-events 정산 단순화 및 알림 outbox 도입

build.gradle

@@ -77,7 +77,7 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter-data-redis'
     runtimeOnly 'com.mysql:mysql-connector-j'
     implementation 'org.springframework.kafka:spring-kafka'
-    implementation 'com.github.da-bom:lib-kafka:v0.5.1'
+    implementation 'com.github.da-bom:lib-kafka:v1.0.0'
 
     // QueryDSL
     implementation 'com.querydsl:querydsl-jpa:5.1.0:jakarta'

src/main/java/com/project/domain/usage/entity/UsageEventOutbox.java

@@ -0,0 +1,87 @@
+package com.project.domain.usage.entity;
+
+import java.time.LocalDateTime;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Index;
+import jakarta.persistence.Table;
+import jakarta.persistence.UniqueConstraint;
+
+import com.project.domain.usage.enums.UsageOutboxStatus;
+import com.project.global.util.BaseEntity;
+
+import lombok.AccessLevel;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+@Entity
+@Getter
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@Table(
+        name = "usage_event_outbox",
+        uniqueConstraints = {
+            @UniqueConstraint(name = "uk_usage_event_outbox_event_id", columnNames = "event_id")
+        },
+        indexes = {
+            @Index(name = "idx_usage_outbox_status_retry", columnList = "status, next_retry_at")
+        })
+public class UsageEventOutbox extends BaseEntity {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @Column(name = "event_id", nullable = false, length = 191)
+    private String eventId;
+
+    @Column(name = "family_id", nullable = false)
+    private Long familyId;
+
+    @Column(name = "customer_id", nullable = false)
+    private Long customerId;
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "status", nullable = false, length = 30)
+    private UsageOutboxStatus status;
+
+    @Column(name = "payload_json", columnDefinition = "TEXT")
+    private String payloadJson;
+
+    @Column(name = "retry_count", nullable = false)
+    private int retryCount;
+
+    @Column(name = "next_retry_at")
+    private LocalDateTime nextRetryAt;
+
+    @Column(name = "last_error", length = 1000)
+    private String lastError;
+
+    @Builder
+    private UsageEventOutbox(
+            Long id,
+            String eventId,
+            Long familyId,
+            Long customerId,
+            UsageOutboxStatus status,
+            String payloadJson,
+            int retryCount,
+            LocalDateTime nextRetryAt,
+            String lastError) {
+        this.id = id;
+        this.eventId = eventId;
+        this.familyId = familyId;
+        this.customerId = customerId;
+        this.status = status;
+        this.payloadJson = payloadJson;
+        this.retryCount = retryCount;
+        this.nextRetryAt = nextRetryAt;
+        this.lastError = lastError;
+    }
+}

src/main/java/com/project/domain/usage/enums/UsageOutboxStatus.java

@@ -0,0 +1,7 @@
+package com.project.domain.usage.enums;
+
+public enum UsageOutboxStatus {
+    PUBLISH_PENDING,
+    SENT,
+    FAILED
+}

src/main/java/com/project/domain/usage/enums/UsagePersistProcessResult.java

@@ -3,6 +3,7 @@
 import java.util.Arrays;
 
 public enum UsagePersistProcessResult {
+    APP_BLOCK(true, "APP_BLOCK"),
     MANUAL(true, "MANUAL"),
     TIME_BLOCK(true, "TIME_BLOCK"),
     MONTHLY_LIMIT_EXCEEDED(true, "MONTHLY_LIMIT_EXCEEDED"),

src/main/java/com/project/domain/usage/repository/UsageEventOutboxRepository.java

@@ -0,0 +1,55 @@
+package com.project.domain.usage.repository;
+
+import java.util.Optional;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+
+import com.project.domain.usage.entity.UsageEventOutbox;
+
+public interface UsageEventOutboxRepository extends JpaRepository<UsageEventOutbox, Long> {
+
+    Optional<UsageEventOutbox> findByEventId(String eventId);
+
+    @Modifying
+    @Query(
+            value =
+                    """
+                    insert ignore into usage_event_outbox
+                    (event_id, family_id, customer_id, status, payload_json, retry_count, created_at, updated_at)
+                    values (:eventId, :familyId, :customerId, 'PUBLISH_PENDING', :payloadJson, 0, now(), now())
+                    """,
+            nativeQuery = true)
+    int insertPublishPendingIgnore(
+            @Param("eventId") String eventId,
+            @Param("familyId") long familyId,
+            @Param("customerId") long customerId,
+            @Param("payloadJson") String payloadJson);
+
+    @Modifying
+    @Query(
+            """
+            update UsageEventOutbox o
+            set o.payloadJson = :payloadJson,
+                o.nextRetryAt = null,
+                o.lastError = null
+            where o.eventId = :eventId
+              and o.status = com.project.domain.usage.enums.UsageOutboxStatus.PUBLISH_PENDING
+            """)
+    int refreshPendingPayload(
+            @Param("eventId") String eventId, @Param("payloadJson") String payloadJson);
+
+    @Modifying
+    @Query(
+            """
+            update UsageEventOutbox o
+            set o.status = com.project.domain.usage.enums.UsageOutboxStatus.SENT,
+                o.nextRetryAt = null,
+                o.lastError = null
+            where o.id = :outboxId
+              and o.status = com.project.domain.usage.enums.UsageOutboxStatus.PUBLISH_PENDING
+            """)
+    int markSentIfPending(@Param("outboxId") Long outboxId);
+}

src/main/java/com/project/domain/usage/service/UsagePersistService.java

@@ -1,8 +1,8 @@
 package com.project.domain.usage.service;
 
-import com.dabom.messaging.kafka.event.dto.EventEnvelope;
-import com.dabom.messaging.kafka.event.dto.usage.UsagePersistPayload;
+import com.dabom.messaging.kafka.event.dto.usage.UsagePayload;
 
 public interface UsagePersistService {
-    void persist(EventEnvelope<UsagePersistPayload> envelope, String recordKey);
+    void persistFromUsageEvent(
+            String eventId, String eventTime, UsagePayload usagePayload, String processResult);
 }

src/main/java/com/project/domain/usage/service/UsagePersistServiceImpl.java

@@ -7,10 +7,10 @@
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
-import com.dabom.messaging.kafka.event.dto.EventEnvelope;
-import com.dabom.messaging.kafka.event.dto.usage.UsagePersistPayload;
+import com.dabom.messaging.kafka.event.dto.usage.UsagePayload;
 import com.project.domain.family.repository.FamilyMemberRepository;
 import com.project.domain.usage.enums.UsagePersistProcessResult;
+import com.project.domain.usage.service.dto.UsagePersistPayload;
 import com.project.domain.usage.service.helper.CustomerQuotaWriter;
 import com.project.domain.usage.service.helper.FamilyQuotaWriter;
 import com.project.domain.usage.service.helper.UsagePersistEventValidator;
@@ -35,25 +35,36 @@ public class UsagePersistServiceImpl implements UsagePersistService {
     private final FamilyQuotaWriter familyQuotaWriter;
     private final LogSanitizer logSanitizer;
 
-    // usage-persist 처리의 전체 흐름을 조율
+    // usage-events 처리 결과를 DB 정산으로 직접 반영한다.
+    @Override
     @Transactional
-    public void persist(EventEnvelope<UsagePersistPayload> envelope, String recordKey) {
-        UsagePersistPayload payload = envelope.payload();
-        String eventId = envelope.eventId();
+    public void persistFromUsageEvent(
+            String eventId, String eventTime, UsagePayload usagePayload, String processResult) {
+        UsagePersistPayload payload =
+                new UsagePersistPayload(
+                        eventId,
+                        usagePayload.familyId(),
+                        usagePayload.customerId(),
+                        usagePayload.bytesUsed(),
+                        usagePayload.appId(),
+                        processResult,
+                        eventTime);
+
+        persistInternal(payload, eventId, String.valueOf(usagePayload.familyId()));
+    }
 
-        // 1) payload 계약 검증
+    // 검증 후 usage_record, quota, family usage를 정산 규칙에 맞게 반영한다.
+    private void persistInternal(UsagePersistPayload payload, String eventId, String recordKey) {
         if (!usagePersistEventValidator.isValidPayload(payload, eventId, recordKey)) {
             return;
         }
-        if (payload == null) {
-            return;
-        }
 
         String originEventId = payload.originEventId();
-        // 2) family-customer 소속 관계 검증
+
+        // 가족-구성원 관계가 유효하지 않으면 정산을 중단한다.
         if (!isValidFamilyMember(payload.familyId(), payload.customerId())) {
             log.warn(
-                    "Skip usage-persist due to invalid family-customer relation. eventId={},"
+                    "Skip usage persistence due to invalid family-customer relation. eventId={},"
                             + " originEventId={}, familyId={}, customerId={}",
                     logSanitizer.sanitize(eventId),
                     logSanitizer.sanitize(originEventId),
@@ -62,34 +73,33 @@ public void persist(EventEnvelope<UsagePersistPayload> envelope, String recordKe
             return;
         }
 
-        // 3) 월 기준 계산 + 처리 결과 해석
         LocalDate currentMonth = resolveCurrentMonth(payload.eventTime());
-        UsagePersistProcessResult processResult =
-                UsagePersistProcessResult.from(payload.processResult());
+        UsagePersistProcessResult result = UsagePersistProcessResult.from(payload.processResult());
 
-        // 차단 이벤트는 usage_record를 남기지 않고 차단 상태만 반영한다.
-        if (processResult.isBlocked()) {
+        // 차단 이벤트는 usage_record를 만들지 않고 차단 상태만 반영한다.
+        if (result.isBlocked()) {
             customerQuotaWriter.persistBlockedQuota(
-                    payload, currentMonth, eventId, originEventId, processResult.blockReason());
+                    payload, currentMonth, eventId, originEventId, result.blockReason());
             return;
         }
 
-        // usage_record 유니크 충돌이면 이미 처리된 이벤트라 quota 반영도 생략한다.
+        // usage_record가 이미 있으면 이미 처리된 이벤트로 본다.
         if (!usageRecordWriter.persistUsageRecord(payload, eventId, originEventId)) {
             return;
         }
 
-        // 4) 허용 이벤트의 월 누적 반영
         customerQuotaWriter.persistAllowedQuota(payload, currentMonth, eventId, originEventId);
         familyQuotaWriter.persistAllowedQuota(
                 payload.familyId(), currentMonth, payload.bytesUsed(), eventId, originEventId);
     }
 
+    // 현재 가족의 유효한 구성원인지 확인한다.
     private boolean isValidFamilyMember(Long familyId, Long customerId) {
         return familyMemberRepository.existsByFamilyIdAndCustomerIdAndDeletedAtIsNull(
                 familyId, customerId);
     }
 
+    // 이벤트 시각을 정산 월로 변환하고 이상 값이면 현재 월로 보정한다.
     private LocalDate resolveCurrentMonth(String eventTime) {
         LocalDate currentMonth = LocalDate.now(TimeConstants.ASIA_SEOUL).withDayOfMonth(1);
         if (eventTime == null || eventTime.isBlank()) {
@@ -101,6 +111,8 @@ private LocalDate resolveCurrentMonth(String eventTime) {
                             .atZone(TimeConstants.ASIA_SEOUL)
                             .toLocalDate()
                             .withDayOfMonth(1);
+
+            // 허용 범위를 벗어난 월은 잘못된 이벤트 시각으로 보고 현재 월로 보정한다.
             if (isOutsideAllowedMonthWindow(parsedMonth, currentMonth)) {
                 log.warn(
                         "Suspicious eventTime month. Fallback to current month. eventTime={},"
@@ -122,8 +134,8 @@ private LocalDate resolveCurrentMonth(String eventTime) {
         }
     }
 
+    // 정산 허용 범위를 벗어난 월인지 확인한다.
     private boolean isOutsideAllowedMonthWindow(LocalDate parsedMonth, LocalDate currentMonth) {
-        // 과거 1개월까지만 허용하고 미래 월은 허용하지 않는다.
         LocalDate minMonth = currentMonth.minusMonths(ALLOWED_PAST_MONTHS);
         LocalDate maxMonth = currentMonth.plusMonths(ALLOWED_FUTURE_MONTHS);
         return parsedMonth.isBefore(minMonth) || parsedMonth.isAfter(maxMonth);