package com.kt.event.analytics.messaging.consumer;

import com.kt.event.analytics.entity.EventStats;
import com.kt.event.analytics.messaging.event.EventCreatedEvent;
import com.kt.event.analytics.repository.EventStatsRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * 이벤트 생성 Consumer
 *
 * 이벤트 생성 시 Analytics 통계 초기화
 */
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.kafka.enabled", havingValue = "true", matchIfMissing = false)
@RequiredArgsConstructor
public class EventCreatedConsumer {

    private final EventStatsRepository eventStatsRepository;
    private final ObjectMapper objectMapper;
    private final RedisTemplate<String, String> redisTemplate;

    private static final String PROCESSED_EVENTS_KEY = "processed_events_v2";
    private static final String CACHE_KEY_PREFIX = "analytics:dashboard:";
    private static final long IDEMPOTENCY_TTL_DAYS = 7;

    /**
     * EventCreated 이벤트 처리 (MVP용 샘플 토픽)
     *
     * @Transactional 필수: DB 저장 작업을 위해 트랜잭션 컨텍스트 필요
     */
    @Transactional
    @KafkaListener(topics = "sample.event.created", groupId = "${spring.kafka.consumer.group-id}")
    public void handleEventCreated(String message) {
        try {
            log.info("📩 EventCreated 이벤트 수신: {}", message);

            EventCreatedEvent event = objectMapper.readValue(message, EventCreatedEvent.class);
            String eventId = event.getEventId();

            // ✅ 1. 멱등성 체크 (중복 처리 방지)
            Boolean isProcessed = redisTemplate.opsForSet().isMember(PROCESSED_EVENTS_KEY, eventId);
            if (Boolean.TRUE.equals(isProcessed)) {
                log.warn("⚠️ 중복 이벤트 스킵 (이미 처리됨): eventId={}", eventId);
                return;
            }

            // 2. 이벤트 통계 초기화 (1:1 관계: storeId → userId 매핑)
            EventStats eventStats = EventStats.builder()
                .eventId(eventId)
                .eventTitle(event.getEventTitle())
                .userId(event.getStoreId())  // MVP: 1 user = 1 store, storeId를 userId로 매핑
                .totalParticipants(0)
                .totalInvestment(event.getTotalInvestment())
                .expectedRevenue(event.getExpectedRevenue() != null ? event.getExpectedRevenue() : BigDecimal.ZERO)
                .status(event.getStatus())
                .startDate(event.getStartDate())
                .endDate(event.getEndDate())
                .build();

            eventStatsRepository.save(eventStats);
            log.info("✅ 이벤트 통계 초기화 완료: eventId={}, userId={}, startDate={}, endDate={}",
                eventId, eventStats.getUserId(), event.getStartDate(), event.getEndDate());

            // 3. 캐시 무효화 (다음 조회 시 최신 데이터 반영)
            String cacheKey = CACHE_KEY_PREFIX + eventId;
            redisTemplate.delete(cacheKey);
            log.debug("🗑️ 캐시 무효화: {}", cacheKey);

            // 4. 멱등성 처리 완료 기록 (7일 TTL)
            redisTemplate.opsForSet().add(PROCESSED_EVENTS_KEY, eventId);
            redisTemplate.expire(PROCESSED_EVENTS_KEY, IDEMPOTENCY_TTL_DAYS, TimeUnit.DAYS);
            log.debug("✅ 멱등성 기록: eventId={}", eventId);

        } catch (Exception e) {
            log.error("❌ EventCreated 이벤트 처리 실패: {}", e.getMessage(), e);
            throw new RuntimeException("EventCreated 처리 실패", e);
        }
    }
}
