package gc.mda.signal_batch.domain.vessel.service;

import com.github.benmanes.caffeine.cache.Cache;
import gc.mda.signal_batch.domain.vessel.dto.VesselBucketPositionDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cache.CacheManager;
import org.springframework.cache.caffeine.CaffeineCache;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.stereotype.Service;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;

/**
 * 선박 이전 버킷 종료 위치 캐시 관리 서비스
 *
 * 용도: 버킷 간 점프 검출을 위한 이전 버킷 종료 위치 캐싱
 *
 * 캐시 구조:
 * - Key: mmsi (예: "440123456")
 * - Value: VesselBucketPositionDto (endLon, endLat, endTime)
 * - TTL: 120분
 *
 * 데이터 흐름:
 * 1. 5분 집계 시작 시 캐시에서 이전 버킷 위치 조회
 * 2. 캐시 미스 시 t_ais_position 테이블에서 최근 2시간 데이터 조회 (fallback)
 * 3. 현재 버킷과 이전 버킷 간 점프 검출
 * 4. 처리 완료 후 현재 버킷 종료 위치를 캐시에 업데이트
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class VesselPreviousBucketCache {

    private final CacheManager cacheManager;

    @Qualifier("queryJdbcTemplate")
    private final JdbcTemplate queryJdbcTemplate;

    private static final String CACHE_NAME = "vesselPreviousBuckets";

    // DB 전체 조회 중복 방지 플래그
    private volatile boolean fallbackLoaded = false;

    // DB 조회 통계
    private volatile int totalDbQueriesCount = 0;
    private volatile int totalVesselsLoadedFromDb = 0;

    /**
     * 단일 선박의 버킷 종료 위치 조회
     *
     * @param mmsi MMSI 식별자
     * @return 이전 버킷 종료 위치 (캐시 미스 시 null)
     */
    public VesselBucketPositionDto get(String mmsi) {
        org.springframework.cache.Cache cache = getCache();

        if (cache != null) {
            org.springframework.cache.Cache.ValueWrapper wrapper = cache.get(mmsi);
            if (wrapper != null && wrapper.get() instanceof VesselBucketPositionDto) {
                return (VesselBucketPositionDto) wrapper.get();
            }
        }

        return null;
    }

    /**
     * 여러 선박의 이전 버킷 위치 일괄 조회 (캐시 + DB Fallback)
     *
     * @param vesselKeys 조회할 선박 키 목록 (mmsi)
     * @return 선박 키 -> 이전 버킷 위치 매핑
     */
    public Map<String, VesselBucketPositionDto> getBatch(List<String> vesselKeys) {
        Map<String, VesselBucketPositionDto> result = new HashMap<>();

        if (vesselKeys == null || vesselKeys.isEmpty()) {
            return result;
        }

        ConcurrentMap<Object, Object> nativeCache = getNativeCache();
        int cacheHits = 0;
        int cacheMisses = 0;

        // 1. 캐시에서 조회
        for (String vesselKey : vesselKeys) {
            if (nativeCache != null && nativeCache.containsKey(vesselKey)) {
                Object value = nativeCache.get(vesselKey);
                if (value instanceof VesselBucketPositionDto position) {
                    result.put(vesselKey, position);
                    cacheHits++;
                }
            } else {
                cacheMisses++;
            }
        }

        // 2. 캐시 미스 시 DB에서 일괄 조회 (최근 2시간) - 첫 실행 시 1회만
        if (cacheMisses > 0 && !fallbackLoaded) {
            synchronized (this) {
                if (!fallbackLoaded) {
                    long startTime = System.currentTimeMillis();
                    Map<String, VesselBucketPositionDto> dbResults = loadFromDatabaseBatch();
                    long elapsed = System.currentTimeMillis() - startTime;

                    // 전체 결과를 캐시에 저장
                    dbResults.values().forEach(this::put);

                    // 요청된 선박들을 결과에 추가
                    for (String vesselKey : vesselKeys) {
                        if (!result.containsKey(vesselKey) && dbResults.containsKey(vesselKey)) {
                            result.put(vesselKey, dbResults.get(vesselKey));
                        }
                    }

                    totalDbQueriesCount++;
                    totalVesselsLoadedFromDb += dbResults.size();
                    fallbackLoaded = true;

                    log.info("Previous bucket cache initialized: loaded {} vessels from DB in {}ms",
                            dbResults.size(), elapsed);
                }
            }
        }

        return result;
    }

    /**
     * DB에서 최근 2시간 데이터 일괄 조회 (Fallback)
     */
    private Map<String, VesselBucketPositionDto> loadFromDatabaseBatch() {
        Map<String, VesselBucketPositionDto> result = new HashMap<>();

        // t_ais_position 테이블에서 최근 2시간 데이터 조회
        String sql = """
            SELECT
                mmsi,
                lon,
                lat,
                last_update
            FROM signal.t_ais_position
            WHERE last_update >= NOW() - INTERVAL '2 hours'
            """;

        try {
            List<VesselBucketPositionDto> positions = queryJdbcTemplate.query(sql, new BucketPositionRowMapper());

            for (VesselBucketPositionDto position : positions) {
                result.put(position.getVesselKey(), position);
            }

            log.debug("Queried {} vessel positions from t_ais_position (last 2 hours)", positions.size());

        } catch (Exception e) {
            log.error("Failed to load previous bucket positions from DB", e);
        }

        return result;
    }

    /**
     * 단일 선박 위치 캐시 저장
     */
    public void put(VesselBucketPositionDto position) {
        if (position == null || position.getMmsi() == null) {
            log.warn("Invalid position data, skipping cache: {}", position);
            return;
        }

        getCache().put(position.getMmsi(), position);
    }

    /**
     * 여러 선박 위치 일괄 캐시 저장
     */
    public void putAll(List<VesselBucketPositionDto> positions) {
        if (positions == null || positions.isEmpty()) {
            log.debug("No positions to cache");
            return;
        }

        int count = 0;
        for (VesselBucketPositionDto position : positions) {
            if (position.getMmsi() != null) {
                put(position);
                count++;
            }
        }

        log.debug("Cached {} vessel bucket positions (Total cache size: {})",
                count, getCacheSize());
    }

    /**
     * 캐시 전체 삭제
     */
    public void clear() {
        org.springframework.cache.Cache cache = getCache();
        if (cache != null) {
            cache.clear();
            fallbackLoaded = false;
            log.info("Previous bucket cache cleared");
        }
    }

    /**
     * Job 시작 시 플래그 리셋
     */
    public void resetFallbackFlag() {
        fallbackLoaded = false;
        totalDbQueriesCount = 0;
        totalVesselsLoadedFromDb = 0;
        log.debug("Fallback flag and statistics reset");
    }

    /**
     * Job 완료 시 통계 출력
     */
    public void logJobStatistics() {
        if (totalDbQueriesCount > 0) {
            log.info("Bucket jump detection stats - Cache size: {}, DB queries: {}, Vessels loaded: {}",
                    getCacheSize(), totalDbQueriesCount, totalVesselsLoadedFromDb);
        }
    }

    /**
     * Spring Cache 객체 가져오기
     */
    private org.springframework.cache.Cache getCache() {
        return cacheManager.getCache(CACHE_NAME);
    }

    /**
     * Caffeine Cache 네이티브 객체 가져오기
     */
    private Cache<Object, Object> getCaffeineCache() {
        org.springframework.cache.Cache cache = getCache();
        if (cache instanceof CaffeineCache caffeineCache) {
            return caffeineCache.getNativeCache();
        }
        return null;
    }

    /**
     * ConcurrentMap 가져오기 (배치 조회용)
     */
    private ConcurrentMap<Object, Object> getNativeCache() {
        Cache<Object, Object> caffeineCache = getCaffeineCache();
        if (caffeineCache != null) {
            return caffeineCache.asMap();
        }
        return null;
    }

    /**
     * 현재 캐시 크기 조회
     */
    private int getCacheSize() {
        ConcurrentMap<Object, Object> nativeCache = getNativeCache();
        return nativeCache != null ? nativeCache.size() : 0;
    }

    /**
     * RowMapper 구현
     */
    private static class BucketPositionRowMapper implements RowMapper<VesselBucketPositionDto> {
        @Override
        public VesselBucketPositionDto mapRow(ResultSet rs, int rowNum) throws SQLException {
            return VesselBucketPositionDto.builder()
                    .mmsi(rs.getString("mmsi"))
                    .endLon(rs.getDouble("lon"))
                    .endLat(rs.getDouble("lat"))
                    .endTime(rs.getTimestamp("last_update") != null ?
                            rs.getTimestamp("last_update").toLocalDateTime() : null)
                    .build();
        }
    }
}
