package com.kt.event.analytics.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.kt.event.analytics.messaging.event.DistributionCompletedEvent;
import com.kt.event.analytics.messaging.event.EventCreatedEvent;
import com.kt.event.analytics.messaging.event.ParticipantRegisteredEvent;
import com.kt.event.analytics.repository.ChannelStatsRepository;
import com.kt.event.analytics.repository.EventStatsRepository;
import com.kt.event.analytics.repository.TimelineDataRepository;
import jakarta.annotation.PreDestroy;
import jakarta.persistence.EntityManager;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import lombok.Data;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DeleteConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
import org.apache.kafka.common.TopicPartition;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * 샘플 데이터 로더 (Kafka Producer 방식)
 *
 * ⚠️ MVP 전용: 다른 마이크로서비스(Event, Participant, Distribution)가
 *             없는 환경에서 해당 서비스들의 역할을 시뮬레이션합니다.
 *
 * ⚠️ 실제 운영: Analytics Service는 순수 Consumer 역할만 수행해야 하며,
 *             이 클래스는 비활성화되어야 합니다.
 *             → SAMPLE_DATA_ENABLED=false 설정
 *
 * - 서비스 시작 시: Kafka 이벤트 발행하여 샘플 데이터 자동 생성
 * - 서비스 종료 시: PostgreSQL 전체 데이터 삭제
 *
 * 활성화 조건: spring.sample-data.enabled=true (기본값: true)
 */
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.sample-data.enabled", havingValue = "true", matchIfMissing = true)
@RequiredArgsConstructor
public class SampleDataLoader implements ApplicationRunner {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final KafkaAdmin kafkaAdmin;
    private final ObjectMapper objectMapper;
    private final EventStatsRepository eventStatsRepository;
    private final ChannelStatsRepository channelStatsRepository;
    private final TimelineDataRepository timelineDataRepository;
    private final EntityManager entityManager;
    private final RedisTemplate<String, String> redisTemplate;

    private final Random random = new Random();

    @Value("${spring.kafka.consumer.group-id}")
    private String consumerGroupId;

    // Kafka Topic Names (MVP용 샘플 토픽)
    private static final String EVENT_CREATED_TOPIC = "sample.event.created";
    private static final String PARTICIPANT_REGISTERED_TOPIC = "sample.participant.registered";
    private static final String DISTRIBUTION_COMPLETED_TOPIC = "sample.distribution.completed";

    private SampleDataConfig sampleDataConfig;

    @Override
    @Transactional
    public void run(ApplicationArguments args) {
        log.info("========================================");
        log.info("🚀 서비스 시작: Kafka 이벤트 발행하여 샘플 데이터 생성");
        log.info("========================================");

        // 항상 기존 데이터 삭제 후 새로 생성
        long existingCount = eventStatsRepository.count();
        if (existingCount > 0) {
            log.info("기존 데이터 {} 건 삭제 중...", existingCount);
            timelineDataRepository.deleteAll();
            channelStatsRepository.deleteAll();
            eventStatsRepository.deleteAll();

            // 삭제 커밋 보장
            entityManager.flush();
            entityManager.clear();

            log.info("✅ 기존 데이터 삭제 완료");
        }

        // Redis 멱등성 키 삭제 (새로운 이벤트 처리를 위해)
        log.info("Redis 멱등성 키 삭제 중...");
        try {
            redisTemplate.delete("processed_events_v2");
            redisTemplate.delete("distribution_completed_v2");
            redisTemplate.delete("processed_participants_v2");
            log.info("✅ Redis 멱등성 키 삭제 완료");
        } catch (Exception e) {
            log.warn("⚠️ Redis 삭제 실패 (read-only replica일 수 있음): {}", e.getMessage());
            log.info("→ Redis 삭제 건너뛰고 계속 진행...");
        }

        try {
            // JSON 파일에서 샘플 데이터 로드
            log.info("📄 sample-data.json 파일 로드 중...");
            sampleDataConfig = loadSampleData();
            log.info("✅ sample-data.json 로드 완료: 이벤트 {}건, 배포 {}건, 참여자 패턴 {}건",
                    sampleDataConfig.getEvents().size(),
                    sampleDataConfig.getDistributions().size(),
                    sampleDataConfig.getParticipants().size());

            // 1. EventCreated 이벤트 발행
            publishEventCreatedEvents();
            log.info("⏳ EventStats 생성 대기 중... (5초)");
            Thread.sleep(5000);  // EventCreatedConsumer가 EventStats 생성할 시간

            // 2. DistributionCompleted 이벤트 발행
            publishDistributionCompletedEvents();
            log.info("⏳ ChannelStats 생성 대기 중... (3초)");
            Thread.sleep(3000);  // DistributionCompletedConsumer가 ChannelStats 생성할 시간

            // 3. ParticipantRegistered 이벤트 발행
            int totalParticipants = publishParticipantRegisteredEvents();
            log.info("⏳ 참여자 등록 이벤트 처리 대기 중... (20초)");
            Thread.sleep(20000);  // ParticipantRegisteredConsumer가 이벤트 처리할 시간 (비관적 락 고려)

            log.info("========================================");
            log.info("🎉 Kafka 이벤트 발행 완료! (Consumer가 처리 중...)");
            log.info("========================================");
            log.info("발행된 이벤트:");
            log.info("  - EventCreated: {}건", sampleDataConfig.getEvents().size());
            log.info("  - DistributionCompleted: {}건", sampleDataConfig.getDistributions().size());
            log.info("  - ParticipantRegistered: {}건", totalParticipants);
            log.info("========================================");

            // Consumer 처리 대기 (5초)
            log.info("⏳ 참여자 수 업데이트 대기 중... (5초)");
            Thread.sleep(5000);

            // 4. TimelineData 생성 (시간대별 데이터)
            createTimelineData();
            log.info("✅ TimelineData 생성 완료");

        } catch (Exception e) {
            log.error("샘플 데이터 적재 중 오류 발생", e);
        }
    }

    /**
     * 서비스 종료 시 전체 데이터 삭제 및 Consumer Offset 리셋
     */
    @PreDestroy
    @Transactional
    public void onShutdown() {
        log.info("========================================");
        log.info("🛑 서비스 종료: PostgreSQL 전체 데이터 삭제 + Kafka Consumer Offset 리셋");
        log.info("========================================");

        try {
            // 1. PostgreSQL 데이터 삭제
            long timelineCount = timelineDataRepository.count();
            long channelCount = channelStatsRepository.count();
            long eventCount = eventStatsRepository.count();

            log.info("삭제 대상: 이벤트={}, 채널={}, 타임라인={}",
                    eventCount, channelCount, timelineCount);

            timelineDataRepository.deleteAll();
            channelStatsRepository.deleteAll();
            eventStatsRepository.deleteAll();

            // 삭제 커밋 보장
            entityManager.flush();
            entityManager.clear();

            log.info("✅ 모든 샘플 데이터 삭제 완료!");

            // 2. Kafka Consumer Offset 리셋 (다음 시작 시 처음부터 읽도록)
            resetConsumerOffsets();

            log.info("========================================");

        } catch (Exception e) {
            log.error("샘플 데이터 삭제 중 오류 발생", e);
        }
    }

    /**
     * Kafka Consumer Group Offset 리셋
     *
     * 서비스 종료 시 Consumer offset을 삭제하여 다음 시작 시
     * auto.offset.reset=earliest 설정에 따라 처음부터 읽도록 함
     */
    private void resetConsumerOffsets() {
        try (AdminClient adminClient = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
            log.info("🔄 Kafka Consumer Offset 리셋 시작: group={}", consumerGroupId);

            // 모든 토픽의 offset 삭제
            Set<TopicPartition> partitions = new HashSet<>();

            // 토픽별 파티션 추가 (설계서상 각 토픽은 3개 파티션)
            for (int i = 0; i < 3; i++) {
                partitions.add(new TopicPartition(EVENT_CREATED_TOPIC, i));
                partitions.add(new TopicPartition(PARTICIPANT_REGISTERED_TOPIC, i));
                partitions.add(new TopicPartition(DISTRIBUTION_COMPLETED_TOPIC, i));
            }

            // Consumer Group Offset 삭제
            DeleteConsumerGroupOffsetsResult result = adminClient.deleteConsumerGroupOffsets(
                consumerGroupId,
                partitions
            );

            // 완료 대기 (최대 10초)
            result.all().get(10, TimeUnit.SECONDS);

            log.info("✅ Kafka Consumer Offset 리셋 완료!");
            log.info("   → 다음 시작 시 처음부터(earliest) 메시지를 읽습니다.");

        } catch (Exception e) {
            // Offset 리셋 실패는 치명적이지 않으므로 경고만 출력
            log.warn("⚠️ Kafka Consumer Offset 리셋 실패 (무시 가능): {}", e.getMessage());
            log.warn("   → 수동으로 Consumer Group ID를 변경하거나, Kafka 도구로 offset을 삭제하세요.");
        }
    }

    /**
     * EventCreated 이벤트 발행 (JSON 기반)
     */
    private void publishEventCreatedEvents() throws Exception {
        for (EventData eventData : sampleDataConfig.getEvents()) {
            EventCreatedEvent event = EventCreatedEvent.builder()
                .eventId(eventData.getEventId())
                .eventTitle(eventData.getEventTitle())
                .storeId(eventData.getStoreId())
                .totalInvestment(eventData.getTotalInvestment())
                .expectedRevenue(eventData.getExpectedRevenue())
                .status(eventData.getStatus())
                .startDate(parseDateTime(eventData.getStartDate()))
                .endDate(eventData.getEndDate() != null ? parseDateTime(eventData.getEndDate()) : null)
                .build();

            publishEvent(EVENT_CREATED_TOPIC, event);
            log.info("  → EventCreated 발행: eventId={}, title={}",
                    eventData.getEventId(), eventData.getEventTitle());
        }

        log.info("✅ EventCreated 이벤트 {}건 발행 완료", sampleDataConfig.getEvents().size());
    }

    /**
     * ISO 8601 형식 문자열을 LocalDateTime으로 파싱
     */
    private java.time.LocalDateTime parseDateTime(String dateTimeStr) {
        return java.time.LocalDateTime.parse(dateTimeStr);
    }

    /**
     * DistributionCompleted 이벤트 발행 (JSON 기반)
     */
    private void publishDistributionCompletedEvents() throws Exception {
        double channelBudgetRatio = sampleDataConfig.getConfig().getChannelBudgetRatio();

        for (DistributionData distributionData : sampleDataConfig.getDistributions()) {
            String eventId = distributionData.getEventId();

            // 해당 이벤트의 총 투자 금액 조회
            EventData eventData = sampleDataConfig.getEvents().stream()
                .filter(e -> e.getEventId().equals(eventId))
                .findFirst()
                .orElseThrow(() -> new IllegalStateException("이벤트를 찾을 수 없습니다: " + eventId));

            BigDecimal totalInvestment = eventData.getTotalInvestment();
            BigDecimal channelBudget = totalInvestment.multiply(BigDecimal.valueOf(channelBudgetRatio));

            // 채널 배열 생성
            List<DistributionCompletedEvent.ChannelDistribution> channels = new ArrayList<>();

            for (ChannelData channelData : distributionData.getChannels()) {
                DistributionCompletedEvent.ChannelDistribution channel =
                    DistributionCompletedEvent.ChannelDistribution.builder()
                        .channel(channelData.getChannel())
                        .channelType(channelData.getChannelType())
                        .status(channelData.getStatus())
                        .expectedViews(channelData.getExpectedViews())
                        .distributionCost(channelBudget.multiply(BigDecimal.valueOf(channelData.getDistributionCostRatio())))
                        .build();

                channels.add(channel);
            }

            // 이벤트 발행
            DistributionCompletedEvent event = DistributionCompletedEvent.builder()
                .eventId(eventId)
                .distributedChannels(channels)
                .completedAt(parseDateTime(distributionData.getCompletedAt()))
                .build();

            publishEvent(DISTRIBUTION_COMPLETED_TOPIC, event);
            log.info("  → DistributionCompleted 발행: eventId={}, 채널={}개",
                    eventId, channels.size());
        }

        log.info("✅ DistributionCompleted 이벤트 {}건 발행 완료", sampleDataConfig.getDistributions().size());
    }

    /**
     * ParticipantRegistered 이벤트 발행 (JSON 기반)
     */
    private int publishParticipantRegisteredEvents() throws Exception {
        String participantIdPrefix = sampleDataConfig.getConfig().getParticipantIdPrefix();
        int participantIdPadding = sampleDataConfig.getConfig().getParticipantIdPadding();

        int totalPublished = 0;

        for (ParticipantData participantData : sampleDataConfig.getParticipants()) {
            String eventId = participantData.getEventId();
            int startUser = participantData.getParticipantRange().getStart();
            int endUser = participantData.getParticipantRange().getEnd();
            int eventParticipants = endUser - startUser + 1;

            log.info("이벤트 {} 참여자 발행 시작: {}{:0" + participantIdPadding + "d}~{}{:0" + participantIdPadding + "d} ({}명)",
                    eventId, participantIdPrefix, startUser, participantIdPrefix, endUser, eventParticipants);

            // 채널별 가중치 누적 합계 계산 (예: SNS=45, 우리동네TV=70, 지니TV=90, 링고비즈=100)
            Map<String, Integer> channelWeights = participantData.getChannelWeights();
            List<String> channels = new ArrayList<>(channelWeights.keySet());
            int[] cumulativeWeights = new int[channels.size()];
            int cumulative = 0;

            for (int i = 0; i < channels.size(); i++) {
                cumulative += channelWeights.get(channels.get(i));
                cumulativeWeights[i] = cumulative;
            }

            // 각 참여자에 대해 ParticipantRegistered 이벤트 발행
            for (int userId = startUser; userId <= endUser; userId++) {
                String participantId = String.format("%s%0" + participantIdPadding + "d",
                        participantIdPrefix, userId);

                // 채널별 가중치 기반 랜덤 배정
                int randomValue = random.nextInt(cumulative);
                String channel = channels.get(0);  // 기본값

                for (int i = 0; i < cumulativeWeights.length; i++) {
                    if (randomValue < cumulativeWeights[i]) {
                        channel = channels.get(i);
                        break;
                    }
                }

                ParticipantRegisteredEvent event = ParticipantRegisteredEvent.builder()
                        .eventId(eventId)
                        .participantId(participantId)
                        .channel(channel)
                        .build();

                publishEvent(PARTICIPANT_REGISTERED_TOPIC, event);
                totalPublished++;

                // 동시성 충돌 방지: 10개마다 100ms 대기
                if (totalPublished % 10 == 0) {
                    Thread.sleep(100);
                }
            }

            log.info("✅ 이벤트 {} 참여자 발행 완료: {}명", eventId, eventParticipants);
        }

        log.info("========================================");
        log.info("✅ ParticipantRegistered 이벤트 {}건 발행 완료", totalPublished);
        log.info("========================================");

        return totalPublished;
    }

    /**
     * TimelineData 생성 (시간대별 샘플 데이터) - JSON 기반
     *
     * - 각 이벤트마다 30일 × 24시간 = 720시간 치 hourly 데이터 생성
     * - interval=hourly: 시간별 표시 (최근 7일 적합)
     * - interval=daily: 일별 자동 집계 (30일 전체)
     * - 참여자 수, 조회수, 참여행동, 전환수, 누적 참여자 수
     */
    private void createTimelineData() {
        log.info("📊 TimelineData 생성 시작...");

        // 각 이벤트별 시간당 기준 참여자 수 계산 (참여자 범위 기반)
        List<EventData> events = sampleDataConfig.getEvents();
        List<ParticipantData> participants = sampleDataConfig.getParticipants();

        for (int eventIndex = 0; eventIndex < events.size(); eventIndex++) {
            EventData event = events.get(eventIndex);
            String eventId = event.getEventId();

            // 해당 이벤트의 총 참여자 수 계산
            ParticipantData participantData = participants.stream()
                    .filter(p -> p.getEventId().equals(eventId))
                    .findFirst()
                    .orElse(null);

            int totalParticipants = 100;  // 기본값
            if (participantData != null) {
                totalParticipants = participantData.getParticipantRange().getEnd()
                        - participantData.getParticipantRange().getStart() + 1;
            }

            // 30일 × 24시간 = 720시간 치 데이터로 나눔
            int baseParticipant = Math.max(1, totalParticipants / (30 * 24));
            int cumulativeParticipants = 0;

            // 이벤트 시작일 파싱
            java.time.LocalDateTime startDate = parseDateTime(event.getStartDate());

            for (int dayOffset = 0; dayOffset < 30; dayOffset++) {
                for (int hour = 0; hour < 24; hour++) {
                    java.time.LocalDateTime timestamp = startDate.plusDays(dayOffset).plusHours(hour);

                    // 시간대별 참여자 수 변화 (낮 시간대 12~20시에 더 많음)
                    int hourMultiplier = (hour >= 12 && hour <= 20) ? 2 : 1;
                    int hourlyParticipants = (baseParticipant * hourMultiplier) + random.nextInt(baseParticipant + 1);

                    cumulativeParticipants += hourlyParticipants;

                    // 조회수는 참여자의 3~5배
                    int hourlyViews = hourlyParticipants * (3 + random.nextInt(3));

                    // 참여행동은 참여자의 1~2배
                    int hourlyEngagement = hourlyParticipants * (1 + random.nextInt(2));

                    // 전환수는 참여자의 50~80%
                    int hourlyConversions = (int) (hourlyParticipants * (0.5 + random.nextDouble() * 0.3));

                    // TimelineData 생성
                    com.kt.event.analytics.entity.TimelineData timelineData =
                            com.kt.event.analytics.entity.TimelineData.builder()
                                    .eventId(eventId)
                                    .timestamp(timestamp)
                                    .participants(hourlyParticipants)
                                    .views(hourlyViews)
                                    .engagement(hourlyEngagement)
                                    .conversions(hourlyConversions)
                                    .cumulativeParticipants(cumulativeParticipants)
                                    .build();

                    timelineDataRepository.save(timelineData);
                }
            }

            log.info("✅ TimelineData 생성 완료: eventId={}, 시작일={}, 30일 × 24시간 = 720건",
                    eventId, startDate.toLocalDate());
        }

        log.info("✅ 전체 TimelineData 생성 완료: {}개 이벤트 × 30일 × 24시간 = {}건",
                events.size(), events.size() * 30 * 24);
    }

    /**
     * Kafka 이벤트 발행 공통 메서드
     */
    private void publishEvent(String topic, Object event) throws Exception {
        String jsonMessage = objectMapper.writeValueAsString(event);
        kafkaTemplate.send(topic, jsonMessage);
    }

    /**
     * JSON 파일에서 샘플 데이터 로드
     */
    private SampleDataConfig loadSampleData() throws IOException {
        ClassPathResource resource = new ClassPathResource("sample-data.json");
        return objectMapper.readValue(resource.getInputStream(), SampleDataConfig.class);
    }

    // ========================================
    // JSON 데이터 구조 (Inner Classes)
    // ========================================

    @Data
    static class SampleDataConfig {
        private List<EventData> events;
        private List<DistributionData> distributions;
        private List<ParticipantData> participants;
        private ConfigData config;
    }

    @Data
    static class EventData {
        private String eventId;
        private String eventTitle;
        private String storeId;
        private BigDecimal totalInvestment;
        private BigDecimal expectedRevenue;
        private String status;
        private String startDate;  // ISO 8601 형식: "2025-01-23T00:00:00"
        private String endDate;    // null 가능
        private String createdAt;
    }

    @Data
    static class DistributionData {
        private String eventId;
        private String completedAt;
        private List<ChannelData> channels;
    }

    @Data
    static class ChannelData {
        private String channel;
        private String channelType;
        private String status;
        private Integer expectedViews;
        private Double distributionCostRatio;
    }

    @Data
    static class ParticipantData {
        private String eventId;
        private ParticipantRange participantRange;
        private Map<String, Integer> channelWeights;
    }

    @Data
    static class ParticipantRange {
        private Integer start;
        private Integer end;
    }

    @Data
    static class ConfigData {
        private Double channelBudgetRatio;
        private String participantIdPrefix;
        private Integer participantIdPadding;
    }
}
