diff --git a/build.gradle b/build.gradle index ba135f6..e669425 100644 --- a/build.gradle +++ b/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' diff --git a/src/main/java/com/project/domain/usage/entity/UsageEventOutbox.java b/src/main/java/com/project/domain/usage/entity/UsageEventOutbox.java new file mode 100644 index 0000000..879382f --- /dev/null +++ b/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; + } +} diff --git a/src/main/java/com/project/domain/usage/enums/UsageOutboxStatus.java b/src/main/java/com/project/domain/usage/enums/UsageOutboxStatus.java new file mode 100644 index 0000000..ace6f1a --- /dev/null +++ b/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 +} diff --git a/src/main/java/com/project/domain/usage/enums/UsagePersistProcessResult.java b/src/main/java/com/project/domain/usage/enums/UsagePersistProcessResult.java index 0b6c1f3..5bf8fb2 100644 --- a/src/main/java/com/project/domain/usage/enums/UsagePersistProcessResult.java +++ b/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"), diff --git a/src/main/java/com/project/domain/usage/infra/messaging/UsagePersistKafkaConsumer.java b/src/main/java/com/project/domain/usage/infra/messaging/UsagePersistKafkaConsumer.java deleted file mode 100644 index a58ccd7..0000000 --- a/src/main/java/com/project/domain/usage/infra/messaging/UsagePersistKafkaConsumer.java +++ /dev/null @@ -1,47 +0,0 @@ -package com.project.domain.usage.infra.messaging; - -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.springframework.kafka.annotation.KafkaListener; -import org.springframework.stereotype.Component; - -import com.dabom.messaging.kafka.contract.KafkaConsumerGroups; -import com.dabom.messaging.kafka.contract.KafkaEventTypes; -import com.dabom.messaging.kafka.contract.KafkaTopics; -import com.dabom.messaging.kafka.event.KafkaEventMessageSupport; -import com.dabom.messaging.kafka.event.consumer.KafkaEventConsumer; -import com.dabom.messaging.kafka.event.dto.EventEnvelope; -import com.dabom.messaging.kafka.event.dto.usage.UsagePersistPayload; -import com.fasterxml.jackson.core.type.TypeReference; -import com.project.domain.usage.service.UsagePersistService; - -import lombok.RequiredArgsConstructor; - -@Component -@RequiredArgsConstructor -public class UsagePersistKafkaConsumer implements KafkaEventConsumer { - - private final KafkaEventMessageSupport kafkaEventMessageSupport; - private final UsagePersistService usagePersistService; - - @KafkaListener( - topics = KafkaTopics.USAGE_PERSIST, - groupId = KafkaConsumerGroups.DABOM_PROCESSOR_USAGE_PERSISTENCE) - public void consume(ConsumerRecord consumerRecord) { - consume(consumerRecord, kafkaEventMessageSupport); - } - - @Override - public String eventType() { - return KafkaEventTypes.USAGE_PERSIST; - } - - @Override - public TypeReference> typeReference() { - return new TypeReference<>() {}; - } - - @Override - public void handle(EventEnvelope envelope, String recordKey) { - usagePersistService.persist(envelope, recordKey); - } -} diff --git a/src/main/java/com/project/domain/usage/repository/UsageEventOutboxRepository.java b/src/main/java/com/project/domain/usage/repository/UsageEventOutboxRepository.java new file mode 100644 index 0000000..f614720 --- /dev/null +++ b/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 { + + Optional 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); +} diff --git a/src/main/java/com/project/domain/usage/service/UsagePersistService.java b/src/main/java/com/project/domain/usage/service/UsagePersistService.java index bf17ff4..53eac09 100644 --- a/src/main/java/com/project/domain/usage/service/UsagePersistService.java +++ b/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 envelope, String recordKey); + void persistFromUsageEvent( + String eventId, String eventTime, UsagePayload usagePayload, String processResult); } diff --git a/src/main/java/com/project/domain/usage/service/UsagePersistServiceImpl.java b/src/main/java/com/project/domain/usage/service/UsagePersistServiceImpl.java index cfb3751..9540f6f 100644 --- a/src/main/java/com/project/domain/usage/service/UsagePersistServiceImpl.java +++ b/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 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 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); diff --git a/src/main/java/com/project/domain/usage/service/UsageSyncServiceImpl.java b/src/main/java/com/project/domain/usage/service/UsageSyncServiceImpl.java index eab3385..4f98e6d 100644 --- a/src/main/java/com/project/domain/usage/service/UsageSyncServiceImpl.java +++ b/src/main/java/com/project/domain/usage/service/UsageSyncServiceImpl.java @@ -7,17 +7,24 @@ import java.util.Locale; import org.springframework.beans.factory.annotation.Value; +import org.springframework.kafka.support.SendResult; import org.springframework.stereotype.Service; import com.dabom.messaging.kafka.contract.KafkaConsumerGroups; import com.dabom.messaging.kafka.contract.KafkaEventTypes; import com.dabom.messaging.kafka.contract.KafkaTopics; +import com.dabom.messaging.kafka.error.KafkaMessageProcessingException; +import com.dabom.messaging.kafka.event.dto.notification.NotificationPayload; import com.dabom.messaging.kafka.event.dto.usage.UsagePayload; import com.dabom.messaging.kafka.metrics.KafkaMetrics; import com.project.domain.policy.service.helper.PolicyConstraintWarmupHelper; import com.project.domain.usage.service.dto.UsageUpdateResult; -import com.project.domain.usage.service.helper.UsageEventPublisher; +import com.project.domain.usage.service.helper.UsageEventOutboxService; +import com.project.domain.usage.service.helper.UsageFamilyMembershipCacheHelper; import com.project.domain.usage.service.helper.UsageLuaExecutor; +import com.project.domain.usage.service.helper.UsageNotificationPayloadMapper; +import com.project.domain.usage.service.helper.UsageNotificationPublisher; +import com.project.domain.usage.service.helper.UsageProcessingDecisionMapper; import com.project.domain.usage.service.helper.UsageRedisWarmupHelper; import com.project.global.common.TimeConstants; import com.project.global.util.LogSanitizer; @@ -38,25 +45,31 @@ public class UsageSyncServiceImpl implements UsageSyncService { private final UsageRedisWarmupHelper usageRedisWarmupHelper; private final PolicyConstraintWarmupHelper policyConstraintWarmupHelper; private final UsageLuaExecutor usageLuaExecutor; - private final UsageEventPublisher usageEventPublisher; + private final UsagePersistService usagePersistService; + private final UsageEventOutboxService usageEventOutboxService; + private final UsageProcessingDecisionMapper usageProcessingDecisionMapper; + private final UsageNotificationPayloadMapper usageNotificationPayloadMapper; + private final UsageNotificationPublisher usageNotificationPublisher; + private final UsageFamilyMembershipCacheHelper usageFamilyMembershipCacheHelper; private final LogSanitizer logSanitizer; private final KafkaMetrics kafkaMetrics; @Value("${app.kafka.dedup.usage-ttl-seconds}") private long dedupTtlSeconds; + // usage-events 1건을 검증하고 Redis/Lua/DB 정산/알림 발행까지 처리한다. @Override public void syncUsage(String eventId, String eventTime, UsagePayload payload) { + long familyId = payload.familyId(); + long customerId = payload.customerId(); - Long familyId = payload.familyId(); - Long customerId = payload.customerId(); - long usageBytes = payload.bytesUsed(); + // 잘못된 family-customer 조합은 초입에서 바로 차단한다. + validateFamilyMembership(eventId, familyId, customerId); - // 1) eventTime 해석 + 월 키 기준 계산 LocalDateTime resolvedEventDateTime = resolveEventDateTime(eventTime); LocalDate eventMonth = resolvedEventDateTime.toLocalDate().withDayOfMonth(1); - // 2) Lua 실행에 필요한 Redis 키 생성 + String normalizedAppId = normalizeAppId(payload.appId()); String infoKey = redisKeyGenerator.generateFamilyInfoKey(familyId, eventMonth); String remainingKey = redisKeyGenerator.generateFamilyRemainingKey(familyId, eventMonth); String monthlyKey = @@ -64,33 +77,36 @@ public void syncUsage(String eventId, String eventTime, UsagePayload payload) { familyId, customerId, eventMonth); String constraintsKey = redisKeyGenerator.generateFamilyCustomerConstraintsKey(familyId, customerId); - String alert50Key = redisKeyGenerator.generateFamilyAlertKey(familyId, 50, eventMonth); - String alert30Key = redisKeyGenerator.generateFamilyAlertKey(familyId, 30, eventMonth); - String alert10Key = redisKeyGenerator.generateFamilyAlertKey(familyId, 10, eventMonth); + String alert50Key = + redisKeyGenerator.generateFamilyCustomerThresholdAlertKey( + familyId, customerId, 50, eventMonth); + String alert30Key = + redisKeyGenerator.generateFamilyCustomerThresholdAlertKey( + familyId, customerId, 30, eventMonth); + String alert10Key = + redisKeyGenerator.generateFamilyCustomerThresholdAlertKey( + familyId, customerId, 10, eventMonth); + String manualAlertKey = + redisKeyGenerator.generateFamilyCustomerBlockAlertKey( + familyId, customerId, "MANUAL", eventMonth); + String appBlockAlertKey = + redisKeyGenerator.generateFamilyCustomerAppBlockAlertKey( + familyId, customerId, normalizedAppId, eventMonth); + String timeBlockAlertKey = + redisKeyGenerator.generateFamilyCustomerBlockAlertKey( + familyId, customerId, "TIME_BLOCK", eventMonth); + String monthlyLimitAlertKey = + redisKeyGenerator.generateFamilyCustomerBlockAlertKey( + familyId, customerId, "MONTHLY_LIMIT_EXCEEDED", eventMonth); + String familyQuotaAlertKey = + redisKeyGenerator.generateFamilyCustomerBlockAlertKey( + familyId, customerId, "FAMILY_QUOTA_EXCEEDED", eventMonth); String dedupKey = redisKeyGenerator.generateUsageEventDedupKey(eventId); - // 3) Redis warmup 보장 - boolean familyInfoRedisWarmup = - usageRedisWarmupHelper.ensureFamilyInfoCached(familyId, eventMonth, infoKey); - boolean familyRemainingRedisWarmup = - usageRedisWarmupHelper.ensureRemainingBytesCached( - familyId, eventMonth, remainingKey); - boolean customerMonthlyUsageRedisWarmup = - usageRedisWarmupHelper.ensureCustomerUsageCached( - familyId, customerId, monthlyKey, eventMonth); - policyConstraintWarmupHelper.warmupIfMissing(familyId, customerId); + // Redis 상태가 준비되지 않으면 Lua 판단을 태우지 않는다. + ensureWarmupOrThrow( + eventId, familyId, customerId, eventMonth, infoKey, remainingKey, monthlyKey); - if (!familyInfoRedisWarmup - || !familyRemainingRedisWarmup - || !customerMonthlyUsageRedisWarmup) { - log.error("Redis Warmup is Failed. eventId={}", logSanitizer.sanitize(eventId)); - return; - } - - String currentHhmm = resolvedEventDateTime.format(HHMM_FORMATTER); - String normalizedAppId = normalizeAppId(payload.appId()); - - // 4) Lua로 정책 판정 + 사용량 반영 + dedup 검사 수행 UsageUpdateResult parsed = usageLuaExecutor.execute( new UsageLuaExecutor.UsageLuaCommand( @@ -101,37 +117,126 @@ public void syncUsage(String eventId, String eventTime, UsagePayload payload) { alert50Key, alert30Key, alert10Key, + manualAlertKey, + appBlockAlertKey, + timeBlockAlertKey, + monthlyLimitAlertKey, + familyQuotaAlertKey, dedupKey, - usageBytes, - currentHhmm, + payload.bytesUsed(), + resolvedEventDateTime.format(HHMM_FORMATTER), normalizedAppId, dedupTtlSeconds), eventId); + log.debug( - "Usage Synced: family={}, customer={}, status={}", + "Usage synced: family={}, customer={}, status={}, notify={}, duplicate={}", familyId, customerId, - parsed.status()); + parsed.status(), + parsed.shouldNotify(), + parsed.duplicate()); - // 5) duplicate면 후속 publish 없이 종료 if (parsed.duplicate()) { kafkaMetrics.incrementDedupHit( KafkaTopics.USAGE_EVENTS, KafkaConsumerGroups.DABOM_PROCESSOR_USAGE_MAIN, KafkaEventTypes.DATA_USAGE); log.info( - "Skip duplicated usage event. eventId={}, familyId={}, customerId={}", + "Duplicate usage event. eventId={}, familyId={}, customerId={}", logSanitizer.sanitize(eventId), familyId, customerId); + } + + // Lua 상태는 중앙 매퍼에서만 해석한다. + UsageProcessingDecisionMapper.UsageProcessingDecision decision = + usageProcessingDecisionMapper.fromLuaStatus(parsed.status()); + + // DB 정산은 usage_record unique와 quota 갱신 규칙으로 멱등하게 재진입한다. + usagePersistService.persistFromUsageEvent( + eventId, eventTime, payload, decision.persistProcessResult()); + + boolean publishNotification = decision.publishNotification() && parsed.shouldNotify(); + if (!publishNotification) { + dispatchPendingNotificationIfExists(eventId); + return; + } + + NotificationPayload notificationPayload = + usageNotificationPayloadMapper.toNotificationPayload( + eventId, resolvedEventDateTime, payload, decision.notificationStatus()); + + usageEventOutboxService + .stageAfterRedisApplied(eventId, notificationPayload, true) + .ifPresent(this::publishAsync); + } + + // family-customer 관계가 틀리면 invalid payload로 간주하고 중단한다. + private void validateFamilyMembership(String eventId, long familyId, long customerId) { + if (usageFamilyMembershipCacheHelper.isValidFamilyCustomer(familyId, customerId)) { return; } + throw new IllegalArgumentException( + "Invalid family-customer relation. eventId=%s familyId=%d customerId=%d" + .formatted(eventId, familyId, customerId)); + } - // 6) downstream 이벤트 발행 - usageEventPublisher.publish( - new UsageEventPublisher.UsageEventContext(eventId, eventTime, payload, parsed)); + // warmup 실패는 일시 장애로 보고 retryable 예외로 전파한다. + private void ensureWarmupOrThrow( + String eventId, + long familyId, + long customerId, + LocalDate eventMonth, + String infoKey, + String remainingKey, + String monthlyKey) { + boolean familyInfoRedisWarmup = + usageRedisWarmupHelper.ensureFamilyInfoCached(familyId, eventMonth, infoKey); + boolean familyRemainingRedisWarmup = + usageRedisWarmupHelper.ensureRemainingBytesCached( + familyId, eventMonth, remainingKey); + boolean customerMonthlyUsageRedisWarmup = + usageRedisWarmupHelper.ensureCustomerUsageCached( + familyId, customerId, monthlyKey, eventMonth); + policyConstraintWarmupHelper.warmupIfMissing(familyId, customerId); + + if (!familyInfoRedisWarmup + || !familyRemainingRedisWarmup + || !customerMonthlyUsageRedisWarmup) { + log.error("Redis warmup failed. eventId={}", logSanitizer.sanitize(eventId)); + throw new KafkaMessageProcessingException( + "Redis warmup failed. eventId=%s familyId=%d customerId=%d" + .formatted(eventId, familyId, customerId), + new IllegalStateException("Redis warmup failed")); + } } + // 이미 만들어진 pending notification이 있으면 다시 즉시 발행을 시도한다. + private void dispatchPendingNotificationIfExists(String eventId) { + usageEventOutboxService.findPendingDispatchByEventId(eventId).ifPresent(this::publishAsync); + } + + // notification은 비동기로 발행하고 성공 시에만 SENT로 마감한다. + private void publishAsync(UsageEventOutboxService.PendingNotificationDispatch pending) { + usageNotificationPublisher + .publishAsync(pending.payload()) + .whenComplete( + (SendResult ignored, Throwable throwable) -> { + if (throwable == null) { + usageEventOutboxService.markSent(pending.outboxId()); + return; + } + + log.warn( + "Notification publish deferred to batch retry. outboxId={}," + + " reason={}", + pending.outboxId(), + throwable.getMessage()); + }); + } + + // eventTime을 파싱하고 실패하면 현재 시각으로 보정한다. private LocalDateTime resolveEventDateTime(String eventTime) { if (eventTime != null && !eventTime.isBlank()) { try { @@ -143,12 +248,11 @@ private LocalDateTime resolveEventDateTime(String eventTime) { return LocalDateTime.now(TimeConstants.ASIA_SEOUL); } + // 앱 차단 키 비교에 사용하도록 appId를 정규화한다. private String normalizeAppId(String appId) { if (appId == null) { return EMPTY_APP_ID; } - - // app 차단 정책 키와 비교할 수 있게 정규화 String normalized = appId.trim().toLowerCase(Locale.ROOT); return normalized.isEmpty() ? EMPTY_APP_ID : normalized; } diff --git a/src/main/java/com/project/domain/usage/service/dto/UsagePersistPayload.java b/src/main/java/com/project/domain/usage/service/dto/UsagePersistPayload.java new file mode 100644 index 0000000..23cbe9e --- /dev/null +++ b/src/main/java/com/project/domain/usage/service/dto/UsagePersistPayload.java @@ -0,0 +1,11 @@ +package com.project.domain.usage.service.dto; + +// usage-events 처리 결과를 DB 정산 로직으로 넘길 때 사용하는 내부 DTO다. +public record UsagePersistPayload( + String originEventId, + Long familyId, + Long customerId, + Long bytesUsed, + String appId, + String processResult, + String eventTime) {} diff --git a/src/main/java/com/project/domain/usage/service/dto/UsageUpdateResult.java b/src/main/java/com/project/domain/usage/service/dto/UsageUpdateResult.java index 237605a..c5ace33 100644 --- a/src/main/java/com/project/domain/usage/service/dto/UsageUpdateResult.java +++ b/src/main/java/com/project/domain/usage/service/dto/UsageUpdateResult.java @@ -7,5 +7,7 @@ public record UsageUpdateResult( long monthlyUsed, double userRatio, long monthlyLimit, - // usage-event duplicate 여부 + // 현재 상태에 대해 알림 발행이 필요한지 여부다. + boolean shouldNotify, + // usage-event duplicate 여부다. boolean duplicate) {} diff --git a/src/main/java/com/project/domain/usage/service/helper/CustomerQuotaWriter.java b/src/main/java/com/project/domain/usage/service/helper/CustomerQuotaWriter.java index 9fdebb0..b198111 100644 --- a/src/main/java/com/project/domain/usage/service/helper/CustomerQuotaWriter.java +++ b/src/main/java/com/project/domain/usage/service/helper/CustomerQuotaWriter.java @@ -5,9 +5,9 @@ import org.springframework.dao.DataIntegrityViolationException; import org.springframework.stereotype.Service; -import com.dabom.messaging.kafka.event.dto.usage.UsagePersistPayload; import com.project.domain.customer.entity.CustomerQuota; import com.project.domain.customer.repository.CustomerQuotaRepository; +import com.project.domain.usage.service.dto.UsagePersistPayload; import com.project.global.util.LogSanitizer; import lombok.RequiredArgsConstructor; diff --git a/src/main/java/com/project/domain/usage/service/helper/UsageEventOutboxService.java b/src/main/java/com/project/domain/usage/service/helper/UsageEventOutboxService.java new file mode 100644 index 0000000..362b93e --- /dev/null +++ b/src/main/java/com/project/domain/usage/service/helper/UsageEventOutboxService.java @@ -0,0 +1,92 @@ +package com.project.domain.usage.service.helper; + +import java.util.Optional; + +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import com.dabom.messaging.kafka.error.NonRetryableKafkaMessageProcessingException; +import com.dabom.messaging.kafka.event.dto.notification.NotificationPayload; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.project.domain.usage.enums.UsageOutboxStatus; +import com.project.domain.usage.repository.UsageEventOutboxRepository; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +@Service +@RequiredArgsConstructor +public class UsageEventOutboxService { + + private final UsageEventOutboxRepository usageEventOutboxRepository; + private final ObjectMapper objectMapper; + + // notification 대상인 경우에만 PUBLISH_PENDING row를 보장한다. + @Transactional + public Optional stageAfterRedisApplied( + String eventId, NotificationPayload payload, boolean publishNotification) { + if (!publishNotification) { + return Optional.empty(); + } + + String payloadJson = toJson(payload); + usageEventOutboxRepository.insertPublishPendingIgnore( + eventId, payload.familyId(), payload.customerId(), payloadJson); + usageEventOutboxRepository.refreshPendingPayload(eventId, payloadJson); + return usageEventOutboxRepository + .findByEventId(eventId) + .filter(row -> row.getStatus() == UsageOutboxStatus.PUBLISH_PENDING) + .map( + row -> + new PendingNotificationDispatch( + row.getId(), + fromJson(row.getPayloadJson(), NotificationPayload.class))); + } + + // eventId 기준으로 아직 발행되지 않은 notification payload를 찾는다. + @Transactional(readOnly = true) + public Optional findPendingDispatchByEventId(String eventId) { + return usageEventOutboxRepository + .findByEventId(eventId) + .filter(row -> row.getStatus() == UsageOutboxStatus.PUBLISH_PENDING) + .map( + row -> + new PendingNotificationDispatch( + row.getId(), + fromJson(row.getPayloadJson(), NotificationPayload.class))); + } + + // 발행 성공 시 Outbox 상태를 SENT로 변경한다. + @Transactional + public void markSent(Long outboxId) { + int updated = usageEventOutboxRepository.markSentIfPending(outboxId); + if (updated == 0) { + log.debug("Skip markSent because outbox is no longer pending. outboxId={}", outboxId); + } + } + + // 저장한 payload_json을 지정한 타입으로 역직렬화한다. + public T fromJson(String payloadJson, Class clazz) { + try { + return objectMapper.readValue(payloadJson, clazz); + } catch (JsonProcessingException e) { + throw new NonRetryableKafkaMessageProcessingException( + "Failed to deserialize outbox payload", e); + } + } + + // Outbox payload를 JSON 문자열로 직렬화한다. + private String toJson(Object payload) { + try { + return objectMapper.writeValueAsString(payload); + } catch (JsonProcessingException e) { + throw new NonRetryableKafkaMessageProcessingException( + "Failed to serialize outbox payload", e); + } + } + + // 즉시 발행 또는 배치 발행에 사용하는 pending payload 묶음이다. + public record PendingNotificationDispatch(Long outboxId, NotificationPayload payload) {} +} diff --git a/src/main/java/com/project/domain/usage/service/helper/UsageEventPublisher.java b/src/main/java/com/project/domain/usage/service/helper/UsageEventPublisher.java deleted file mode 100644 index 32297ec..0000000 --- a/src/main/java/com/project/domain/usage/service/helper/UsageEventPublisher.java +++ /dev/null @@ -1,107 +0,0 @@ -package com.project.domain.usage.service.helper; - -import org.springframework.stereotype.Component; - -import com.dabom.messaging.kafka.contract.KafkaEventTypes; -import com.dabom.messaging.kafka.contract.KafkaTopics; -import com.dabom.messaging.kafka.event.dto.notification.CustomerBlockedPayload; -import com.dabom.messaging.kafka.event.dto.notification.NotificationEventSupport; -import com.dabom.messaging.kafka.event.dto.notification.ThresholdAlertPayload; -import com.dabom.messaging.kafka.event.dto.usage.UsagePayload; -import com.dabom.messaging.kafka.event.dto.usage.UsagePersistPayload; -import com.dabom.messaging.kafka.event.dto.usage.UsageRealtimePayload; -import com.dabom.messaging.kafka.event.publisher.KafkaEventPublisher; -import com.project.domain.usage.service.dto.UsageUpdateResult; - -import lombok.RequiredArgsConstructor; - -@Component -@RequiredArgsConstructor -public class UsageEventPublisher { - - private static final String STATUS_APP_BLOCK = "APP_BLOCK"; - private static final String STATUS_WARNING_PREFIX = "WARNING"; - private static final String STATUS_NORMAL_PREFIX = "NORMAL"; - private static final String PERSIST_STATUS_ALLOWED = "ALLOWED"; - - private final KafkaEventPublisher kafkaEventPublisher; - - public void publish(UsageEventContext ctx) { - - UsagePayload payload = ctx.payload(); - UsageUpdateResult result = ctx.result(); - - long familyId = payload.familyId(); - long customerId = payload.customerId(); - - long totalUsed = result.totalUsed(); - long remaining = result.remaining(); - String status = result.status(); - long monthlyUsed = result.monthlyUsed(); - double userRatio = result.userRatio(); - long monthlyLimit = result.monthlyLimit(); - - long totalLimit = totalUsed + remaining; - double usedPercent = totalLimit > 0 ? (double) totalUsed / totalLimit * 100.0 : 0.0; - - if (!STATUS_APP_BLOCK.equals(status)) { - kafkaEventPublisher.publish( - KafkaTopics.USAGE_PERSIST, - KafkaEventTypes.USAGE_PERSIST, - new UsagePersistPayload( - ctx.eventId(), - familyId, - customerId, - payload.bytesUsed(), - payload.appId(), - status.startsWith(STATUS_WARNING_PREFIX) - || status.equals(STATUS_NORMAL_PREFIX) - ? PERSIST_STATUS_ALLOWED - : status, - ctx.eventTime())); - - kafkaEventPublisher.publish( - KafkaTopics.USAGE_REALTIME, - KafkaEventTypes.USAGE_REALTIME, - new UsageRealtimePayload( - familyId, - customerId, - totalUsed, - totalLimit, - remaining, - usedPercent, - monthlyUsed, - userRatio * 100.0, - monthlyLimit)); - } - - if (status.startsWith(STATUS_WARNING_PREFIX)) { - int percent = parsePercent(status); - kafkaEventPublisher.publish( - KafkaTopics.NOTIFICATION, - NotificationEventSupport.toEnvelope( - new ThresholdAlertPayload( - familyId, - percent, - String.format("가족 데이터가 %d%% 미만입니다.", percent)))); - - } else if (!status.startsWith(STATUS_NORMAL_PREFIX)) { - kafkaEventPublisher.publish( - KafkaTopics.NOTIFICATION, - NotificationEventSupport.toEnvelope( - new CustomerBlockedPayload( - familyId, customerId, status, ctx.eventTime()))); - } - } - - private int parsePercent(String status) { - try { - return Integer.parseInt(status.split("_")[1]); - } catch (NumberFormatException | ArrayIndexOutOfBoundsException e) { - throw new IllegalArgumentException("Invalid warning status format: " + status, e); - } - } - - public record UsageEventContext( - String eventId, String eventTime, UsagePayload payload, UsageUpdateResult result) {} -} diff --git a/src/main/java/com/project/domain/usage/service/helper/UsageEventValidator.java b/src/main/java/com/project/domain/usage/service/helper/UsageEventValidator.java index 2b81d62..1f32f54 100644 --- a/src/main/java/com/project/domain/usage/service/helper/UsageEventValidator.java +++ b/src/main/java/com/project/domain/usage/service/helper/UsageEventValidator.java @@ -9,13 +9,13 @@ @Slf4j @Component public class UsageEventValidator { + + // usage-events payload의 기본 형식을 검증한다. public boolean isValid(UsagePayload payload, String eventId) { - // Payload 자체 null 체크 if (payload == null) { log.warn("Usage payload is null. eventId={}", eventId); return false; } - // 필수 ID 값 체크 (FamilyId, CustomerId) if (payload.familyId() == null || payload.familyId() <= 0) { log.warn("Invalid familyId. eventId={}, familyId={}", eventId, payload.familyId()); return false; @@ -25,8 +25,8 @@ public boolean isValid(UsagePayload payload, String eventId) { "Invalid customerId. eventId={}, customerId={}", eventId, payload.customerId()); return false; } - // 사용량 값 체크 (음수, 0) - if (payload.bytesUsed() == null || payload.bytesUsed() < 0) { + // 사용량은 양수만 허용한다. + if (payload.bytesUsed() == null || payload.bytesUsed() <= 0) { log.warn("Invalid bytesUsed. eventId={}, bytes={}", eventId, payload.bytesUsed()); return false; } diff --git a/src/main/java/com/project/domain/usage/service/helper/UsageFamilyMembershipCacheHelper.java b/src/main/java/com/project/domain/usage/service/helper/UsageFamilyMembershipCacheHelper.java new file mode 100644 index 0000000..1dfc3f0 --- /dev/null +++ b/src/main/java/com/project/domain/usage/service/helper/UsageFamilyMembershipCacheHelper.java @@ -0,0 +1,102 @@ +package com.project.domain.usage.service.helper; + +import java.util.List; +import java.util.concurrent.TimeUnit; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.stereotype.Component; + +import com.dabom.messaging.kafka.error.KafkaMessageProcessingException; +import com.project.domain.family.repository.FamilyMemberRepository; +import com.project.global.util.RedisKeyGenerator; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +@Component +@RequiredArgsConstructor +public class UsageFamilyMembershipCacheHelper { + + private final StringRedisTemplate stringRedisTemplate; + private final FamilyMemberRepository familyMemberRepository; + private final RedisKeyGenerator redisKeyGenerator; + + @Value("${app.redis.membership-cache-ttl-seconds:600}") + private long membershipCacheTtlSeconds; + + // family-customer 관계를 Redis 우선으로 검증하고 miss면 DB fallback 한다. + public boolean isValidFamilyCustomer(long familyId, long customerId) { + String membersKey = redisKeyGenerator.generateFamilyMembersKey(familyId); + + try { + Boolean hasMembersKey = stringRedisTemplate.hasKey(membersKey); + if (Boolean.TRUE.equals(hasMembersKey)) { + Boolean isMember = + stringRedisTemplate + .opsForSet() + .isMember(membersKey, String.valueOf(customerId)); + if (Boolean.TRUE.equals(isMember)) { + return true; + } + + // 캐시가 오래되었을 수 있으므로 false일 때도 DB fallback으로 한 번 더 검증한다. + return fallbackAndCacheMembership(familyId, customerId, membersKey); + } + + return warmupFamilyMembersAndCheck(familyId, customerId, membersKey); + } catch (DataAccessException e) { + log.error( + "Family membership cache access failed. familyId={}, customerId={}", + familyId, + customerId, + e); + return checkMembershipFromDatabase(familyId, customerId, membersKey); + } + } + + // DB에서 가족 구성원 전체를 읽어 Redis set을 채운 뒤 포함 여부를 확인한다. + private boolean warmupFamilyMembersAndCheck(long familyId, long customerId, String membersKey) { + List members = + familyMemberRepository.findAllActiveTargetsByFamilyId(familyId); + if (members.isEmpty()) { + return false; + } + + String[] memberIds = + members.stream() + .map(FamilyMemberRepository.FamilyMemberTargetProjection::getCustomerId) + .map(String::valueOf) + .toArray(String[]::new); + + stringRedisTemplate.opsForSet().add(membersKey, memberIds); + stringRedisTemplate.expire(membersKey, membershipCacheTtlSeconds, TimeUnit.SECONDS); + return Boolean.TRUE.equals( + stringRedisTemplate.opsForSet().isMember(membersKey, String.valueOf(customerId))); + } + + // 캐시 불일치 시 DB fallback 결과를 Redis에 반영한다. + private boolean fallbackAndCacheMembership(long familyId, long customerId, String membersKey) { + boolean exists = checkMembershipFromDatabase(familyId, customerId, membersKey); + if (exists) { + stringRedisTemplate.opsForSet().add(membersKey, String.valueOf(customerId)); + stringRedisTemplate.expire(membersKey, membershipCacheTtlSeconds, TimeUnit.SECONDS); + } + return exists; + } + + // DB fallback도 실패하면 retryable 예외로 전파해 membership 검증 자체를 재시도한다. + private boolean checkMembershipFromDatabase(long familyId, long customerId, String membersKey) { + try { + return familyMemberRepository.existsByFamilyIdAndCustomerIdAndDeletedAtIsNull( + familyId, customerId); + } catch (DataAccessException e) { + throw new KafkaMessageProcessingException( + "Family membership lookup failed. familyId=%d customerId=%d key=%s" + .formatted(familyId, customerId, membersKey), + e); + } + } +} diff --git a/src/main/java/com/project/domain/usage/service/helper/UsageLuaExecutor.java b/src/main/java/com/project/domain/usage/service/helper/UsageLuaExecutor.java index 5e35b9d..5635119 100644 --- a/src/main/java/com/project/domain/usage/service/helper/UsageLuaExecutor.java +++ b/src/main/java/com/project/domain/usage/service/helper/UsageLuaExecutor.java @@ -6,6 +6,7 @@ import org.springframework.data.redis.core.script.RedisScript; import org.springframework.stereotype.Component; +import com.dabom.messaging.kafka.error.KafkaMessageProcessingException; import com.project.domain.usage.service.dto.UsageUpdateResult; import com.project.global.util.LogSanitizer; @@ -21,7 +22,7 @@ public class UsageLuaExecutor { private final RedisScript> usageUpdateScript; private final LogSanitizer logSanitizer; - // usage Lua 실행 + 결과 파싱 + // usage Lua를 실행하고 결과를 파싱한다. public UsageUpdateResult execute(UsageLuaCommand command, String eventId) { List result = redisTemplate.execute( @@ -34,6 +35,11 @@ public UsageUpdateResult execute(UsageLuaCommand command, String eventId) { command.alert50Key(), command.alert30Key(), command.alert10Key(), + command.manualAlertKey(), + command.appBlockAlertKey(), + command.timeBlockAlertKey(), + command.monthlyLimitAlertKey(), + command.familyQuotaAlertKey(), command.dedupKey()), String.valueOf(command.usageBytes()), command.currentHhmm(), @@ -42,20 +48,24 @@ public UsageUpdateResult execute(UsageLuaCommand command, String eventId) { if (result == null || result.isEmpty()) { log.error("Usage update script returned null. eventId={}", eventId); - throw new IllegalStateException("Usage update script returned null"); + throw new KafkaMessageProcessingException( + "Usage Lua returned null. eventId=%s".formatted(eventId), + new IllegalStateException("Usage update script returned null")); } return parseScriptResult(result, eventId); } - // Lua 결과 파싱 + // Lua 결과 배열을 UsageUpdateResult로 변환한다. private UsageUpdateResult parseScriptResult(List result, String eventId) { - if (result.size() < 7) { + if (result.size() < 8) { log.error( "Usage update script returned invalid result. eventId={}, result={}", logSanitizer.sanitize(eventId), result); - throw new IllegalStateException("Invalid Lua script result"); + throw new KafkaMessageProcessingException( + "Usage Lua returned invalid result. eventId=%s".formatted(eventId), + new IllegalStateException("Invalid Lua script result")); } long totalUsed = ((Number) result.get(0)).longValue(); @@ -70,14 +80,21 @@ private UsageUpdateResult parseScriptResult(List result, String eventId) : Double.parseDouble(userRatioObj.toString()); long monthlyLimit = ((Number) result.get(5)).longValue(); - // 마지막 값은 duplicate 여부 - boolean duplicate = ((Number) result.get(6)).longValue() == 1L; + boolean shouldNotify = ((Number) result.get(6)).longValue() == 1L; + boolean duplicate = ((Number) result.get(7)).longValue() == 1L; return new UsageUpdateResult( - totalUsed, remaining, status, monthlyUsed, userRatio, monthlyLimit, duplicate); + totalUsed, + remaining, + status, + monthlyUsed, + userRatio, + monthlyLimit, + shouldNotify, + duplicate); } - // Lua 실행에 필요한 인자 묶음 + // Lua 실행에 필요한 인자를 묶는다. public record UsageLuaCommand( String infoKey, String remainingKey, @@ -86,6 +103,11 @@ public record UsageLuaCommand( String alert50Key, String alert30Key, String alert10Key, + String manualAlertKey, + String appBlockAlertKey, + String timeBlockAlertKey, + String monthlyLimitAlertKey, + String familyQuotaAlertKey, String dedupKey, long usageBytes, String currentHhmm, diff --git a/src/main/java/com/project/domain/usage/service/helper/UsageNotificationPayloadMapper.java b/src/main/java/com/project/domain/usage/service/helper/UsageNotificationPayloadMapper.java new file mode 100644 index 0000000..6348fa2 --- /dev/null +++ b/src/main/java/com/project/domain/usage/service/helper/UsageNotificationPayloadMapper.java @@ -0,0 +1,117 @@ +package com.project.domain.usage.service.helper; + +import java.time.LocalDateTime; +import java.util.HashMap; +import java.util.Map; + +import org.springframework.stereotype.Component; + +import com.dabom.messaging.kafka.event.dto.notification.NotificationPayload; +import com.dabom.messaging.kafka.event.dto.notification.NotificationType; +import com.dabom.messaging.kafka.event.dto.usage.UsagePayload; + +@Component +public class UsageNotificationPayloadMapper { + + // usage 처리 결과를 notification payload로 변환한다. + public NotificationPayload toNotificationPayload( + String eventId, + LocalDateTime eventDateTime, + UsagePayload usagePayload, + String notificationStatus) { + return switch (notificationStatus) { + case "WARNING_50" -> + buildThresholdAlert( + eventId, eventDateTime, usagePayload, notificationStatus, 50); + case "WARNING_30" -> + buildThresholdAlert( + eventId, eventDateTime, usagePayload, notificationStatus, 30); + case "WARNING_10" -> + buildThresholdAlert( + eventId, eventDateTime, usagePayload, notificationStatus, 10); + case "MANUAL", + "APP_BLOCK", + "TIME_BLOCK", + "MONTHLY_LIMIT_EXCEEDED", + "FAMILY_QUOTA_EXCEEDED" -> + buildBlockedAlert(eventId, eventDateTime, usagePayload, notificationStatus); + default -> + throw new IllegalArgumentException( + "Unsupported notification status: " + notificationStatus); + }; + } + + // 경고 알림 payload를 만든다. + private NotificationPayload buildThresholdAlert( + String eventId, + LocalDateTime eventDateTime, + UsagePayload usagePayload, + String notificationStatus, + int threshold) { + Map data = + createBaseData(eventId, eventDateTime, usagePayload, notificationStatus); + data.put("threshold", threshold); + + return new NotificationPayload( + usagePayload.familyId(), + usagePayload.customerId(), + NotificationType.THRESHOLD_ALERT, + "데이터 사용량 경고", + "가족 데이터 잔여량이 " + threshold + "% 이하입니다.", + data); + } + + // 차단 알림 payload를 만든다. + private NotificationPayload buildBlockedAlert( + String eventId, + LocalDateTime eventDateTime, + UsagePayload usagePayload, + String notificationStatus) { + Map data = + createBaseData(eventId, eventDateTime, usagePayload, notificationStatus); + data.put("reason", notificationStatus); + + String message = + switch (notificationStatus) { + case "MANUAL" -> "현재 데이터 사용이 관리자 설정으로 차단되었습니다."; + case "APP_BLOCK" -> buildAppBlockMessage(usagePayload.appId()); + case "TIME_BLOCK" -> "현재 시간에는 데이터 사용이 제한됩니다."; + case "MONTHLY_LIMIT_EXCEEDED" -> "개인 월 사용량 한도를 초과했습니다."; + case "FAMILY_QUOTA_EXCEEDED" -> "가족 데이터 사용량을 모두 소진했습니다."; + default -> "현재 데이터 사용이 차단되었습니다."; + }; + + return new NotificationPayload( + usagePayload.familyId(), + usagePayload.customerId(), + NotificationType.BLOCKED, + "데이터 사용 차단", + message, + data); + } + + // 앱 차단 알림 문구에 앱 정보를 함께 넣는다. + private String buildAppBlockMessage(String appId) { + if (appId == null || appId.isBlank()) { + return "현재 앱 사용이 차단되어 있습니다."; + } + return "현재 앱 사용이 차단되어 있습니다. 대상 앱: " + appId; + } + + // 공통 추적 정보를 payload data에 담는다. + private Map createBaseData( + String eventId, + LocalDateTime eventDateTime, + UsagePayload usagePayload, + String notificationStatus) { + Map data = new HashMap<>(); + data.put("originEventId", eventId); + data.put("eventTime", eventDateTime.toString()); + data.put("status", notificationStatus); + data.put("familyId", usagePayload.familyId()); + data.put("customerId", usagePayload.customerId()); + data.put("appId", usagePayload.appId()); + data.put("bytesUsed", usagePayload.bytesUsed()); + return data; + } +} diff --git a/src/main/java/com/project/domain/usage/service/helper/UsageNotificationPublisher.java b/src/main/java/com/project/domain/usage/service/helper/UsageNotificationPublisher.java new file mode 100644 index 0000000..f6be4bd --- /dev/null +++ b/src/main/java/com/project/domain/usage/service/helper/UsageNotificationPublisher.java @@ -0,0 +1,31 @@ +package com.project.domain.usage.service.helper; + +import java.util.concurrent.CompletableFuture; + +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.SendResult; +import org.springframework.stereotype.Component; + +import com.dabom.messaging.kafka.contract.KafkaTopics; +import com.dabom.messaging.kafka.event.KafkaEventMessageSupport; +import com.dabom.messaging.kafka.event.dto.EventEnvelope; +import com.dabom.messaging.kafka.event.dto.notification.NotificationEventSupport; +import com.dabom.messaging.kafka.event.dto.notification.NotificationPayload; + +import lombok.RequiredArgsConstructor; + +@Component +@RequiredArgsConstructor +public class UsageNotificationPublisher { + + private final KafkaTemplate kafkaTemplate; + private final KafkaEventMessageSupport kafkaEventMessageSupport; + + // notification을 비동기로 발행하고 broker ack future를 반환한다. + public CompletableFuture> publishAsync(NotificationPayload payload) { + EventEnvelope envelope = NotificationEventSupport.toEnvelope(payload); + String serialized = kafkaEventMessageSupport.serialize(envelope); + return kafkaTemplate.send( + KafkaTopics.NOTIFICATION, String.valueOf(payload.customerId()), serialized); + } +} diff --git a/src/main/java/com/project/domain/usage/service/helper/UsagePersistEventValidator.java b/src/main/java/com/project/domain/usage/service/helper/UsagePersistEventValidator.java index 35f8670..4c4d771 100644 --- a/src/main/java/com/project/domain/usage/service/helper/UsagePersistEventValidator.java +++ b/src/main/java/com/project/domain/usage/service/helper/UsagePersistEventValidator.java @@ -2,8 +2,8 @@ import org.springframework.stereotype.Component; -import com.dabom.messaging.kafka.event.dto.usage.UsagePersistPayload; import com.project.domain.usage.enums.UsagePersistProcessResult; +import com.project.domain.usage.service.dto.UsagePersistPayload; import com.project.global.util.LogSanitizer; import lombok.RequiredArgsConstructor; @@ -18,13 +18,13 @@ public class UsagePersistEventValidator { public boolean isValidPayload(UsagePersistPayload payload, String eventId, String recordKey) { if (payload == null) { - log.warn("usage-persist payload is null. recordKey={}", recordKey); + log.warn("usage payload is null. recordKey={}", recordKey); return false; } if (payload.originEventId() == null || payload.originEventId().isBlank()) { log.warn( - "usage-persist originEventId is empty. eventId={}, familyId={}, customerId={}", + "usage originEventId is empty. eventId={}, familyId={}, customerId={}", logSanitizer.sanitize(eventId), payload.familyId(), payload.customerId()); @@ -33,7 +33,7 @@ public boolean isValidPayload(UsagePersistPayload payload, String eventId, Strin if (payload.familyId() == null || payload.customerId() == null) { log.warn( - "usage-persist family/customer is invalid. eventId={}, originEventId={}," + "usage family/customer is invalid. eventId={}, originEventId={}," + " familyId={}, customerId={}", logSanitizer.sanitize(eventId), logSanitizer.sanitize(payload.originEventId()), @@ -44,8 +44,7 @@ public boolean isValidPayload(UsagePersistPayload payload, String eventId, Strin if (payload.bytesUsed() == null || payload.bytesUsed() <= 0) { log.warn( - "usage-persist bytesUsed is invalid. eventId={}, originEventId={}," - + " bytesUsed={}", + "usage bytesUsed is invalid. eventId={}, originEventId={}," + " bytesUsed={}", logSanitizer.sanitize(eventId), logSanitizer.sanitize(payload.originEventId()), payload.bytesUsed()); @@ -56,7 +55,7 @@ public boolean isValidPayload(UsagePersistPayload payload, String eventId, Strin UsagePersistProcessResult.from(payload.processResult()); } catch (IllegalArgumentException e) { log.warn( - "usage-persist processResult is invalid. eventId={}, originEventId={}," + "usage processResult is invalid. eventId={}, originEventId={}," + " processResult={}", logSanitizer.sanitize(eventId), logSanitizer.sanitize(payload.originEventId()), diff --git a/src/main/java/com/project/domain/usage/service/helper/UsageProcessingDecisionMapper.java b/src/main/java/com/project/domain/usage/service/helper/UsageProcessingDecisionMapper.java new file mode 100644 index 0000000..098ab14 --- /dev/null +++ b/src/main/java/com/project/domain/usage/service/helper/UsageProcessingDecisionMapper.java @@ -0,0 +1,40 @@ +package com.project.domain.usage.service.helper; + +import org.springframework.stereotype.Component; + +import com.dabom.messaging.kafka.error.NonRetryableKafkaMessageProcessingException; +import com.project.domain.usage.enums.UsagePersistProcessResult; + +@Component +public class UsageProcessingDecisionMapper { + + // Lua 상태 문자열을 DB 정산/알림 판단용 결정으로 변환한다. + public UsageProcessingDecision fromLuaStatus(String rawStatus) { + if (rawStatus == null || rawStatus.isBlank()) { + throw new NonRetryableKafkaMessageProcessingException("Lua status is null or blank"); + } + + // 허용/경고/차단 상태를 명시적으로만 해석한다. + return switch (rawStatus) { + case "NORMAL" -> + new UsageProcessingDecision( + UsagePersistProcessResult.ALLOWED.name(), false, rawStatus); + case "WARNING_50", "WARNING_30", "WARNING_10" -> + new UsageProcessingDecision( + UsagePersistProcessResult.ALLOWED.name(), true, rawStatus); + case "MANUAL", + "APP_BLOCK", + "TIME_BLOCK", + "MONTHLY_LIMIT_EXCEEDED", + "FAMILY_QUOTA_EXCEEDED" -> + new UsageProcessingDecision(rawStatus, true, rawStatus); + default -> + throw new NonRetryableKafkaMessageProcessingException( + "Unsupported Lua status: " + rawStatus); + }; + } + + // usage 처리 결과를 각 후속 단계가 공통으로 참조하는 결정 객체다. + public record UsageProcessingDecision( + String persistProcessResult, boolean publishNotification, String notificationStatus) {} +} diff --git a/src/main/java/com/project/domain/usage/service/helper/UsageRecordWriter.java b/src/main/java/com/project/domain/usage/service/helper/UsageRecordWriter.java index 464c3d0..de110e8 100644 --- a/src/main/java/com/project/domain/usage/service/helper/UsageRecordWriter.java +++ b/src/main/java/com/project/domain/usage/service/helper/UsageRecordWriter.java @@ -5,8 +5,8 @@ import org.springframework.stereotype.Service; -import com.dabom.messaging.kafka.event.dto.usage.UsagePersistPayload; import com.project.domain.usage.repository.UsageRecordRepository; +import com.project.domain.usage.service.dto.UsagePersistPayload; import com.project.global.common.TimeConstants; import com.project.global.util.LogSanitizer; diff --git a/src/main/java/com/project/global/util/RedisKeyGenerator.java b/src/main/java/com/project/global/util/RedisKeyGenerator.java index 5fa7651..29c3b79 100644 --- a/src/main/java/com/project/global/util/RedisKeyGenerator.java +++ b/src/main/java/com/project/global/util/RedisKeyGenerator.java @@ -9,19 +9,19 @@ public class RedisKeyGenerator { private static final String KEY_SEPARATOR = ":"; + private static final String ALERT_SEGMENT = "alert"; private static final String FAMILY_KEY_PREFIX = "family"; private static final String POLICY_EVENT_DEDUP_KEY_PREFIX = "event:dedup:policy"; private static final String USAGE_EVENT_DEDUP_KEY_PREFIX = "event:dedup:usage"; private static final DateTimeFormatter MONTH_SUFFIX_FORMATTER = DateTimeFormatter.ofPattern("yyyyMM"); - // 가족 알림 상태 키 - public String generateFamilyAlertKey(Long familyId, int threshold, LocalDate eventMonth) { - return FAMILY_KEY_PREFIX - + KEY_SEPARATOR - + familyId + // 개인 기준 경고 알림 키를 만든다. + public String generateFamilyCustomerThresholdAlertKey( + Long familyId, Long customerId, int threshold, LocalDate eventMonth) { + return familyCustomerPrefix(familyId, customerId) + KEY_SEPARATOR - + "alert" + + ALERT_SEGMENT + KEY_SEPARATOR + "THRESHOLD" + KEY_SEPARATOR @@ -30,7 +30,38 @@ public String generateFamilyAlertKey(Long familyId, int threshold, LocalDate eve + formatMonth(eventMonth); } - // 가족 quota 정보 키 + // 개인 기준 차단 알림 키를 만든다. + public String generateFamilyCustomerBlockAlertKey( + Long familyId, Long customerId, String alertType, LocalDate eventMonth) { + return familyCustomerPrefix(familyId, customerId) + + KEY_SEPARATOR + + ALERT_SEGMENT + + KEY_SEPARATOR + + alertType + + KEY_SEPARATOR + + formatMonth(eventMonth); + } + + // 앱 차단 알림은 앱 단위로 분리한다. + public String generateFamilyCustomerAppBlockAlertKey( + Long familyId, Long customerId, String appId, LocalDate eventMonth) { + return familyCustomerPrefix(familyId, customerId) + + KEY_SEPARATOR + + ALERT_SEGMENT + + KEY_SEPARATOR + + "APP_BLOCK" + + KEY_SEPARATOR + + appId + + KEY_SEPARATOR + + formatMonth(eventMonth); + } + + // 가족 구성원 set 키를 만든다. + public String generateFamilyMembersKey(Long familyId) { + return FAMILY_KEY_PREFIX + KEY_SEPARATOR + familyId + KEY_SEPARATOR + "members"; + } + + // 가족 quota 정보 키를 만든다. public String generateFamilyInfoKey(Long familyId, LocalDate eventMonth) { return FAMILY_KEY_PREFIX + KEY_SEPARATOR @@ -41,7 +72,7 @@ public String generateFamilyInfoKey(Long familyId, LocalDate eventMonth) { + formatMonth(eventMonth); } - // 가족 잔여 데이터 키 + // 가족 남은 사용량 키를 만든다. public String generateFamilyRemainingKey(Long familyId, LocalDate eventMonth) { return FAMILY_KEY_PREFIX + KEY_SEPARATOR @@ -52,16 +83,10 @@ public String generateFamilyRemainingKey(Long familyId, LocalDate eventMonth) { + formatMonth(eventMonth); } - // 고객 월별 사용량 키 + // 개인 월 사용량 키를 만든다. public String generateFamilyCustomerMonthlyUsageKey( Long familyId, Long customerId, LocalDate eventMonth) { - return FAMILY_KEY_PREFIX - + KEY_SEPARATOR - + familyId - + KEY_SEPARATOR - + "customer" - + KEY_SEPARATOR - + customerId + return familyCustomerPrefix(familyId, customerId) + KEY_SEPARATOR + "usage" + KEY_SEPARATOR @@ -70,29 +95,33 @@ public String generateFamilyCustomerMonthlyUsageKey( + formatMonth(eventMonth); } - // 고객 정책 제약 키 + // 개인 제약 조건 키를 만든다. public String generateFamilyCustomerConstraintsKey(Long familyId, Long customerId) { - return FAMILY_KEY_PREFIX - + KEY_SEPARATOR - + familyId - + KEY_SEPARATOR - + "customer" - + KEY_SEPARATOR - + customerId - + KEY_SEPARATOR - + "constraints"; + return familyCustomerPrefix(familyId, customerId) + KEY_SEPARATOR + "constraints"; } - // policy 이벤트 dedup 키 + // policy 이벤트 dedup 키를 만든다. public String generatePolicyEventDedupKey(String eventId, Long customerId) { return POLICY_EVENT_DEDUP_KEY_PREFIX + KEY_SEPARATOR + eventId + KEY_SEPARATOR + customerId; } - // usage-event dedup 키 + // usage 이벤트 dedup 키를 만든다. public String generateUsageEventDedupKey(String eventId) { return USAGE_EVENT_DEDUP_KEY_PREFIX + KEY_SEPARATOR + eventId; } + // family-customer 공통 prefix를 만든다. + private String familyCustomerPrefix(Long familyId, Long customerId) { + return FAMILY_KEY_PREFIX + + KEY_SEPARATOR + + familyId + + KEY_SEPARATOR + + "customer" + + KEY_SEPARATOR + + customerId; + } + + // 월 suffix를 공통 형식으로 만든다. private String formatMonth(LocalDate eventMonth) { return eventMonth.format(MONTH_SUFFIX_FORMATTER); } diff --git a/src/main/resources/lua/usage_update.lua b/src/main/resources/lua/usage_update.lua index f7d7364..b26218e 100644 --- a/src/main/resources/lua/usage_update.lua +++ b/src/main/resources/lua/usage_update.lua @@ -2,26 +2,62 @@ -- KEYS[2]: family:{fid}:remaining:{yyyyMM} -- KEYS[3]: family:{fid}:customer:{uid}:usage:monthly:{yyyyMM} -- KEYS[4]: family:{fid}:customer:{uid}:constraints --- KEYS[5]: family:{fid}:alert:THRESHOLD:50:{yyyyMM} --- KEYS[6]: family:{fid}:alert:THRESHOLD:30:{yyyyMM} --- KEYS[7]: family:{fid}:alert:THRESHOLD:10:{yyyyMM} --- KEYS[8]: event:dedup:usage:{eventId} +-- KEYS[5]: family:{fid}:customer:{uid}:alert:THRESHOLD:50:{yyyyMM} +-- KEYS[6]: family:{fid}:customer:{uid}:alert:THRESHOLD:30:{yyyyMM} +-- KEYS[7]: family:{fid}:customer:{uid}:alert:THRESHOLD:10:{yyyyMM} +-- KEYS[8]: family:{fid}:customer:{uid}:alert:MANUAL:{yyyyMM} +-- KEYS[9]: family:{fid}:customer:{uid}:alert:APP_BLOCK:{appId}:{yyyyMM} +-- KEYS[10]: family:{fid}:customer:{uid}:alert:TIME_BLOCK:{yyyyMM} +-- KEYS[11]: family:{fid}:customer:{uid}:alert:MONTHLY_LIMIT_EXCEEDED:{yyyyMM} +-- KEYS[12]: family:{fid}:customer:{uid}:alert:FAMILY_QUOTA_EXCEEDED:{yyyyMM} +-- KEYS[13]: event:dedup:usage:{eventId} -- ARGV[1]: usageBytes -- ARGV[2]: currentHHmm (e.g. 2230) -- ARGV[3]: normalizedAppId -- ARGV[4]: dedupTtlSeconds +local STATUS_NORMAL = 'NORMAL' +local STATUS_MANUAL = 'MANUAL' +local STATUS_APP_BLOCK = 'APP_BLOCK' +local STATUS_TIME_BLOCK = 'TIME_BLOCK' +local STATUS_MONTHLY_LIMIT_EXCEEDED = 'MONTHLY_LIMIT_EXCEEDED' +local STATUS_FAMILY_QUOTA_EXCEEDED = 'FAMILY_QUOTA_EXCEEDED' +local STATUS_WARNING_50 = 'WARNING_50' +local STATUS_WARNING_30 = 'WARNING_30' +local STATUS_WARNING_10 = 'WARNING_10' +local STATUS_DUPLICATE = 'DUPLICATE' + +local CONSTRAINT_BLOCK_ACCESS = 'BLOCK:ACCESS' +local CONSTRAINT_BLOCK_TIME = 'BLOCK:TIME' +local CONSTRAINT_LIMIT_DATA_MONTHLY = 'LIMIT:DATA:MONTHLY' +local CONSTRAINT_BLOCK_APP_PREFIX = 'BLOCK:APP:' + +local ALERT_PUBLISHED = 'PUBLISHED' + local usageBytes = tonumber(ARGV[1]) local currentHHmm = tonumber(ARGV[2] or '0') local appId = ARGV[3] or '' local dedupTtlSeconds = tonumber(ARGV[4] or '0') local monthlyLimit = -1 --- 공통 반환 형식 -local function getResult(status, currentMonthlyUsed, duplicate) +-- 잘못된 입력은 Redis를 건드리기 전에 즉시 중단한다. +if not usageBytes or usageBytes <= 0 then + return redis.error_reply('usageBytes must be a positive number') +end + +if not currentHHmm then + return redis.error_reply('currentHHmm must be a number') +end + +-- Lua 반환 형식에 맞는 결과 배열을 만든다. +local function build_result(status, currentMonthlyUsed, duplicate, currentRemaining, shouldNotify) local totalLimit = tonumber(redis.call('HGET', KEYS[1], 'totalQuota') or '0') - local currentRemaining = tonumber(redis.call('GET', KEYS[2]) or totalLimit) - local totalUsed = totalLimit - currentRemaining + local remaining = currentRemaining + if remaining == nil then + remaining = tonumber(redis.call('GET', KEYS[2]) or totalLimit) + end + + local totalUsed = totalLimit - remaining local userRatio = 0 if totalLimit > 0 then userRatio = currentMonthlyUsed / totalLimit @@ -29,132 +65,239 @@ local function getResult(status, currentMonthlyUsed, duplicate) return { totalUsed, - currentRemaining, + remaining, status, currentMonthlyUsed, userRatio, monthlyLimit, + shouldNotify and 1 or 0, duplicate and 1 or 0 } end --- 1) 동일 eventId 재처리 방지 -if dedupTtlSeconds > 0 then - -- 같은 eventId는 월별 상태 반영 전에 바로 차단함 - local firstSeen = redis.call('SET', KEYS[8], '1', 'NX', 'EX', dedupTtlSeconds) - if not firstSeen then - return {0, 0, 'DUPLICATE', 0, 0, -1, 1} +-- dedup 캐시 문자열을 Lua 결과 형식으로 복원한다. +local function decode_cached_result(raw) + if not raw then + return nil + end + + local parts = {} + for token in string.gmatch(raw, "[^|]+") do + table.insert(parts, token) + end + + if #parts ~= 7 then + return nil end + + return { + tonumber(parts[1]) or 0, + tonumber(parts[2]) or 0, + parts[3] or STATUS_DUPLICATE, + tonumber(parts[4]) or 0, + tonumber(parts[5]) or 0, + tonumber(parts[6]) or -1, + tonumber(parts[7]) or 0, + 1 + } end --- 2) 고객별 제약 조회 -local constraintsArray = redis.call('HGETALL', KEYS[4]) -local constraints = {} -for i = 1, #constraintsArray, 2 do - constraints[constraintsArray[i]] = constraintsArray[i + 1] +-- Lua 결과를 dedup 캐시 문자열로 직렬화한다. +local function encode_cached_result(result) + return table.concat({ + tostring(result[1]), + tostring(result[2]), + tostring(result[3]), + tostring(result[4]), + tostring(result[5]), + tostring(result[6]), + tostring(result[7]) + }, '|') end -local monthlyLimitStr = constraints['LIMIT:DATA:MONTHLY'] -if monthlyLimitStr then - monthlyLimit = tonumber(monthlyLimitStr) +-- dedup TTL이 켜져 있으면 현재 결과를 캐시에 보관한다. +local function cache_result(result) + if dedupTtlSeconds <= 0 then + return + end + + redis.call('SET', KEYS[13], encode_cached_result(result), 'EX', dedupTtlSeconds) end -local currentMonthly = tonumber(redis.call('GET', KEYS[3]) or '0') +-- 차단 또는 제한으로 종료하는 결과를 캐시에 넣고 반환한다. +local function finalize(status, currentMonthlyUsed, currentRemaining, shouldNotify) + local result = build_result(status, currentMonthlyUsed, false, currentRemaining, shouldNotify) + cache_result(result) + return result +end --- 3) 차단/제한 조건 확인 -if constraints['BLOCK:ACCESS'] == '1' then - return getResult('MANUAL', currentMonthly, false) +-- 알림 key가 이미 있으면 재발행을 막고, 없으면 이번 달 최초 알림으로 기록한다. +local function consume_alert(key) + local isSent = redis.call('EXISTS', key) + if isSent == 1 then + return false + end + + redis.call('SET', key, ALERT_PUBLISHED) + return true end -if appId ~= '' and constraints['BLOCK:APP:' .. appId] == '1' then - return getResult('APP_BLOCK', currentMonthly, false) +-- 제약 조건 해시를 Lua 테이블로 읽어온다. +local function load_constraints() + local constraintsArray = redis.call('HGETALL', KEYS[4]) + local constraints = {} + for i = 1, #constraintsArray, 2 do + constraints[constraintsArray[i]] = constraintsArray[i + 1] + end + return constraints end -local blockStart = nil -local blockEnd = nil +-- 월 한도 값을 읽고 없으면 제한 없음으로 본다. +local function resolve_monthly_limit(constraints) + local monthlyLimitStr = constraints[CONSTRAINT_LIMIT_DATA_MONTHLY] + if monthlyLimitStr then + return tonumber(monthlyLimitStr) + end + return -1 +end -local blockTimeRange = constraints['BLOCK:TIME'] -if blockTimeRange then - local dashPos = string.find(blockTimeRange, '-', 1, true) - if dashPos then - local startStr = string.sub(blockTimeRange, 1, dashPos - 1) - local endStr = string.sub(blockTimeRange, dashPos + 1) - blockStart = tonumber(startStr) - blockEnd = tonumber(endStr) +-- 현재 가족 잔여량을 읽고 없으면 totalQuota로 복원한다. +local function resolve_current_remaining() + local currentRemaining = tonumber(redis.call('GET', KEYS[2])) + if currentRemaining ~= nil then + return currentRemaining end + + local totalLimit = tonumber(redis.call('HGET', KEYS[1], 'totalQuota') or '0') + return totalLimit end -if blockStart and blockEnd then +-- 시간 차단 규칙과 현재 시각의 충돌 여부를 확인한다. +local function resolve_time_block_status(blockTimeRange, hhmm) + if not blockTimeRange then + return nil + end + + local dashPos = string.find(blockTimeRange, '-', 1, true) + if not dashPos then + return nil + end + + local startStr = string.sub(blockTimeRange, 1, dashPos - 1) + local endStr = string.sub(blockTimeRange, dashPos + 1) + local blockStart = tonumber(startStr) + local blockEnd = tonumber(endStr) + + if not blockStart or not blockEnd then + return nil + end + if blockStart < blockEnd then - if currentHHmm >= blockStart and currentHHmm < blockEnd then - return getResult('TIME_BLOCK', currentMonthly, false) + if hhmm >= blockStart and hhmm < blockEnd then + return STATUS_TIME_BLOCK end - elseif blockStart > blockEnd then - if currentHHmm >= blockStart or currentHHmm < blockEnd then - return getResult('TIME_BLOCK', currentMonthly, false) + return nil + end + + if blockStart > blockEnd then + if hhmm >= blockStart or hhmm < blockEnd then + return STATUS_TIME_BLOCK end - else - return getResult('TIME_BLOCK', currentMonthly, false) + return nil end + + return STATUS_TIME_BLOCK end -if monthlyLimit ~= -1 then - if (currentMonthly + usageBytes) > monthlyLimit then - return getResult('MONTHLY_LIMIT_EXCEEDED', currentMonthly, false) +-- 차단 또는 제한 조건을 우선순서대로 평가해 즉시 종료 상태를 결정한다. +local function resolve_block_status( + constraints, + normalizedAppId, + hhmm, + currentMonthlyUsed, + requestBytes, + currentRemaining +) + if constraints[CONSTRAINT_BLOCK_ACCESS] == '1' then + return STATUS_MANUAL, KEYS[8] end -end -local currentRemaining = tonumber(redis.call('GET', KEYS[2])) -if currentRemaining == nil then - -- 월초 첫 이벤트면 remaining이 아직 없을 수 있어서 totalQuota로 시작함 - local totalLimit = tonumber(redis.call('HGET', KEYS[1], 'totalQuota') or '0') - currentRemaining = totalLimit -end + if normalizedAppId ~= '' and constraints[CONSTRAINT_BLOCK_APP_PREFIX .. normalizedAppId] == '1' then + return STATUS_APP_BLOCK, KEYS[9] + end -if currentRemaining < usageBytes then - return getResult('FAMILY_QUOTA_EXCEEDED', currentMonthly, false) -end + local timeBlockStatus = resolve_time_block_status(constraints[CONSTRAINT_BLOCK_TIME], hhmm) + if timeBlockStatus then + return timeBlockStatus, KEYS[10] + end --- 4) 사용량 반영 -local newRemaining = redis.call('DECRBY', KEYS[2], usageBytes) -local newMonthly = redis.call('INCRBY', KEYS[3], usageBytes) + if monthlyLimit ~= -1 and (currentMonthlyUsed + requestBytes) > monthlyLimit then + return STATUS_MONTHLY_LIMIT_EXCEEDED, KEYS[11] + end --- 5) 후속 상태 계산 -local limitStr = redis.call('HGET', KEYS[1], 'totalQuota') -local totalLimit = tonumber(limitStr or '0') -local status = 'NORMAL' -local ratio = 0 + if currentRemaining < requestBytes then + return STATUS_FAMILY_QUOTA_EXCEEDED, KEYS[12] + end -if totalLimit > 0 then - ratio = newRemaining / totalLimit + return nil, nil end -local alertKey = nil -if ratio < 0.1 then - alertKey = KEYS[7] - status = 'WARNING_10' -elseif ratio < 0.3 then - alertKey = KEYS[6] - status = 'WARNING_30' -elseif ratio < 0.5 then - alertKey = KEYS[5] - status = 'WARNING_50' +-- Redis 반영 후 경고 상태와 알림 dedup 상태를 계산한다. +local function resolve_alert_status(newRemaining, newMonthly) + local totalLimit = tonumber(redis.call('HGET', KEYS[1], 'totalQuota') or '0') + if totalLimit <= 0 then + return build_result(STATUS_NORMAL, newMonthly, false, newRemaining, false) + end + + local ratio = newRemaining / totalLimit + local alertKey = nil + local status = STATUS_NORMAL + + if ratio < 0.1 then + alertKey = KEYS[7] + status = STATUS_WARNING_10 + elseif ratio < 0.3 then + alertKey = KEYS[6] + status = STATUS_WARNING_30 + elseif ratio < 0.5 then + alertKey = KEYS[5] + status = STATUS_WARNING_50 + end + + if not alertKey then + return build_result(status, newMonthly, false, newRemaining, false) + end + + return build_result(status, newMonthly, false, newRemaining, consume_alert(alertKey)) end -if alertKey then - -- 같은 월 같은 임계치는 suffix key 존재 여부로 한 번만 발행함 - local isSent = redis.call('EXISTS', alertKey) - if isSent == 1 then - status = 'NORMAL' - else - redis.call('SET', alertKey, 'PUBLISHED') +-- 같은 eventId 결과가 캐시에 남아 있으면 그대로 재사용한다. +if dedupTtlSeconds > 0 then + local cached = decode_cached_result(redis.call('GET', KEYS[13])) + if cached then + return cached end end -local totalUsed = totalLimit - newRemaining -local userRatio = 0 -if totalLimit > 0 then - userRatio = newMonthly / totalLimit +local constraints = load_constraints() +monthlyLimit = resolve_monthly_limit(constraints) + +local currentMonthly = tonumber(redis.call('GET', KEYS[3]) or '0') +local currentRemaining = resolve_current_remaining() + +-- 차단 또는 제한 조건이면 Redis 증감 없이 현재 상태만 반환한다. +local blockedStatus, blockedAlertKey = + resolve_block_status( + constraints, appId, currentHHmm, currentMonthly, usageBytes, currentRemaining) +if blockedStatus then + return finalize(blockedStatus, currentMonthly, currentRemaining, consume_alert(blockedAlertKey)) end -return {totalUsed, newRemaining, status, newMonthly, userRatio, monthlyLimit, 0} +-- 허용인 경우에만 Redis 사용량을 실제로 반영한다. +local newRemaining = redis.call('DECRBY', KEYS[2], usageBytes) +local newMonthly = redis.call('INCRBY', KEYS[3], usageBytes) + +-- 반영 후 경고 상태를 계산하고 결과를 캐시해 둔다. +local finalResult = resolve_alert_status(newRemaining, newMonthly) +cache_result(finalResult) +return finalResult diff --git a/src/test/java/com/project/domain/policy/service/PolicyConstraintSyncServiceImplTest.java b/src/test/java/com/project/domain/policy/service/PolicyConstraintSyncServiceImplTest.java index c1e268e..9a6cd61 100644 --- a/src/test/java/com/project/domain/policy/service/PolicyConstraintSyncServiceImplTest.java +++ b/src/test/java/com/project/domain/policy/service/PolicyConstraintSyncServiceImplTest.java @@ -69,11 +69,7 @@ void sync_SkipWhenConstraintsKeyMissing() { 10L, 20L, PolicyConstraintKeyConstants.LIMIT_DATA_MONTHLY, "1024", true); EventEnvelope envelope = new EventEnvelope<>( - "evt-1", - KafkaEventTypes.POLICY_UPDATED, - null, - LocalDateTime.now(), - payload); + "evt-1", KafkaEventTypes.POLICY_UPDATED, LocalDateTime.now(), payload); given(policyEventValidator.isValidPayload(payload, "evt-1", "record-1")).willReturn(true); given( @@ -105,11 +101,7 @@ void sync_ApplyWhenConstraintsKeyExists() { 10L, 20L, PolicyConstraintKeyConstants.LIMIT_DATA_MONTHLY, "1024", true); EventEnvelope envelope = new EventEnvelope<>( - "evt-2", - KafkaEventTypes.POLICY_UPDATED, - null, - LocalDateTime.now(), - payload); + "evt-2", KafkaEventTypes.POLICY_UPDATED, LocalDateTime.now(), payload); given(policyEventValidator.isValidPayload(payload, "evt-2", "record-2")).willReturn(true); given( @@ -156,11 +148,7 @@ void sync_BlockAppSkipWhenConstraintsKeyMissing() { true); EventEnvelope envelope = new EventEnvelope<>( - "evt-3", - KafkaEventTypes.POLICY_UPDATED, - null, - LocalDateTime.now(), - payload); + "evt-3", KafkaEventTypes.POLICY_UPDATED, LocalDateTime.now(), payload); given(policyEventValidator.isValidPayload(payload, "evt-3", "record-3")).willReturn(true); given(policyEventValidator.isAllowedPolicyKey(PolicyConstraintKeyConstants.BLOCK_APP)) diff --git a/src/test/java/com/project/domain/usage/service/UsageEventValidatorTest.java b/src/test/java/com/project/domain/usage/service/UsageEventValidatorTest.java index 92498d2..dd82b4a 100644 --- a/src/test/java/com/project/domain/usage/service/UsageEventValidatorTest.java +++ b/src/test/java/com/project/domain/usage/service/UsageEventValidatorTest.java @@ -15,44 +15,42 @@ class UsageEventValidatorTest { private final UsageEventValidator validator = new UsageEventValidator(); @Test - @DisplayName("유효한 페이로드는 검증을 통과해야 한다") + @DisplayName("유효한 usage payload는 검증을 통과한다") void validPayload() { - // given UsagePayload payload = new UsagePayload(100L, 1L, "com.app.test", 1024L, Map.of()); - // when boolean result = validator.isValid(payload, "evt_1"); - // then assertThat(result).isTrue(); } @Test - @DisplayName("필수 값이 누락되면 검증 실패해야 한다") + @DisplayName("필수 값이 없으면 검증에 실패한다") void invalidPayload() { - // given UsagePayload payload = new UsagePayload(null, null, null, null, null); - // when boolean result = validator.isValid(payload, "evt_1"); - // then assertThat(result).isFalse(); } @Test - @DisplayName("음수 사용량은 검증 실패해야 한다") + @DisplayName("음수 사용량이면 검증에 실패한다") void negativeBytesUsed() { - // given - UsagePayload payload = - new UsagePayload( - 100L, 1L, "appId", -1L, // 음수 - Map.of()); + UsagePayload payload = new UsagePayload(100L, 1L, "appId", -1L, Map.of()); + + boolean result = validator.isValid(payload, "evt_1"); + + assertThat(result).isFalse(); + } + + @Test + @DisplayName("0 사용량이면 검증에 실패한다") + void zeroBytesUsed() { + UsagePayload payload = new UsagePayload(100L, 1L, "appId", 0L, Map.of()); - // when boolean result = validator.isValid(payload, "evt_1"); - // then assertThat(result).isFalse(); } } diff --git a/src/test/java/com/project/domain/usage/service/UsagePersistServiceImplTest.java b/src/test/java/com/project/domain/usage/service/UsagePersistServiceImplTest.java index 4f54000..fb96ddd 100644 --- a/src/test/java/com/project/domain/usage/service/UsagePersistServiceImplTest.java +++ b/src/test/java/com/project/domain/usage/service/UsagePersistServiceImplTest.java @@ -2,14 +2,13 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.nullable; -import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import java.time.LocalDate; -import java.time.LocalDateTime; +import java.util.Map; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; @@ -19,10 +18,9 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import com.dabom.messaging.kafka.contract.KafkaEventTypes; -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.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; @@ -53,51 +51,60 @@ void setUp() { } @Test - @DisplayName("허용 이벤트면 eventMonth 기준으로 quota와 family_quota를 함께 갱신한다") - void persist_AllowedEvent_UpdatesQuotaAndFamilyByEventMonth() { - UsagePersistPayload payload = - new UsagePersistPayload( - "origin_1", 100L, 1L, 2048L, "app", "ALLOWED", "2026-03-15T01:02:03"); - EventEnvelope envelope = - new EventEnvelope<>( - "evt_1", KafkaEventTypes.USAGE_PERSIST, null, LocalDateTime.now(), payload); + @DisplayName("ALLOWED 상태면 customer_quota와 family_quota를 함께 반영한다") + void persistFromUsageEvent_AllowedStatus_PersistsAsAllowed() { + String eventId = "evt_1"; + String eventTime = "2026-03-15T01:02:03"; + UsagePayload usagePayload = new UsagePayload(100L, 1L, "app", 2048L, Map.of()); LocalDate eventMonth = LocalDate.of(2026, 3, 1); - given(usagePersistEventValidator.isValidPayload(payload, "evt_1", "recordKey")) + org.mockito.BDDMockito.given( + usagePersistEventValidator.isValidPayload( + any(UsagePersistPayload.class), any(), any())) .willReturn(true); - given(familyMemberRepository.existsByFamilyIdAndCustomerIdAndDeletedAtIsNull(100L, 1L)) + org.mockito.BDDMockito.given( + familyMemberRepository.existsByFamilyIdAndCustomerIdAndDeletedAtIsNull( + 100L, 1L)) + .willReturn(true); + org.mockito.BDDMockito.given(usageRecordWriter.persistUsageRecord(any(), any(), any())) .willReturn(true); - given(usageRecordWriter.persistUsageRecord(payload, "evt_1", "origin_1")).willReturn(true); - usagePersistService.persist(envelope, "recordKey"); + usagePersistService.persistFromUsageEvent(eventId, eventTime, usagePayload, "ALLOWED"); verify(customerQuotaWriter, times(1)) - .persistAllowedQuota(payload, eventMonth, "evt_1", "origin_1"); + .persistAllowedQuota( + any(UsagePersistPayload.class), any(LocalDate.class), any(), any()); verify(familyQuotaWriter, times(1)) - .persistAllowedQuota(100L, eventMonth, 2048L, "evt_1", "origin_1"); + .persistAllowedQuota(100L, eventMonth, 2048L, eventId, eventId); verify(customerQuotaWriter, never()).persistBlockedQuota(any(), any(), any(), any(), any()); } @Test - @DisplayName("차단 이벤트면 usage_record와 family_quota 누적 없이 차단 상태만 반영한다") - void persist_BlockedEvent_OnlyPersistsBlockState() { - UsagePersistPayload payload = - new UsagePersistPayload( - "origin_2", 200L, 2L, 1024L, "app", "TIME_BLOCK", "2026-03-20T10:20:30"); - EventEnvelope envelope = - new EventEnvelope<>( - "evt_2", KafkaEventTypes.USAGE_PERSIST, null, LocalDateTime.now(), payload); + @DisplayName("APP_BLOCK 상태면 usage_record 없이 차단 상태만 반영한다") + void persistFromUsageEvent_AppBlock_OnlyPersistsBlockState() { + String eventId = "evt_2"; + String eventTime = "2026-03-20T10:20:30"; + UsagePayload usagePayload = new UsagePayload(200L, 2L, "app", 1024L, Map.of()); LocalDate eventMonth = LocalDate.of(2026, 3, 1); - given(usagePersistEventValidator.isValidPayload(payload, "evt_2", "recordKey")) + org.mockito.BDDMockito.given( + usagePersistEventValidator.isValidPayload( + any(UsagePersistPayload.class), any(), any())) .willReturn(true); - given(familyMemberRepository.existsByFamilyIdAndCustomerIdAndDeletedAtIsNull(200L, 2L)) + org.mockito.BDDMockito.given( + familyMemberRepository.existsByFamilyIdAndCustomerIdAndDeletedAtIsNull( + 200L, 2L)) .willReturn(true); - usagePersistService.persist(envelope, "recordKey"); + usagePersistService.persistFromUsageEvent(eventId, eventTime, usagePayload, "APP_BLOCK"); verify(customerQuotaWriter, times(1)) - .persistBlockedQuota(payload, eventMonth, "evt_2", "origin_2", "TIME_BLOCK"); + .persistBlockedQuota( + any(UsagePersistPayload.class), + org.mockito.ArgumentMatchers.eq(eventMonth), + org.mockito.ArgumentMatchers.eq(eventId), + org.mockito.ArgumentMatchers.eq(eventId), + org.mockito.ArgumentMatchers.eq("APP_BLOCK")); verify(usageRecordWriter, never()).persistUsageRecord(any(), any(), any()); verify(customerQuotaWriter, never()).persistAllowedQuota(any(), any(), any(), any()); verify(familyQuotaWriter, never()).persistAllowedQuota(any(), any(), any(), any(), any()); diff --git a/src/test/java/com/project/domain/usage/service/UsageSyncServiceImplTest.java b/src/test/java/com/project/domain/usage/service/UsageSyncServiceImplTest.java index 22d14d3..01a7878 100644 --- a/src/test/java/com/project/domain/usage/service/UsageSyncServiceImplTest.java +++ b/src/test/java/com/project/domain/usage/service/UsageSyncServiceImplTest.java @@ -1,17 +1,20 @@ package com.project.domain.usage.service; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.nullable; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import java.time.LocalDate; import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; @@ -26,12 +29,20 @@ import com.dabom.messaging.kafka.contract.KafkaConsumerGroups; import com.dabom.messaging.kafka.contract.KafkaEventTypes; import com.dabom.messaging.kafka.contract.KafkaTopics; +import com.dabom.messaging.kafka.error.KafkaMessageProcessingException; +import com.dabom.messaging.kafka.error.NonRetryableKafkaMessageProcessingException; +import com.dabom.messaging.kafka.event.dto.notification.NotificationPayload; +import com.dabom.messaging.kafka.event.dto.notification.NotificationType; import com.dabom.messaging.kafka.event.dto.usage.UsagePayload; import com.dabom.messaging.kafka.metrics.KafkaMetrics; import com.project.domain.policy.service.helper.PolicyConstraintWarmupHelper; import com.project.domain.usage.service.dto.UsageUpdateResult; -import com.project.domain.usage.service.helper.UsageEventPublisher; +import com.project.domain.usage.service.helper.UsageEventOutboxService; +import com.project.domain.usage.service.helper.UsageFamilyMembershipCacheHelper; import com.project.domain.usage.service.helper.UsageLuaExecutor; +import com.project.domain.usage.service.helper.UsageNotificationPayloadMapper; +import com.project.domain.usage.service.helper.UsageNotificationPublisher; +import com.project.domain.usage.service.helper.UsageProcessingDecisionMapper; import com.project.domain.usage.service.helper.UsageRedisWarmupHelper; import com.project.global.util.LogSanitizer; import com.project.global.util.RedisKeyGenerator; @@ -45,7 +56,12 @@ class UsageSyncServiceImplTest { @Mock private UsageRedisWarmupHelper usageRedisWarmupHelper; @Mock private PolicyConstraintWarmupHelper policyConstraintWarmupHelper; @Mock private UsageLuaExecutor usageLuaExecutor; - @Mock private UsageEventPublisher usageEventPublisher; + @Mock private UsagePersistService usagePersistService; + @Mock private UsageEventOutboxService usageEventOutboxService; + @Mock private UsageProcessingDecisionMapper usageProcessingDecisionMapper; + @Mock private UsageNotificationPayloadMapper usageNotificationPayloadMapper; + @Mock private UsageNotificationPublisher usageNotificationPublisher; + @Mock private UsageFamilyMembershipCacheHelper usageFamilyMembershipCacheHelper; @Mock private LogSanitizer logSanitizer; @Mock private KafkaMetrics kafkaMetrics; @@ -55,141 +71,288 @@ void setUp() { lenient() .when(logSanitizer.sanitize(nullable(String.class))) .thenAnswer( - invocation -> { - String raw = invocation.getArgument(0); - return raw == null ? "null" : raw; - }); + invocation -> + invocation.getArgument(0) == null + ? "null" + : invocation.getArgument(0)); } @Test - @DisplayName("정상 흐름이면 Lua 실행 후 이벤트 발행기로 위임한다") + @DisplayName("정상 이벤트면 DB 정산 후 notification을 비동기로 발행한다") void syncUsage_SuccessFlow() { String eventId = "evt_1"; String eventTime = "2026-03-04T12:34:56"; LocalDate eventMonth = LocalDate.of(2026, 3, 1); UsagePayload payload = new UsagePayload(100L, 1L, "appId", 1024L, Map.of()); + UsageProcessingDecisionMapper.UsageProcessingDecision decision = + new UsageProcessingDecisionMapper.UsageProcessingDecision( + "ALLOWED", true, "WARNING_10"); + NotificationPayload notificationPayload = + new NotificationPayload( + 100L, 1L, NotificationType.THRESHOLD_ALERT, "title", "message", Map.of()); - stubCommon(100L, 1L, eventMonth, eventId); - - UsageUpdateResult luaResult = - new UsageUpdateResult(5000L, 5000L, "NORMAL", 1000L, 0.1, 10000L, false); + stubCommon(100L, 1L, eventMonth, eventId, "appid"); given(usageLuaExecutor.execute(any(UsageLuaExecutor.UsageLuaCommand.class), eq(eventId))) - .willReturn(luaResult); + .willReturn( + new UsageUpdateResult( + 5000L, 5000L, "WARNING_10", 1000L, 0.1, 10000L, true, false)); + given(usageProcessingDecisionMapper.fromLuaStatus("WARNING_10")).willReturn(decision); + given( + usageNotificationPayloadMapper.toNotificationPayload( + any(), any(), any(), eq("WARNING_10"))) + .willReturn(notificationPayload); + given(usageEventOutboxService.stageAfterRedisApplied(eventId, notificationPayload, true)) + .willReturn( + Optional.of( + new UsageEventOutboxService.PendingNotificationDispatch( + 11L, notificationPayload))); + given(usageNotificationPublisher.publishAsync(notificationPayload)) + .willReturn(CompletableFuture.completedFuture(null)); usageSyncServiceImpl.syncUsage(eventId, eventTime, payload); ArgumentCaptor commandCaptor = ArgumentCaptor.forClass(UsageLuaExecutor.UsageLuaCommand.class); - verify(usageLuaExecutor, times(1)).execute(commandCaptor.capture(), eq(eventId)); + verify(usageLuaExecutor).execute(commandCaptor.capture(), eq(eventId)); UsageLuaExecutor.UsageLuaCommand command = commandCaptor.getValue(); - assertEquals("family:100:info:202603", command.infoKey()); - assertEquals("family:100:remaining:202603", command.remainingKey()); - assertEquals("monthlyKey", command.monthlyKey()); - assertEquals("constraintsKey", command.constraintsKey()); - assertEquals("family:100:alert:THRESHOLD:50:202603", command.alert50Key()); - assertEquals("family:100:alert:THRESHOLD:30:202603", command.alert30Key()); - assertEquals("family:100:alert:THRESHOLD:10:202603", command.alert10Key()); - assertEquals("event:dedup:usage:evt_1", command.dedupKey()); - assertEquals(1024L, command.usageBytes()); - assertEquals("1234", command.currentHhmm()); - assertEquals("appid", command.appId()); - assertEquals(60L, command.dedupTtlSeconds()); - - verify(usageEventPublisher, times(1)) - .publish(any(UsageEventPublisher.UsageEventContext.class)); - verify(kafkaMetrics, never()).incrementDedupHit(any(), any(), any()); + assertEquals("family:100:customer:1:alert:THRESHOLD:50:202603", command.alert50Key()); + assertEquals("family:100:customer:1:alert:THRESHOLD:30:202603", command.alert30Key()); + assertEquals("family:100:customer:1:alert:THRESHOLD:10:202603", command.alert10Key()); + assertEquals("family:100:customer:1:alert:MANUAL:202603", command.manualAlertKey()); + assertEquals( + "family:100:customer:1:alert:APP_BLOCK:appid:202603", command.appBlockAlertKey()); + assertEquals("family:100:customer:1:alert:TIME_BLOCK:202603", command.timeBlockAlertKey()); + assertEquals( + "family:100:customer:1:alert:MONTHLY_LIMIT_EXCEEDED:202603", + command.monthlyLimitAlertKey()); + assertEquals( + "family:100:customer:1:alert:FAMILY_QUOTA_EXCEEDED:202603", + command.familyQuotaAlertKey()); + + verify(usagePersistService).persistFromUsageEvent(eventId, eventTime, payload, "ALLOWED"); + verify(usageNotificationPublisher).publishAsync(notificationPayload); + verify(usageEventOutboxService).markSent(11L); } @Test - @DisplayName("appId는 Lua 전달 전에 소문자로 정규화한다") - void syncUsage_NormalizesAppIdBeforeLuaExecution() { - String eventId = "evt_3"; + @DisplayName("가족 구성원 관계가 다르면 초입에서 즉시 중단한다") + void syncUsage_InvalidFamilyMembershipThrows() { + UsagePayload payload = new UsagePayload(100L, 1L, "appId", 1024L, Map.of()); + given(usageFamilyMembershipCacheHelper.isValidFamilyCustomer(100L, 1L)).willReturn(false); + + assertThrows( + IllegalArgumentException.class, + () -> + usageSyncServiceImpl.syncUsage( + "evt_invalid", "2026-03-04T12:34:56", payload)); + + verify(usageLuaExecutor, never()).execute(any(), any()); + verify(usagePersistService, never()).persistFromUsageEvent(any(), any(), any(), any()); + } + + @Test + @DisplayName("membership 검증 중 인프라 실패가 나면 retryable 예외로 전파한다") + void syncUsage_MembershipLookupFailureThrowsRetryableException() { + UsagePayload payload = new UsagePayload(100L, 1L, "appId", 1024L, Map.of()); + given(usageFamilyMembershipCacheHelper.isValidFamilyCustomer(100L, 1L)) + .willThrow( + new KafkaMessageProcessingException( + "Family membership lookup failed. familyId=100 customerId=1" + + " key=family:100:members", + new RuntimeException("db unavailable"))); + + assertThrows( + KafkaMessageProcessingException.class, + () -> + usageSyncServiceImpl.syncUsage( + "evt_membership_retry", "2026-03-04T12:34:56", payload)); + + verify(usageLuaExecutor, never()).execute(any(), any()); + } + + @Test + @DisplayName("중복 이벤트여도 DB 정산은 멱등하게 다시 진입한다") + void syncUsage_DuplicateStillReentersPersist() { + String eventId = "evt_dup"; String eventTime = "2026-03-04T12:34:56"; LocalDate eventMonth = LocalDate.of(2026, 3, 1); - UsagePayload payload = new UsagePayload(100L, 1L, " Com.YouTube.App ", 1024L, Map.of()); + UsagePayload payload = new UsagePayload(100L, 1L, "appId", 1024L, Map.of()); + UsageProcessingDecisionMapper.UsageProcessingDecision decision = + new UsageProcessingDecisionMapper.UsageProcessingDecision( + "ALLOWED", true, "WARNING_10"); + NotificationPayload notificationPayload = + new NotificationPayload( + 100L, 1L, NotificationType.THRESHOLD_ALERT, "title", "message", Map.of()); - stubCommon(100L, 1L, eventMonth, eventId); + stubCommon(100L, 1L, eventMonth, eventId, "appid"); given(usageLuaExecutor.execute(any(UsageLuaExecutor.UsageLuaCommand.class), eq(eventId))) .willReturn( new UsageUpdateResult( - 5000L, 5000L, "APP_BLOCK", 1000L, 0.1, 10000L, false)); + 5000L, 5000L, "WARNING_10", 1000L, 0.1, 10000L, true, true)); + given(usageProcessingDecisionMapper.fromLuaStatus("WARNING_10")).willReturn(decision); + given( + usageNotificationPayloadMapper.toNotificationPayload( + any(), any(), any(), eq("WARNING_10"))) + .willReturn(notificationPayload); + given(usageEventOutboxService.stageAfterRedisApplied(eventId, notificationPayload, true)) + .willReturn( + Optional.of( + new UsageEventOutboxService.PendingNotificationDispatch( + 21L, notificationPayload))); + given(usageNotificationPublisher.publishAsync(notificationPayload)) + .willReturn(CompletableFuture.completedFuture(null)); usageSyncServiceImpl.syncUsage(eventId, eventTime, payload); - ArgumentCaptor commandCaptor = - ArgumentCaptor.forClass(UsageLuaExecutor.UsageLuaCommand.class); - verify(usageLuaExecutor).execute(commandCaptor.capture(), eq(eventId)); - - assertEquals("com.youtube.app", commandCaptor.getValue().appId()); + verify(usagePersistService).persistFromUsageEvent(eventId, eventTime, payload, "ALLOWED"); + verify(kafkaMetrics) + .incrementDedupHit( + KafkaTopics.USAGE_EVENTS, + KafkaConsumerGroups.DABOM_PROCESSOR_USAGE_MAIN, + KafkaEventTypes.DATA_USAGE); + verify(usageNotificationPublisher).publishAsync(notificationPayload); } @Test - @DisplayName("Warmup 실패 시 Lua 실행과 이벤트 발행을 하지 않는다") - void syncUsage_WarmupFailed() { - String eventId = "evt_2"; - String eventTime = "2026-03-04T10:10:10"; + @DisplayName("중복 이벤트이고 이미 pending notification이 있으면 다시 즉시 발행을 시도한다") + void syncUsage_DuplicateRepublishesExistingPendingNotification() { + String eventId = "evt_dup_pending"; + String eventTime = "2026-03-04T12:34:56"; LocalDate eventMonth = LocalDate.of(2026, 3, 1); UsagePayload payload = new UsagePayload(100L, 1L, "appId", 1024L, Map.of()); + UsageProcessingDecisionMapper.UsageProcessingDecision decision = + new UsageProcessingDecisionMapper.UsageProcessingDecision( + "ALLOWED", false, "NORMAL"); + NotificationPayload notificationPayload = + new NotificationPayload( + 100L, 1L, NotificationType.THRESHOLD_ALERT, "title", "message", Map.of()); - given(redisKeyGenerator.generateFamilyInfoKey(100L, eventMonth)) - .willReturn("family:100:info:202603"); - given(redisKeyGenerator.generateFamilyRemainingKey(100L, eventMonth)) - .willReturn("family:100:remaining:202603"); - given(redisKeyGenerator.generateFamilyCustomerMonthlyUsageKey(100L, 1L, eventMonth)) - .willReturn("monthlyKey"); - given(redisKeyGenerator.generateFamilyCustomerConstraintsKey(100L, 1L)) - .willReturn("constraintsKey"); - given(redisKeyGenerator.generateFamilyAlertKey(100L, 50, eventMonth)) - .willReturn("family:100:alert:THRESHOLD:50:202603"); - given(redisKeyGenerator.generateFamilyAlertKey(100L, 30, eventMonth)) - .willReturn("family:100:alert:THRESHOLD:30:202603"); - given(redisKeyGenerator.generateFamilyAlertKey(100L, 10, eventMonth)) - .willReturn("family:100:alert:THRESHOLD:10:202603"); - given(redisKeyGenerator.generateUsageEventDedupKey(eventId)) - .willReturn("event:dedup:usage:" + eventId); + stubCommon(100L, 1L, eventMonth, eventId, "appid"); + given(usageLuaExecutor.execute(any(UsageLuaExecutor.UsageLuaCommand.class), eq(eventId))) + .willReturn( + new UsageUpdateResult( + 5000L, 5000L, "NORMAL", 1000L, 0.1, 10000L, false, true)); + given(usageProcessingDecisionMapper.fromLuaStatus("NORMAL")).willReturn(decision); + given(usageEventOutboxService.findPendingDispatchByEventId(eventId)) + .willReturn( + Optional.of( + new UsageEventOutboxService.PendingNotificationDispatch( + 31L, notificationPayload))); + given(usageNotificationPublisher.publishAsync(notificationPayload)) + .willReturn(CompletableFuture.completedFuture(null)); - given( - usageRedisWarmupHelper.ensureFamilyInfoCached( - 100L, eventMonth, "family:100:info:202603")) - .willReturn(false); - given( - usageRedisWarmupHelper.ensureRemainingBytesCached( - 100L, eventMonth, "family:100:remaining:202603")) - .willReturn(true); - given(usageRedisWarmupHelper.ensureCustomerUsageCached(100L, 1L, "monthlyKey", eventMonth)) - .willReturn(true); + usageSyncServiceImpl.syncUsage(eventId, eventTime, payload); + + verify(usagePersistService).persistFromUsageEvent(eventId, eventTime, payload, "ALLOWED"); + verify(usageEventOutboxService, never()).stageAfterRedisApplied(any(), any(), anyBoolean()); + verify(usageNotificationPublisher).publishAsync(notificationPayload); + verify(usageEventOutboxService).markSent(31L); + } + + @Test + @DisplayName("알림 dedup에 걸리면 DB 정산만 수행하고 outbox는 만들지 않는다") + void syncUsage_SkipsNotificationWhenShouldNotifyFalse() { + String eventId = "evt_skip_notify"; + String eventTime = "2026-03-04T12:34:56"; + LocalDate eventMonth = LocalDate.of(2026, 3, 1); + UsagePayload payload = new UsagePayload(100L, 1L, "appId", 1024L, Map.of()); + UsageProcessingDecisionMapper.UsageProcessingDecision decision = + new UsageProcessingDecisionMapper.UsageProcessingDecision( + "APP_BLOCK", true, "APP_BLOCK"); + + stubCommon(100L, 1L, eventMonth, eventId, "appid"); + given(usageLuaExecutor.execute(any(UsageLuaExecutor.UsageLuaCommand.class), eq(eventId))) + .willReturn( + new UsageUpdateResult( + 5000L, 5000L, "APP_BLOCK", 1000L, 0.1, 10000L, false, false)); + given(usageProcessingDecisionMapper.fromLuaStatus("APP_BLOCK")).willReturn(decision); + given(usageEventOutboxService.findPendingDispatchByEventId(eventId)) + .willReturn(Optional.empty()); usageSyncServiceImpl.syncUsage(eventId, eventTime, payload); - verify(usageLuaExecutor, never()).execute(any(), any()); - verify(usageEventPublisher, never()).publish(any()); + verify(usagePersistService).persistFromUsageEvent(eventId, eventTime, payload, "APP_BLOCK"); + verify(usageNotificationPayloadMapper, never()) + .toNotificationPayload(any(), any(), any(), any()); + verify(usageEventOutboxService, never()).stageAfterRedisApplied(any(), any(), anyBoolean()); + verify(usageNotificationPublisher, never()).publishAsync(any()); } @Test - @DisplayName("중복 이벤트면 publish를 생략하고 dedup metric만 기록한다") - void syncUsage_DuplicateSkipsPublish() { - String eventId = "evt_dup"; + @DisplayName("NORMAL 이벤트는 notification payload를 만들지 않고 끝난다") + void syncUsage_NormalEventSkipsPayloadCreation() { + String eventId = "evt_normal"; String eventTime = "2026-03-04T12:34:56"; LocalDate eventMonth = LocalDate.of(2026, 3, 1); UsagePayload payload = new UsagePayload(100L, 1L, "appId", 1024L, Map.of()); + UsageProcessingDecisionMapper.UsageProcessingDecision decision = + new UsageProcessingDecisionMapper.UsageProcessingDecision( + "ALLOWED", false, "NORMAL"); - stubCommon(100L, 1L, eventMonth, eventId); + stubCommon(100L, 1L, eventMonth, eventId, "appid"); given(usageLuaExecutor.execute(any(UsageLuaExecutor.UsageLuaCommand.class), eq(eventId))) .willReturn( - new UsageUpdateResult(5000L, 5000L, "DUPLICATE", 1000L, 0.1, 10000L, true)); + new UsageUpdateResult( + 5000L, 5000L, "NORMAL", 1000L, 0.1, 10000L, false, false)); + given(usageProcessingDecisionMapper.fromLuaStatus("NORMAL")).willReturn(decision); + given(usageEventOutboxService.findPendingDispatchByEventId(eventId)) + .willReturn(Optional.empty()); usageSyncServiceImpl.syncUsage(eventId, eventTime, payload); - verify(usageEventPublisher, never()).publish(any()); - verify(kafkaMetrics, times(1)) - .incrementDedupHit( - KafkaTopics.USAGE_EVENTS, - KafkaConsumerGroups.DABOM_PROCESSOR_USAGE_MAIN, - KafkaEventTypes.DATA_USAGE); + verify(usagePersistService).persistFromUsageEvent(eventId, eventTime, payload, "ALLOWED"); + verify(usageNotificationPayloadMapper, never()) + .toNotificationPayload(any(), any(), any(), any()); + verify(usageEventOutboxService, never()).stageAfterRedisApplied(any(), any(), anyBoolean()); + } + + @Test + @DisplayName("알 수 없는 Lua status면 즉시 실패한다") + void syncUsage_UnknownLuaStatusThrows() { + String eventId = "evt_unknown"; + String eventTime = "2026-03-04T12:34:56"; + LocalDate eventMonth = LocalDate.of(2026, 3, 1); + UsagePayload payload = new UsagePayload(100L, 1L, "appId", 1024L, Map.of()); + + stubCommon(100L, 1L, eventMonth, eventId, "appid"); + given(usageLuaExecutor.execute(any(UsageLuaExecutor.UsageLuaCommand.class), eq(eventId))) + .willReturn( + new UsageUpdateResult( + 5000L, 5000L, "NEW_STATUS", 1000L, 0.1, 10000L, true, false)); + given(usageProcessingDecisionMapper.fromLuaStatus("NEW_STATUS")) + .willThrow( + new NonRetryableKafkaMessageProcessingException( + "Unsupported Lua status: NEW_STATUS")); + + assertThrows( + NonRetryableKafkaMessageProcessingException.class, + () -> usageSyncServiceImpl.syncUsage(eventId, eventTime, payload)); + + verify(usagePersistService, never()).persistFromUsageEvent(any(), any(), any(), any()); + } + + @Test + @DisplayName("warmup 실패 시 Lua를 실행하지 않고 예외를 던진다") + void syncUsage_WarmupFailedThrows() { + String eventId = "evt_2"; + String eventTime = "2026-03-04T10:10:10"; + LocalDate eventMonth = LocalDate.of(2026, 3, 1); + UsagePayload payload = new UsagePayload(100L, 1L, "appId", 1024L, Map.of()); + + stubCommonFailure(100L, 1L, eventMonth, eventId, "appid"); + + assertThrows( + KafkaMessageProcessingException.class, + () -> usageSyncServiceImpl.syncUsage(eventId, eventTime, payload)); + + verify(usageLuaExecutor, never()).execute(any(), any()); } - private void stubCommon(long familyId, long customerId, LocalDate eventMonth, String eventId) { + private void stubCommon( + long familyId, long customerId, LocalDate eventMonth, String eventId, String appId) { + given(usageFamilyMembershipCacheHelper.isValidFamilyCustomer(familyId, customerId)) + .willReturn(true); given(redisKeyGenerator.generateFamilyInfoKey(familyId, eventMonth)) .willReturn("family:" + familyId + ":info:202603"); given(redisKeyGenerator.generateFamilyRemainingKey(familyId, eventMonth)) @@ -200,12 +363,76 @@ private void stubCommon(long familyId, long customerId, LocalDate eventMonth, St .willReturn("monthlyKey"); given(redisKeyGenerator.generateFamilyCustomerConstraintsKey(familyId, customerId)) .willReturn("constraintsKey"); - given(redisKeyGenerator.generateFamilyAlertKey(familyId, 50, eventMonth)) - .willReturn("family:" + familyId + ":alert:THRESHOLD:50:202603"); - given(redisKeyGenerator.generateFamilyAlertKey(familyId, 30, eventMonth)) - .willReturn("family:" + familyId + ":alert:THRESHOLD:30:202603"); - given(redisKeyGenerator.generateFamilyAlertKey(familyId, 10, eventMonth)) - .willReturn("family:" + familyId + ":alert:THRESHOLD:10:202603"); + given( + redisKeyGenerator.generateFamilyCustomerThresholdAlertKey( + familyId, customerId, 50, eventMonth)) + .willReturn( + "family:" + + familyId + + ":customer:" + + customerId + + ":alert:THRESHOLD:50:202603"); + given( + redisKeyGenerator.generateFamilyCustomerThresholdAlertKey( + familyId, customerId, 30, eventMonth)) + .willReturn( + "family:" + + familyId + + ":customer:" + + customerId + + ":alert:THRESHOLD:30:202603"); + given( + redisKeyGenerator.generateFamilyCustomerThresholdAlertKey( + familyId, customerId, 10, eventMonth)) + .willReturn( + "family:" + + familyId + + ":customer:" + + customerId + + ":alert:THRESHOLD:10:202603"); + given( + redisKeyGenerator.generateFamilyCustomerBlockAlertKey( + familyId, customerId, "MANUAL", eventMonth)) + .willReturn( + "family:" + familyId + ":customer:" + customerId + ":alert:MANUAL:202603"); + given( + redisKeyGenerator.generateFamilyCustomerAppBlockAlertKey( + familyId, customerId, appId, eventMonth)) + .willReturn( + "family:" + + familyId + + ":customer:" + + customerId + + ":alert:APP_BLOCK:" + + appId + + ":202603"); + given( + redisKeyGenerator.generateFamilyCustomerBlockAlertKey( + familyId, customerId, "TIME_BLOCK", eventMonth)) + .willReturn( + "family:" + + familyId + + ":customer:" + + customerId + + ":alert:TIME_BLOCK:202603"); + given( + redisKeyGenerator.generateFamilyCustomerBlockAlertKey( + familyId, customerId, "MONTHLY_LIMIT_EXCEEDED", eventMonth)) + .willReturn( + "family:" + + familyId + + ":customer:" + + customerId + + ":alert:MONTHLY_LIMIT_EXCEEDED:202603"); + given( + redisKeyGenerator.generateFamilyCustomerBlockAlertKey( + familyId, customerId, "FAMILY_QUOTA_EXCEEDED", eventMonth)) + .willReturn( + "family:" + + familyId + + ":customer:" + + customerId + + ":alert:FAMILY_QUOTA_EXCEEDED:202603"); given(redisKeyGenerator.generateUsageEventDedupKey(eventId)) .willReturn("event:dedup:usage:" + eventId); given( @@ -221,4 +448,13 @@ private void stubCommon(long familyId, long customerId, LocalDate eventMonth, St familyId, customerId, "monthlyKey", eventMonth)) .willReturn(true); } + + private void stubCommonFailure( + long familyId, long customerId, LocalDate eventMonth, String eventId, String appId) { + stubCommon(familyId, customerId, eventMonth, eventId, appId); + given( + usageRedisWarmupHelper.ensureFamilyInfoCached( + familyId, eventMonth, "family:" + familyId + ":info:202603")) + .willReturn(false); + } } diff --git a/src/test/java/com/project/domain/usage/service/helper/UsageEventOutboxServiceTest.java b/src/test/java/com/project/domain/usage/service/helper/UsageEventOutboxServiceTest.java new file mode 100644 index 0000000..6206379 --- /dev/null +++ b/src/test/java/com/project/domain/usage/service/helper/UsageEventOutboxServiceTest.java @@ -0,0 +1,117 @@ +package com.project.domain.usage.service.helper; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +import java.util.Map; +import java.util.Optional; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; + +import com.dabom.messaging.kafka.event.dto.notification.NotificationPayload; +import com.dabom.messaging.kafka.event.dto.notification.NotificationType; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.project.domain.usage.entity.UsageEventOutbox; +import com.project.domain.usage.enums.UsageOutboxStatus; +import com.project.domain.usage.repository.UsageEventOutboxRepository; + +@ExtendWith(MockitoExtension.class) +class UsageEventOutboxServiceTest { + + @InjectMocks private UsageEventOutboxService usageEventOutboxService; + @Mock private UsageEventOutboxRepository usageEventOutboxRepository; + @Spy private ObjectMapper objectMapper = new ObjectMapper(); + + @Test + @DisplayName("notification 대상이면 PUBLISH_PENDING row를 보장한다") + void stageAfterRedisApplied_ToPublishPending() { + NotificationPayload payload = + new NotificationPayload( + 100L, 1L, NotificationType.THRESHOLD_ALERT, "title", "message", Map.of()); + UsageEventOutbox pending = + UsageEventOutbox.builder() + .id(10L) + .eventId("evt_2") + .familyId(100L) + .customerId(1L) + .status(UsageOutboxStatus.PUBLISH_PENDING) + .payloadJson(objectMapper.valueToTree(payload).toString()) + .retryCount(0) + .build(); + + given( + usageEventOutboxRepository.insertPublishPendingIgnore( + eq("evt_2"), eq(100L), eq(1L), any(String.class))) + .willReturn(1); + given(usageEventOutboxRepository.refreshPendingPayload(eq("evt_2"), any(String.class))) + .willReturn(1); + given(usageEventOutboxRepository.findByEventId("evt_2")).willReturn(Optional.of(pending)); + + Optional dispatch = + usageEventOutboxService.stageAfterRedisApplied("evt_2", payload, true); + + assertTrue(dispatch.isPresent()); + assertEquals(100L, dispatch.get().payload().familyId()); + verify(usageEventOutboxRepository) + .insertPublishPendingIgnore(eq("evt_2"), eq(100L), eq(1L), any(String.class)); + verify(usageEventOutboxRepository).refreshPendingPayload(eq("evt_2"), any(String.class)); + } + + @Test + @DisplayName("notification 비대상이면 outbox row를 만들지 않는다") + void stageAfterRedisApplied_WhenNotificationSkipped_ReturnsEmpty() { + NotificationPayload payload = + new NotificationPayload( + 100L, 1L, NotificationType.THRESHOLD_ALERT, "title", "message", Map.of()); + + Optional dispatch = + usageEventOutboxService.stageAfterRedisApplied("evt_3", payload, false); + + assertTrue(dispatch.isEmpty()); + verify(usageEventOutboxRepository, never()) + .insertPublishPendingIgnore(any(), any(Long.class), any(Long.class), any()); + verify(usageEventOutboxRepository, never()).refreshPendingPayload(any(), any()); + } + + @Test + @DisplayName("pending payload는 eventId 기준으로 다시 읽을 수 있다") + void findPendingDispatchByEventId_ReturnsPayload() { + NotificationPayload payload = + new NotificationPayload( + 100L, + 1L, + NotificationType.THRESHOLD_ALERT, + "title", + "message", + Map.of("threshold", 10)); + UsageEventOutbox pending = + UsageEventOutbox.builder() + .id(20L) + .eventId("evt_4") + .familyId(100L) + .customerId(1L) + .status(UsageOutboxStatus.PUBLISH_PENDING) + .payloadJson(objectMapper.valueToTree(payload).toString()) + .retryCount(0) + .build(); + given(usageEventOutboxRepository.findByEventId("evt_4")).willReturn(Optional.of(pending)); + + Optional found = + usageEventOutboxService.findPendingDispatchByEventId("evt_4"); + + assertTrue(found.isPresent()); + assertEquals(20L, found.get().outboxId()); + assertEquals(NotificationType.THRESHOLD_ALERT, found.get().payload().type()); + } +} diff --git a/src/test/java/com/project/domain/usage/service/helper/UsageEventPublisherTest.java b/src/test/java/com/project/domain/usage/service/helper/UsageEventPublisherTest.java deleted file mode 100644 index e045e0b..0000000 --- a/src/test/java/com/project/domain/usage/service/helper/UsageEventPublisherTest.java +++ /dev/null @@ -1,137 +0,0 @@ -package com.project.domain.usage.service.helper; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; - -import java.util.Map; - -import org.junit.jupiter.api.DisplayName; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.ArgumentCaptor; -import org.mockito.InjectMocks; -import org.mockito.Mock; -import org.mockito.junit.jupiter.MockitoExtension; - -import com.dabom.messaging.kafka.contract.KafkaEventTypes; -import com.dabom.messaging.kafka.contract.KafkaTopics; -import com.dabom.messaging.kafka.event.dto.EventEnvelope; -import com.dabom.messaging.kafka.event.dto.notification.NotificationPayload; -import com.dabom.messaging.kafka.event.dto.notification.NotificationSubTypes; -import com.dabom.messaging.kafka.event.dto.usage.UsagePayload; -import com.dabom.messaging.kafka.event.dto.usage.UsagePersistPayload; -import com.dabom.messaging.kafka.event.publisher.KafkaEventPublisher; -import com.project.domain.usage.service.dto.UsageUpdateResult; - -@ExtendWith(MockitoExtension.class) -class UsageEventPublisherTest { - - @InjectMocks private UsageEventPublisher usageEventPublisher; - - @Mock private KafkaEventPublisher kafkaEventPublisher; - - @Test - @DisplayName("NORMAL 상태면 persist를 ALLOWED로 발행하고 알림은 발행하지 않는다") - void publish_Normal() { - UsagePayload payload = new UsagePayload(100L, 1L, "app", 1024L, Map.of()); - UsageUpdateResult result = - new UsageUpdateResult(5000L, 5000L, "NORMAL", 1000L, 0.1, 10000L, false); - UsageEventPublisher.UsageEventContext ctx = - new UsageEventPublisher.UsageEventContext( - "evt_1", "2026-02-24T12:00:00", payload, result); - - usageEventPublisher.publish(ctx); - - ArgumentCaptor persistCaptor = - ArgumentCaptor.forClass(UsagePersistPayload.class); - - verify(kafkaEventPublisher, times(1)) - .publish( - eq(KafkaTopics.USAGE_PERSIST), - eq(KafkaEventTypes.USAGE_PERSIST), - persistCaptor.capture()); - verify(kafkaEventPublisher, times(1)) - .publish(eq(KafkaTopics.USAGE_REALTIME), eq(KafkaEventTypes.USAGE_REALTIME), any()); - verify(kafkaEventPublisher, never()) - .publish(eq(KafkaTopics.NOTIFICATION), any(EventEnvelope.class)); - - assertEquals("ALLOWED", persistCaptor.getValue().processResult()); - } - - @Test - @DisplayName("WARNING 상태면 임계치 알림을 발행한다") - void publish_Warning() { - UsagePayload payload = new UsagePayload(100L, 1L, "app", 1024L, Map.of()); - UsageUpdateResult result = - new UsageUpdateResult(9000L, 1000L, "WARNING_10", 2000L, 0.2, 10000L, false); - UsageEventPublisher.UsageEventContext ctx = - new UsageEventPublisher.UsageEventContext( - "evt_2", "2026-02-24T12:00:00", payload, result); - - usageEventPublisher.publish(ctx); - - @SuppressWarnings("unchecked") - ArgumentCaptor> envelopeCaptor = - ArgumentCaptor.forClass((Class) EventEnvelope.class); - - verify(kafkaEventPublisher, times(1)) - .publish(eq(KafkaTopics.NOTIFICATION), envelopeCaptor.capture()); - - EventEnvelope envelope = envelopeCaptor.getValue(); - assertEquals(KafkaEventTypes.NOTIFICATION, envelope.eventType()); - assertEquals(NotificationSubTypes.THRESHOLD_ALERT, envelope.subType()); - } - - @Test - @DisplayName("차단 상태면 차단 알림을 발행한다") - void publish_Blocked() { - UsagePayload payload = new UsagePayload(100L, 1L, "app", 1024L, Map.of()); - UsageUpdateResult result = - new UsageUpdateResult( - 8000L, 2000L, "MONTHLY_LIMIT_EXCEEDED", 10001L, 1.0, 10000L, false); - UsageEventPublisher.UsageEventContext ctx = - new UsageEventPublisher.UsageEventContext( - "evt_3", "2026-02-24T12:00:00", payload, result); - - usageEventPublisher.publish(ctx); - - @SuppressWarnings("unchecked") - ArgumentCaptor> envelopeCaptor = - ArgumentCaptor.forClass((Class) EventEnvelope.class); - - verify(kafkaEventPublisher, times(1)) - .publish(eq(KafkaTopics.NOTIFICATION), envelopeCaptor.capture()); - - assertEquals(NotificationSubTypes.CUSTOMER_BLOCKED, envelopeCaptor.getValue().subType()); - } - - @Test - @DisplayName("APP_BLOCK 상태면 persist와 realtime은 생략하고 차단 알림만 발행한다") - void publish_AppBlock_SkipsPersistAndRealtime() { - UsagePayload payload = new UsagePayload(100L, 1L, "app", 1024L, Map.of()); - UsageUpdateResult result = - new UsageUpdateResult(8000L, 2000L, "APP_BLOCK", 10001L, 1.0, 10000L, false); - UsageEventPublisher.UsageEventContext ctx = - new UsageEventPublisher.UsageEventContext( - "evt_4", "2026-02-24T12:00:00", payload, result); - - usageEventPublisher.publish(ctx); - - @SuppressWarnings("unchecked") - ArgumentCaptor> envelopeCaptor = - ArgumentCaptor.forClass((Class) EventEnvelope.class); - - verify(kafkaEventPublisher, never()) - .publish(eq(KafkaTopics.USAGE_PERSIST), eq(KafkaEventTypes.USAGE_PERSIST), any()); - verify(kafkaEventPublisher, never()) - .publish(eq(KafkaTopics.USAGE_REALTIME), eq(KafkaEventTypes.USAGE_REALTIME), any()); - verify(kafkaEventPublisher, times(1)) - .publish(eq(KafkaTopics.NOTIFICATION), envelopeCaptor.capture()); - - assertEquals(NotificationSubTypes.CUSTOMER_BLOCKED, envelopeCaptor.getValue().subType()); - } -} diff --git a/src/test/java/com/project/domain/usage/service/helper/UsageLuaExecutorTest.java b/src/test/java/com/project/domain/usage/service/helper/UsageLuaExecutorTest.java index cb2d346..2350e66 100644 --- a/src/test/java/com/project/domain/usage/service/helper/UsageLuaExecutorTest.java +++ b/src/test/java/com/project/domain/usage/service/helper/UsageLuaExecutorTest.java @@ -1,7 +1,9 @@ package com.project.domain.usage.service.helper; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.eq; @@ -23,6 +25,7 @@ import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.RedisScript; +import com.dabom.messaging.kafka.error.KafkaMessageProcessingException; import com.project.domain.usage.service.dto.UsageUpdateResult; import com.project.global.util.LogSanitizer; @@ -40,10 +43,10 @@ void setUp() { lenient() .when(logSanitizer.sanitize(nullable(String.class))) .thenAnswer( - invocation -> { - String raw = invocation.getArgument(0); - return raw == null ? "null" : raw; - }); + invocation -> + invocation.getArgument(0) == null + ? "null" + : invocation.getArgument(0)); } @Test @@ -55,9 +58,14 @@ void execute_ParseSuccess() { "family:100:remaining:202603", "monthlyKey", "constraintsKey", - "family:100:alert:THRESHOLD:50:202603", - "family:100:alert:THRESHOLD:30:202603", - "family:100:alert:THRESHOLD:10:202603", + "family:100:customer:1:alert:THRESHOLD:50:202603", + "family:100:customer:1:alert:THRESHOLD:30:202603", + "family:100:customer:1:alert:THRESHOLD:10:202603", + "family:100:customer:1:alert:MANUAL:202603", + "family:100:customer:1:alert:APP_BLOCK:com.youtube.app:202603", + "family:100:customer:1:alert:TIME_BLOCK:202603", + "family:100:customer:1:alert:MONTHLY_LIMIT_EXCEEDED:202603", + "family:100:customer:1:alert:FAMILY_QUOTA_EXCEEDED:202603", "event:dedup:usage:evt_1", 1024L, "2230", @@ -72,7 +80,7 @@ void execute_ParseSuccess() { any(Object.class), any(Object.class), any(Object.class))) - .willReturn(List.of(5000L, 5000L, "NORMAL", 1000L, 0.1, 10000L, 0L)); + .willReturn(List.of(5000L, 5000L, "WARNING_10", 1000L, 0.1, 10000L, 1L, 0L)); UsageUpdateResult result = usageLuaExecutor.execute(command, "evt_1"); @@ -86,32 +94,28 @@ void execute_ParseSuccess() { eq("com.youtube.app"), eq("60")); + assertEquals(13, keysCaptor.getValue().size()); assertEquals( - List.of( - "family:100:info:202603", - "family:100:remaining:202603", - "monthlyKey", - "constraintsKey", - "family:100:alert:THRESHOLD:50:202603", - "family:100:alert:THRESHOLD:30:202603", - "family:100:alert:THRESHOLD:10:202603", - "event:dedup:usage:evt_1"), - keysCaptor.getValue()); + "family:100:customer:1:alert:FAMILY_QUOTA_EXCEEDED:202603", + keysCaptor.getValue().get(11)); + assertEquals("event:dedup:usage:evt_1", keysCaptor.getValue().get(12)); assertEquals(5000L, result.totalUsed()); assertEquals(5000L, result.remaining()); - assertEquals("NORMAL", result.status()); + assertEquals("WARNING_10", result.status()); assertEquals(1000L, result.monthlyUsed()); assertEquals(0.1, result.userRatio()); assertEquals(10000L, result.monthlyLimit()); - assertEquals(false, result.duplicate()); + assertTrue(result.shouldNotify()); + assertFalse(result.duplicate()); } @Test - @DisplayName("duplicate 플래그를 파싱한다") + @DisplayName("duplicate와 notify 플래그를 함께 파싱한다") void execute_ParseDuplicateFlag() { UsageLuaExecutor.UsageLuaCommand command = new UsageLuaExecutor.UsageLuaCommand( - "a", "b", "c", "d", "e", "f", "g", "dup", 1L, "0000", "", 60L); + "a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "dup", 1L, + "0000", "", 60L); given( redisTemplate.execute( eq(usageUpdateScript), @@ -120,12 +124,13 @@ void execute_ParseDuplicateFlag() { any(Object.class), any(Object.class), any(Object.class))) - .willReturn(List.of(100L, 900L, "DUPLICATE", 50L, 0.05, -1L, 1L)); + .willReturn(List.of(100L, 900L, "APP_BLOCK", 50L, 0.05, -1L, 0L, 1L)); UsageUpdateResult result = usageLuaExecutor.execute(command, "evt_dup"); - assertEquals(true, result.duplicate()); - assertEquals("DUPLICATE", result.status()); + assertTrue(result.duplicate()); + assertFalse(result.shouldNotify()); + assertEquals("APP_BLOCK", result.status()); } @Test @@ -133,7 +138,8 @@ void execute_ParseDuplicateFlag() { void execute_NullResult() { UsageLuaExecutor.UsageLuaCommand command = new UsageLuaExecutor.UsageLuaCommand( - "a", "b", "c", "d", "e", "f", "g", "dup", 1L, "0000", "", 60L); + "a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "dup", 1L, + "0000", "", 60L); given( redisTemplate.execute( eq(usageUpdateScript), @@ -144,7 +150,9 @@ void execute_NullResult() { any(Object.class))) .willReturn(null); - assertThrows(IllegalStateException.class, () -> usageLuaExecutor.execute(command, "evt_2")); + assertThrows( + KafkaMessageProcessingException.class, + () -> usageLuaExecutor.execute(command, "evt_2")); } @Test @@ -152,7 +160,8 @@ void execute_NullResult() { void execute_InvalidResultSize() { UsageLuaExecutor.UsageLuaCommand command = new UsageLuaExecutor.UsageLuaCommand( - "a", "b", "c", "d", "e", "f", "g", "dup", 1L, "0000", "", 60L); + "a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "dup", 1L, + "0000", "", 60L); given( redisTemplate.execute( eq(usageUpdateScript), @@ -163,6 +172,8 @@ void execute_InvalidResultSize() { any(Object.class))) .willReturn(List.of(1L, 2L, "NORMAL")); - assertThrows(IllegalStateException.class, () -> usageLuaExecutor.execute(command, "evt_3")); + assertThrows( + KafkaMessageProcessingException.class, + () -> usageLuaExecutor.execute(command, "evt_3")); } } diff --git a/src/test/java/com/project/domain/usage/service/helper/UsageProcessingDecisionMapperTest.java b/src/test/java/com/project/domain/usage/service/helper/UsageProcessingDecisionMapperTest.java new file mode 100644 index 0000000..f973d43 --- /dev/null +++ b/src/test/java/com/project/domain/usage/service/helper/UsageProcessingDecisionMapperTest.java @@ -0,0 +1,46 @@ +package com.project.domain.usage.service.helper; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import com.dabom.messaging.kafka.error.NonRetryableKafkaMessageProcessingException; +import com.project.domain.usage.enums.UsagePersistProcessResult; + +class UsageProcessingDecisionMapperTest { + + private final UsageProcessingDecisionMapper mapper = new UsageProcessingDecisionMapper(); + + @Test + @DisplayName("WARNING 상태는 ALLOWED 정산과 notification 발행 대상으로 해석한다") + void warningStatus() { + UsageProcessingDecisionMapper.UsageProcessingDecision decision = + mapper.fromLuaStatus("WARNING_10"); + + assertThat(decision.persistProcessResult()) + .isEqualTo(UsagePersistProcessResult.ALLOWED.name()); + assertThat(decision.publishNotification()).isTrue(); + assertThat(decision.notificationStatus()).isEqualTo("WARNING_10"); + } + + @Test + @DisplayName("차단 상태는 차단 정산과 notification 발행 대상으로 해석한다") + void blockedStatus() { + UsageProcessingDecisionMapper.UsageProcessingDecision decision = + mapper.fromLuaStatus("APP_BLOCK"); + + assertThat(decision.persistProcessResult()).isEqualTo("APP_BLOCK"); + assertThat(decision.publishNotification()).isTrue(); + assertThat(decision.notificationStatus()).isEqualTo("APP_BLOCK"); + } + + @Test + @DisplayName("알 수 없는 상태는 즉시 DLQ 대상 예외로 처리한다") + void unknownStatus() { + assertThrows( + NonRetryableKafkaMessageProcessingException.class, + () -> mapper.fromLuaStatus("NEW_STATUS")); + } +}