UPLUS-16 데이터 사용량 관측 시스템 구현

src/main/java/com/project/consumer/PlanChangeConsumer.java

@@ -0,0 +1,53 @@
+package com.project.consumer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.project.consumer.util.PlanChangeUtil;
+import com.project.consumer.util.RedisUtil;
+import com.project.global.exception.ApplicationException;
+import com.project.global.exception.code.domain.GlobalErrorCode;
+import com.project.producer.schema.CalculatedLimitSchema;
+import com.project.producer.schema.PlanChangeSchema;
+
+import lombok.RequiredArgsConstructor;
+
+@Component
+@RequiredArgsConstructor
+public class PlanChangeConsumer {
+
+    private final ObjectMapper objectMapper;
+    private final PlanChangeUtil planChangeUtil;
+    private final RedisUtil redisUtil;
+
+    @KafkaListener(
+            id = "plan-change-consumer",
+            topics = "change_plan",
+            groupId = "plan-change-consumer",
+            containerFactory = "batchKafkaListenerContainerFactory")
+    public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
+        if (records == null || records.isEmpty()) {
+            ack.acknowledge();
+            return;
+        }
+        List<PlanChangeSchema> events = new ArrayList<>(records.size());
+        try {
+            for (ConsumerRecord<String, String> rec : records) {
+                events.add(objectMapper.readValue(rec.value(), PlanChangeSchema.class));
+            }
+
+            List<CalculatedLimitSchema> limits = planChangeUtil.calculate(events);
+            redisUtil.writePlanChangeBatch(limits);
+
+            ack.acknowledge();
+        } catch (Exception e) {
+            throw new ApplicationException(GlobalErrorCode.PLAN_CHANGE_EVENT_PRODUCE_INVALID);
+        }
+    }
+}

src/main/java/com/project/consumer/UsageConsumer.java

@@ -0,0 +1,67 @@
+package com.project.consumer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.project.consumer.util.RedisUtil;
+import com.project.global.exception.ApplicationException;
+import com.project.global.exception.code.domain.GlobalErrorCode;
+import com.project.producer.NotificationProducer;
+import com.project.producer.schema.UsageEventSchema;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class UsageConsumer {
+
+    private final ObjectMapper objectMapper;
+    private final RedisUtil redisUtil;
+    private final NotificationProducer notificationProducer;
+
+    @KafkaListener(
+            id = "usage-batch-consumer",
+            topics = "usage-data",
+            groupId = "usage-consumer",
+            containerFactory = "batchKafkaListenerContainerFactory")
+    public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
+
+        if (records == null || records.isEmpty()) {
+            ack.acknowledge();
+            return;
+        }
+
+        List<UsageEventSchema> events = new ArrayList<>(records.size());
+
+        try {
+
+            for (ConsumerRecord<String, String> rec : records) {
+                UsageEventSchema schema =
+                        objectMapper.readValue(rec.value(), UsageEventSchema.class);
+                events.add(schema);
+            }
+
+            List<String> notifications = redisUtil.applyUsageBatch(events);
+
+            // 임계치 넘은 것만 notification_topic으로 발행
+            for (String notification : notifications) {
+                notificationProducer.sendNotification(notification);
+            }
+
+            // 성공한 배치 ack
+            ack.acknowledge();
+        } catch (Exception e) {
+            log.error("usage batch failed", e);
+            // 여기서 ack 안 하면 같은 배치가 재시도됨 (at-least-once)
+            throw new ApplicationException(GlobalErrorCode.NOTIFICATION_EVENT_PRODUCE_INVALID);
+        }
+    }
+}

src/main/java/com/project/consumer/util/LuaScriptLoader.java

@@ -0,0 +1,29 @@
+package com.project.consumer.util;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+
+import com.project.global.exception.ApplicationException;
+import com.project.global.exception.code.domain.GlobalErrorCode;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class LuaScriptLoader {
+
+    public static String load(String path) {
+        try (InputStream is = LuaScriptLoader.class.getClassLoader().getResourceAsStream(path)) {
+
+            if (is == null) {
+                throw new IllegalStateException("Lua script not found: " + path);
+            }
+
+            return new String(is.readAllBytes(), StandardCharsets.UTF_8);
+        } catch (Exception e) {
+            log.error("Failed to load lua script");
+            throw new ApplicationException(GlobalErrorCode.LUA_SCRIPT_LOAD_INVALID);
+        }
+    }
+
+    private LuaScriptLoader() {}
+}

src/main/java/com/project/consumer/util/PlanChangeUtil.java

@@ -0,0 +1,108 @@
+package com.project.consumer.util;
+
+import static com.project.consumer.util.UsageTimeUtil.toYearMonth;
+
+import java.time.Duration;
+import java.time.OffsetDateTime;
+import java.time.YearMonth;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Service;
+
+import com.project.producer.schema.CalculatedLimitSchema;
+import com.project.producer.schema.PlanChangeSchema;
+import com.project.producer.test.PlanUnit;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class PlanChangeUtil {
+
+    private final StringRedisTemplate redisTemplate;
+
+    public List<CalculatedLimitSchema> calculate(List<PlanChangeSchema> events) {
+
+        List<CalculatedLimitSchema> results = new ArrayList<>();
+
+        String yearMonth;
+        long finalLimit;
+        String processedKey;
+        Long added;
+
+        for (PlanChangeSchema event : events) {
+            yearMonth = toYearMonth(event.changedAt().toString());
+
+            processedKey = "processed:plan:" + yearMonth + ":" + event.subscriptionId();
+            added = redisTemplate.opsForSet().add(processedKey, event.eventId());
+
+            redisTemplate.expire(
+                    processedKey,
+                    Duration.ofSeconds(
+                            UsageTimeUtil.ttlToNextMonthWithBufferSec(
+                                    event.changedAt().toString(), 2)));
+
+            if (added == null || added == 0L) {
+                // 이미 처리된 요금제 변경 이벤트 → skip
+                continue;
+            }
+
+            String unitKey = "plan:unit:" + event.subscriptionId();
+            String prevUnit = redisTemplate.opsForValue().get(unitKey);
+
+            if (event.unit().equals(PlanUnit.ULTIMATE)) {
+                finalLimit = -1L;
+            } else if (event.unit().equals(PlanUnit.DAY)) {
+                finalLimit = event.allowanceAmount();
+            } else {
+                finalLimit = getMonthFinalLimit(yearMonth, event, prevUnit);
+            }
+
+            long ttlSec =
+                    UsageTimeUtil.ttlToNextMonthWithBufferSec(event.changedAt().toString(), 2);
+
+            results.add(
+                    new CalculatedLimitSchema(
+                            event.subscriptionId(), yearMonth, finalLimit, ttlSec, event.unit()));
+        }
+
+        return results;
+    }
+
+    private long getPreviousLimit(String key) {
+        String value = redisTemplate.opsForValue().get(key);
+        return value == null ? 0L : Long.parseLong(value);
+    }
+
+    // 사용자가 월 기준 중도에 요금제를 변경했을 경우 사용 가능 데이터 집계 로직
+    private long getMonthFinalLimit(String yearMonth, PlanChangeSchema event, String prevUnit) {
+        OffsetDateTime kstTime =
+                event.changedAt().atZoneSameInstant(ZoneId.of("Asia/Seoul")).toOffsetDateTime();
+
+        YearMonth month = YearMonth.from(kstTime);
+        int totalDays = month.lengthOfMonth();
+        int changeDay = kstTime.getDayOfMonth();
+
+        int daysBefore = changeDay - 1;
+        int daysAfter = totalDays - daysBefore;
+
+        // 만약 전 요금제가 무제한 or DAY 요금제 였다면 새로 추가된 요금제의 데이터 양만 계산
+        if (!"MONTH".equals(prevUnit)) {
+            return event.allowanceAmount() * daysAfter / totalDays;
+        }
+
+        String limitKey = "limit:" + yearMonth + ":" + event.subscriptionId();
+
+        long prevLimit = getPreviousLimit(limitKey);
+
+        long allowanceBefore = prevLimit * daysBefore / totalDays;
+        long allowanceAfter = event.allowanceAmount() * daysAfter / totalDays;
+
+        return allowanceBefore + allowanceAfter;
+    }
+}

src/main/java/com/project/consumer/util/RedisUtil.java

@@ -0,0 +1,90 @@
+package com.project.consumer.util;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.data.redis.connection.RedisStringCommands;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.script.DefaultRedisScript;
+import org.springframework.data.redis.core.types.Expiration;
+import org.springframework.stereotype.Service;
+
+import com.project.producer.schema.CalculatedLimitSchema;
+import com.project.producer.schema.UsageEventSchema;
+
+import lombok.RequiredArgsConstructor;
+
+@Service
+@RequiredArgsConstructor
+public class RedisUtil {
+
+    private final StringRedisTemplate redisTemplate;
+
+    private final DefaultRedisScript<List<String>> script =
+            new DefaultRedisScript<>(
+                    LuaScriptLoader.load("lua/usage_batch.lua"),
+                    (Class<List<String>>) (Class<?>) List.class);
+
+    public List<String> applyUsageBatch(List<UsageEventSchema> events) {
+        if (events == null || events.isEmpty()) {
+            return List.of();
+        }
+
+        List<String> args = new ArrayList<>();
+        args.add(String.valueOf(events.size()));
+
+        for (UsageEventSchema e : events) {
+
+            args.add(String.valueOf(e.subscriptionId()));
+            args.add(e.eventId());
+            args.add(String.valueOf(e.usageBytes()));
+            args.add(e.timeStamp());
+        }
+
+        Object result = redisTemplate.execute(script, Collections.emptyList(), args.toArray());
+        return result == null ? List.of() : (List<String>) result;
+    }
+
+    public void writePlanChangeBatch(List<CalculatedLimitSchema> limits) {
+        redisTemplate.executePipelined(
+                (RedisCallback<Void>)
+                        connection -> {
+                            for (CalculatedLimitSchema limit : limits) {
+                                String ke =
+                                        "limit:" + limit.yearMonth() + ":" + limit.subscriptionId();
+
+                                byte[] key = redisTemplate.getStringSerializer().serialize(ke);
+                                byte[] value =
+                                        redisTemplate
+                                                .getStringSerializer()
+                                                .serialize(String.valueOf(limit.limit()));
+
+                                connection
+                                        .stringCommands()
+                                        .set(
+                                                key,
+                                                value,
+                                                Expiration.seconds(limit.ttlSec()),
+                                                RedisStringCommands.SetOption.UPSERT);
+
+                                String unitKey = "plan:unit:" + limit.subscriptionId();
+                                byte[] uk = redisTemplate.getStringSerializer().serialize(unitKey);
+                                byte[] uv =
+                                        redisTemplate
+                                                .getStringSerializer()
+                                                .serialize(limit.unit().name());
+
+                                connection
+                                        .stringCommands()
+                                        .set(
+                                                uk,
+                                                uv,
+                                                Expiration.seconds(limit.ttlSec()),
+                                                RedisStringCommands.SetOption.UPSERT);
+                            }
+                            return null;
+                        });
+    }
+}

src/main/java/com/project/consumer/util/UsageTimeUtil.java

@@ -0,0 +1,30 @@
+package com.project.consumer.util;
+
+import java.time.Duration;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeFormatter;
+
+public class UsageTimeUtil {
+
+    private static final DateTimeFormatter ISO = DateTimeFormatter.ISO_OFFSET_DATE_TIME;
+
+    private UsageTimeUtil() {}
+
+    public static String toYearMonth(String isoTs) {
+        OffsetDateTime odt = OffsetDateTime.parse(isoTs, ISO);
+        return odt.format(DateTimeFormatter.ofPattern("yyyyMM"));
+    }
+
+    public static long ttlToNextMonthWithBufferSec(String isoTs, int bufferDays) {
+        OffsetDateTime now = OffsetDateTime.parse(isoTs, ISO);
+
+        OffsetDateTime nextMonthStart =
+                now.withDayOfMonth(1).with(LocalTime.MIDNIGHT).plusMonths(1);
+
+        Duration base = Duration.between(now, nextMonthStart);
+        Duration buffer = Duration.ofDays(bufferDays);
+
+        return Math.max(base.plus(buffer).getSeconds(), 3600);
+    }
+}