package gc.mda.signal_batch.batch.processor;

import gc.mda.signal_batch.domain.vessel.model.VesselTrack;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.jdbc.core.JdbcTemplate;

import java.math.BigDecimal;
import java.sql.ResultSet;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import gc.mda.signal_batch.global.util.LineStringMUtils;
import gc.mda.signal_batch.global.util.TrackSimplificationUtils;
import javax.sql.DataSource;

@Slf4j
@ConditionalOnProperty(name = "vessel.batch.scheduler.enabled", havingValue = "true", matchIfMissing = true)
@RequiredArgsConstructor
public class HourlyTrackProcessor implements ItemProcessor<VesselTrack.VesselKey, VesselTrack> {
    
    private final DataSource queryDataSource;
    private final JdbcTemplate jdbcTemplate;
    private static final DateTimeFormatter TIMESTAMP_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    
    @Override
    public VesselTrack process(VesselTrack.VesselKey vesselKey) throws Exception {
        LocalDateTime hourBucket = vesselKey.getTimeBucket()
                .withMinute(0)
                .withSecond(0)
                .withNano(0);
        
        String sql = """
            WITH ordered_tracks AS (
                SELECT *
                FROM signal.t_vessel_tracks_5min
                WHERE sig_src_cd = ?
                    AND target_id = ?
                    AND time_bucket >= ?
                    AND time_bucket < ?
                    AND track_geom IS NOT NULL
                    AND public.ST_NPoints(track_geom) > 0
                ORDER BY time_bucket
            ),
            merged_coords AS (
                SELECT
                    sig_src_cd,
                    target_id,
                    string_agg(
                        substring(public.ST_AsText(track_geom) from 'M \\((.+)\\)'),
                        ','
                        ORDER BY time_bucket
                    ) FILTER (WHERE track_geom IS NOT NULL) as all_coords
                FROM ordered_tracks
                GROUP BY sig_src_cd, target_id
            ),
            merged_tracks AS (
                SELECT
                    mc.sig_src_cd,
                    mc.target_id,
                    TO_TIMESTAMP(?, 'YYYY-MM-DD HH24:MI:SS') as time_bucket,
                    public.ST_GeomFromText('LINESTRING M(' || mc.all_coords || ')') as merged_geom,
                    (SELECT MAX(max_speed) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as max_speed,
                    (SELECT SUM(point_count) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as total_points,
                    (SELECT MIN(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as start_time,
                    (SELECT MAX(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as end_time,
                    (SELECT start_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket LIMIT 1) as start_pos,
                    (SELECT end_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket DESC LIMIT 1) as end_pos
                FROM merged_coords mc
            ),
            calculated_tracks AS (
                SELECT
                    *,
                    public.ST_Length(merged_geom::geography) / 1852.0 as total_distance,
                    CASE
                        WHEN public.ST_NPoints(merged_geom) > 0 THEN
                            public.ST_M(public.ST_PointN(merged_geom, public.ST_NPoints(merged_geom))) -
                            public.ST_M(public.ST_PointN(merged_geom, 1))
                        ELSE
                            EXTRACT(EPOCH FROM
                                TO_TIMESTAMP(end_pos->>'time', 'YYYY-MM-DD HH24:MI:SS') - TO_TIMESTAMP(start_pos->>'time', 'YYYY-MM-DD HH24:MI:SS')
                            )
                    END as time_diff_seconds
                FROM merged_tracks
            )
            SELECT
                sig_src_cd,
                target_id,
                time_bucket,
                merged_geom,
                total_distance,
                CASE
                    WHEN time_diff_seconds > 0 THEN
                        CAST(LEAST((total_distance / (time_diff_seconds / 3600.0)), 9999.99) AS numeric(6,2))
                    ELSE 0
                END as avg_speed,
                max_speed,
                total_points,
                start_time,
                end_time,
                start_pos,
                end_pos,
                public.ST_AsText(merged_geom) as geom_text
            FROM calculated_tracks
        """;
        
        LocalDateTime startTime = hourBucket;
        LocalDateTime endTime = hourBucket.plusHours(1);

        // Convert to java.sql.Timestamp for proper PostgreSQL type handling
        Timestamp startTimestamp = Timestamp.valueOf(startTime);
        Timestamp endTimestamp = Timestamp.valueOf(endTime);
        Timestamp hourBucketTimestamp = Timestamp.valueOf(hourBucket);

        log.debug("HourlyTrackProcessor params - sig_src_cd: {}, target_id: {}, startTime: {}, endTime: {}, hourBucket: {}",
            vesselKey.getSigSrcCd(), vesselKey.getTargetId(), startTimestamp, endTimestamp, hourBucketTimestamp);

        try {
            return jdbcTemplate.queryForObject(sql,
                (rs, rowNum) -> {
                    try {
                        return buildHourlyTrack(rs, hourBucket);
                    } catch (Exception e) {
                        throw new RuntimeException("Failed to build hourly track", e);
                    }
                },
                vesselKey.getSigSrcCd(), vesselKey.getTargetId(),
                startTimestamp, endTimestamp, hourBucketTimestamp
            );
        } catch (org.springframework.dao.EmptyResultDataAccessException e) {
            log.warn("No 5min data found for vessel {} in time range {}-{}, skipping hourly aggregation",
                vesselKey.getSigSrcCd() + "_" + vesselKey.getTargetId(), startTimestamp, endTimestamp);
            return null;
        } catch (Exception e) {
            log.error("Failed to process hourly track for vessel {}: {}",
                vesselKey.getSigSrcCd() + "_" + vesselKey.getTargetId(), e.getMessage(), e);
            return null;
        }
    }
    
    private VesselTrack buildHourlyTrack(ResultSet rs, LocalDateTime hourBucket) throws Exception {
        // Start/End position 추출
        VesselTrack.TrackPosition startPos = null;
        VesselTrack.TrackPosition endPos = null;
        
        String startPosJson = rs.getString("start_pos");
        String endPosJson = rs.getString("end_pos");
        
        if (startPosJson != null) {
            startPos = parseTrackPosition(startPosJson);
        }
        
        if (endPosJson != null) {
            endPos = parseTrackPosition(endPosJson);
        }
        
        // M값은 이미 SQL에서 재계산됨
        String hourlyLineStringM = rs.getString("geom_text");
        
        // 이동이 거의 없는 포인트 간소화 (10m 이내 생략, 최대 10분 간격)
        String simplifiedLineStringM = TrackSimplificationUtils.simplifyHourlyTrack(hourlyLineStringM);
        
        // 간소화 통계 로깅
        if (!hourlyLineStringM.equals(simplifiedLineStringM)) {
            TrackSimplificationUtils.SimplificationStats stats = 
                TrackSimplificationUtils.getSimplificationStats(hourlyLineStringM, simplifiedLineStringM);
            log.debug("시간별 궤적 간소화 - vessel: {}/{}, 원본: {}포인트, 간소화: {}포인트 ({}% 감소)",
                rs.getString("sig_src_cd"), rs.getString("target_id"),
                stats.originalPoints, stats.simplifiedPoints, (int)stats.reductionRate);
        }
        
        // track_geom만 사용
        return VesselTrack.builder()
                .sigSrcCd(rs.getString("sig_src_cd"))
                .targetId(rs.getString("target_id"))
                .timeBucket(hourBucket)
                .trackGeom(simplifiedLineStringM)
                .distanceNm(rs.getBigDecimal("total_distance"))
                .avgSpeed(rs.getBigDecimal("avg_speed"))
                .maxSpeed(rs.getBigDecimal("max_speed"))
                .pointCount(rs.getInt("total_points"))
                .startPosition(startPos)
                .endPosition(endPos)
                .build();
    }
    
    private VesselTrack.TrackPosition parseTrackPosition(String json) {
        try {
            String latStr = LineStringMUtils.extractJsonValue(json, "lat");
            String lonStr = LineStringMUtils.extractJsonValue(json, "lon");
            String timeStr = LineStringMUtils.extractJsonValue(json, "time");
            String sogStr = LineStringMUtils.extractJsonValue(json, "sog");
            
            return VesselTrack.TrackPosition.builder()
                    .lat(latStr != null ? Double.parseDouble(latStr) : null)
                    .lon(lonStr != null ? Double.parseDouble(lonStr) : null)
                    .time(timeStr != null ? LocalDateTime.parse(timeStr, TIMESTAMP_FORMATTER) : null)
                    .sog(sogStr != null ? new BigDecimal(sogStr) : null)
                    .build();
        } catch (Exception e) {
            log.error("Failed to parse track position: {}", json, e);
            return null;
        }
    }
}