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

import com.kt.event.analytics.entity.ChannelStats;
import com.kt.event.analytics.entity.EventStats;
import com.kt.event.analytics.messaging.event.ParticipantRegisteredEvent;
import com.kt.event.analytics.repository.ChannelStatsRepository;
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.util.concurrent.TimeUnit;

/**
 * 참여자 등록 Consumer
 *
 * 참여자 등록 시 실시간 참여자 수 업데이트
 */
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.kafka.enabled", havingValue = "true", matchIfMissing = false)
@RequiredArgsConstructor
public class ParticipantRegisteredConsumer {

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

    private static final String PROCESSED_PARTICIPANTS_KEY = "processed_participants_v2";
    private static final String CACHE_KEY_PREFIX = "analytics:dashboard:";
    private static final long IDEMPOTENCY_TTL_DAYS = 7;

    /**
     * ParticipantRegistered 이벤트 처리 (MVP용 샘플 토픽)
     *
     * @Transactional 필수: 비관적 락 사용을 위해 트랜잭션 컨텍스트 필요
     */
    @Transactional
    @KafkaListener(topics = "sample.participant.registered", groupId = "${spring.kafka.consumer.group-id}")
    public void handleParticipantRegistered(String message) {
        try {
            log.info("📩 ParticipantRegistered 이벤트 수신: {}", message);

            ParticipantRegisteredEvent event = objectMapper.readValue(message, ParticipantRegisteredEvent.class);
            String participantId = event.getParticipantId();
            String eventId = event.getEventId();
            String channel = event.getChannel();

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

            // 2. 이벤트 통계 업데이트 (참여자 수 +1) - 비관적 락 적용
            eventStatsRepository.findByEventIdWithLock(eventId)
                .ifPresentOrElse(
                    eventStats -> {
                        eventStats.incrementParticipants();
                        eventStatsRepository.save(eventStats);
                        log.info("✅ 참여자 수 업데이트: eventId={}, totalParticipants={}",
                            eventId, eventStats.getTotalParticipants());
                    },
                    () -> log.warn("⚠️ 이벤트 통계 없음: eventId={}", eventId)
                );

            // 3. 채널별 참여자 수 업데이트 - 비관적 락 적용
            if (channel != null && !channel.isEmpty()) {
                channelStatsRepository.findByEventIdAndChannelNameWithLock(eventId, channel)
                    .ifPresentOrElse(
                        channelStats -> {
                            channelStats.incrementParticipants();
                            channelStatsRepository.save(channelStats);
                            log.info("✅ 채널별 참여자 수 업데이트: eventId={}, channel={}, participants={}",
                                eventId, channel, channelStats.getParticipants());
                        },
                        () -> log.warn("⚠️ 채널 통계 없음: eventId={}, channel={}", eventId, channel)
                    );
            }

            // 4. 캐시 무효화 (다음 조회 시 최신 참여자 수 반영)
            String cacheKey = CACHE_KEY_PREFIX + eventId;
            redisTemplate.delete(cacheKey);
            log.debug("🗑️ 캐시 무효화: {}", cacheKey);

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

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