package gc.mda.signal_batch.batch.reader;

import gc.mda.signal_batch.domain.vessel.model.AisTargetEntity;
import gc.mda.signal_batch.domain.vessel.model.VesselData;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.item.ItemReader;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Caffeine 캐시 기반 선박 궤적 데이터 Reader
 *
 * AisTargetCacheManager에서 캐시 스냅샷을 추출하여
 * VesselData 형식으로 변환, MMSI + 5분 time_bucket별 그룹화하여 반환.
 * 하나의 MMSI가 여러 time_bucket에 걸친 데이터를 가질 수 있으므로
 * 각 (MMSI, time_bucket) 조합이 별도의 처리 단위가 된다.
 *
 * 기존 InMemoryVesselTrackDataReader + VesselTrackDataJobListener 대체
 */
@Slf4j
@ConditionalOnProperty(name = "vessel.batch.scheduler.enabled", havingValue = "true", matchIfMissing = true)
@RequiredArgsConstructor
public class CacheBasedVesselTrackDataReader implements ItemReader<List<VesselData>> {

    private final AisTargetCacheManager cacheManager;
    private final int staleDataThresholdDays;

    private Iterator<List<VesselData>> groupIterator;
    private boolean initialized = false;

    @Override
    public List<VesselData> read() {
        if (!initialized) {
            initialize();
            initialized = true;
        }

        if (groupIterator != null && groupIterator.hasNext()) {
            return groupIterator.next();
        }

        return null; // 더 이상 데이터 없음
    }

    private void initialize() {
        // 트랙 버퍼에서 누적 데이터 drain (1분마다 쌓인 위치 이력)
        Map<String, List<AisTargetEntity>> trackBuffer = cacheManager.drainTrackBuffer();

        if (trackBuffer.isEmpty()) {
            log.info("트랙 버퍼에 데이터 없음 — 궤적 생성 스킵");
            groupIterator = Collections.emptyIterator();
            return;
        }

        // AisTargetEntity → VesselData 변환 + MMSI × 5분 time_bucket 이중 그룹화
        LocalDateTime staleCutoff = LocalDateTime.now().minusDays(staleDataThresholdDays);
        List<List<VesselData>> allGroups = new ArrayList<>();
        long totalPoints = 0;
        int totalVessels = 0;
        int skippedStaleGroups = 0;

        for (Map.Entry<String, List<AisTargetEntity>> entry : trackBuffer.entrySet()) {
            List<VesselData> vesselDataList = entry.getValue().stream()
                    .filter(e -> e.getLat() != null && e.getLon() != null)
                    .map(this::toVesselData)
                    .sorted(Comparator.comparing(VesselData::getMessageTime,
                            Comparator.nullsLast(Comparator.naturalOrder())))
                    .collect(Collectors.toList());

            if (vesselDataList.isEmpty()) {
                continue;
            }

            totalVessels++;

            // MMSI 내에서 5분 time_bucket별로 서브 그룹 분할
            Map<LocalDateTime, List<VesselData>> bucketGroups = vesselDataList.stream()
                    .collect(Collectors.groupingBy(
                            (VesselData vd) -> calculateTimeBucket(vd.getMessageTime()),
                            LinkedHashMap::new,
                            Collectors.toList()));

            for (Map.Entry<LocalDateTime, List<VesselData>> bucketEntry : bucketGroups.entrySet()) {
                if (bucketEntry.getKey().isBefore(staleCutoff)) {
                    skippedStaleGroups++;
                    continue;
                }
                allGroups.add(bucketEntry.getValue());
                totalPoints += bucketEntry.getValue().size();
            }
        }

        if (skippedStaleGroups > 0) {
            log.info("Stale data 필터: {}개 그룹 스킵 ({}일 이전 데이터)", skippedStaleGroups, staleDataThresholdDays);
        }
        log.info("트랙 버퍼 Reader 초기화: {} 선박, {} 그룹(MMSI×버킷), {} 포인트 (평균 {}pt/그룹)",
                totalVessels, allGroups.size(), totalPoints,
                allGroups.isEmpty() ? "0.0" : String.format("%.1f", (double) totalPoints / allGroups.size()));

        groupIterator = allGroups.iterator();
    }

    private LocalDateTime calculateTimeBucket(LocalDateTime messageTime) {
        return messageTime.withSecond(0).withNano(0)
                .minusMinutes(messageTime.getMinute() % 5);
    }

    private VesselData toVesselData(AisTargetEntity entity) {
        LocalDateTime messageTime = entity.getMessageTimestamp() != null
                ? entity.getMessageTimestamp().atZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime()
                : LocalDateTime.now();

        return VesselData.builder()
                .mmsi(entity.getMmsi())
                .messageTime(messageTime)
                .lat(entity.getLat())
                .lon(entity.getLon())
                .sog(entity.getSog() != null ? BigDecimal.valueOf(entity.getSog()) : null)
                .cog(entity.getCog() != null ? BigDecimal.valueOf(entity.getCog()) : null)
                .heading(entity.getHeading() != null ? entity.getHeading().intValue() : null)
                .shipNm(entity.getName())
                .shipTy(entity.getVesselType())
                .rot(entity.getRot())
                .build();
    }
}
