package gc.mda.signal_batch.batch.job;

import gc.mda.signal_batch.domain.vessel.model.VesselData;
import gc.mda.signal_batch.domain.vessel.model.VesselTrack;
import gc.mda.signal_batch.domain.vessel.dto.VesselBucketPositionDto;
import gc.mda.signal_batch.domain.vessel.service.VesselPreviousBucketCache;
import gc.mda.signal_batch.batch.processor.VesselTrackProcessor;
import gc.mda.signal_batch.batch.processor.AbnormalTrackDetector;
import gc.mda.signal_batch.batch.processor.AbnormalTrackDetector.AbnormalDetectionResult;
import gc.mda.signal_batch.batch.reader.AisTargetCacheManager;
import gc.mda.signal_batch.batch.reader.CacheBasedVesselTrackDataReader;
import gc.mda.signal_batch.global.util.TrackClippingUtils;
import gc.mda.signal_batch.batch.writer.VesselTrackBulkWriter;
import gc.mda.signal_batch.batch.writer.AbnormalTrackWriter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.CompositeItemWriter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

import org.springframework.batch.item.Chunk;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Profile;

import javax.sql.DataSource;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import jakarta.annotation.PostConstruct;


@Slf4j
@Configuration
@Profile("!query")  // query 프로파일에서는 배치 작업 비활성화
@ConditionalOnProperty(name = "vessel.batch.scheduler.enabled", havingValue = "true", matchIfMissing = true)
public class VesselTrackStepConfig {

    private final JobRepository jobRepository;
    private final PlatformTransactionManager transactionManager;
    private final DataSource queryDataSource;
    private final VesselTrackProcessor vesselTrackProcessor;
    private final AisTargetCacheManager aisTargetCacheManager;
    private final VesselTrackBulkWriter vesselTrackBulkWriter;
    private final TrackClippingUtils trackClippingUtils;
    private final AbnormalTrackDetector abnormalTrackDetector;
    private final AbnormalTrackWriter abnormalTrackWriter;
    private final VesselPreviousBucketCache previousBucketCache;

    // 현재 처리 중인 버킷의 종료 위치 저장 (캐시 업데이트용)
    private final Map<String, VesselBucketPositionDto> currentBucketEndPositions = new ConcurrentHashMap<>();

    public VesselTrackStepConfig(
            JobRepository jobRepository,
            PlatformTransactionManager transactionManager,
            @Qualifier("queryDataSource") DataSource queryDataSource,
            VesselTrackProcessor vesselTrackProcessor,
            AisTargetCacheManager aisTargetCacheManager,
            VesselTrackBulkWriter vesselTrackBulkWriter,
            TrackClippingUtils trackClippingUtils,
            AbnormalTrackDetector abnormalTrackDetector,
            AbnormalTrackWriter abnormalTrackWriter,
            VesselPreviousBucketCache previousBucketCache) {
        this.jobRepository = jobRepository;
        this.transactionManager = transactionManager;
        this.queryDataSource = queryDataSource;
        this.vesselTrackProcessor = vesselTrackProcessor;
        this.aisTargetCacheManager = aisTargetCacheManager;
        this.vesselTrackBulkWriter = vesselTrackBulkWriter;
        this.trackClippingUtils = trackClippingUtils;
        this.abnormalTrackDetector = abnormalTrackDetector;
        this.abnormalTrackWriter = abnormalTrackWriter;
        this.previousBucketCache = previousBucketCache;
    }

    @Value("${vessel.batch.chunk-size:1000}")
    private int chunkSize;

    @Value("${partition.retention.tables.t_vessel_tracks_5min.retention-days:7}")
    private int trackRetentionDays;
    
    @PostConstruct
    public void init() {
        // 5분 Job의 이름을 명시적으로 설정
        abnormalTrackWriter.setJobName("vesselTrackAggregationJob");
        log.info("AbnormalTrackWriter initialized with job name: vesselTrackAggregationJob");
    }

    @Bean
    public Step vesselTrackStep() {
        return new StepBuilder("vesselTrackStep", jobRepository)
                .<List<VesselData>, List<VesselTrack>>chunk(chunkSize, transactionManager)
                .reader(trackDataReader())
                .processor(trackProcessorWithSimpleFilter())
                .writer(compositeTrackWriter())
                .build();
    }

    @Bean
    @StepScope
    public CacheBasedVesselTrackDataReader trackDataReader() {
        return new CacheBasedVesselTrackDataReader(aisTargetCacheManager, trackRetentionDays);
    }

    @Bean
    @StepScope
    public ItemProcessor<List<VesselData>, List<VesselTrack>> trackProcessorWithSimpleFilter() {
        return items -> {
            // 1. 기본 처리
            List<VesselTrack> tracks = vesselTrackProcessor.process(items);
            if (tracks == null || tracks.isEmpty()) {
                return null;
            }

            // 2. 이전 버킷 위치 조회 (캐시 + DB Fallback)
            List<String> vesselKeys = tracks.stream()
                    .map(VesselTrack::getMmsi)
                    .distinct()
                    .collect(Collectors.toList());

            Map<String, VesselBucketPositionDto> previousPositions =
                    previousBucketCache.getBatch(vesselKeys);

            // 3. 강화된 비정상 궤적 필터링 (버킷 내 + 버킷 간 점프 검출)
            List<VesselTrack> filteredTracks = new ArrayList<>();

            for (VesselTrack track : tracks) {
                boolean isAbnormal = false;
                String abnormalReason = "";

                // S&P AIS API는 선박 전용 — 항공기 구분 불필요
                double speedLimit = 100.0;
                double distanceLimit = 10.0;

                // 버킷 내 평균속도 체크
                if (track.getAvgSpeed() != null && track.getAvgSpeed().doubleValue() >= speedLimit) {
                    isAbnormal = true;
                    abnormalReason = "within_bucket_speed";
                }

                // 버킷 내 이동거리 체크
                if (track.getDistanceNm() != null && track.getDistanceNm().doubleValue() >= distanceLimit) {
                    isAbnormal = true;
                    abnormalReason = "within_bucket_distance";
                }

                // 버킷 간 점프 검출
                if (!isAbnormal && track.getStartPosition() != null) {
                    String vesselKey = track.getMmsi();
                    VesselBucketPositionDto prevPosition = previousPositions.get(vesselKey);

                    if (prevPosition != null) {
                        double jumpDistance = calculateDistance(
                                prevPosition.getEndLat(), prevPosition.getEndLon(),
                                track.getStartPosition().getLat(), track.getStartPosition().getLon()
                        );

                        // S&P AIS API: 위성/지상 구분 불가 → 보수적 30분 gap 허용
                        double maxGapMinutes = 30.0;
                        double expectedMaxDistance = maxGapMinutes / 60.0 * 50.0;

                        if (jumpDistance > expectedMaxDistance) {
                            isAbnormal = true;
                            abnormalReason = "bucket_to_bucket_jump";
                            log.warn("버킷 간 점프 검출: vessel={}, jump_distance={}nm, threshold={}nm, " +
                                    "prev_pos=[{},{}], curr_pos=[{},{}]",
                                    vesselKey,
                                    String.format("%.1f", jumpDistance),
                                    String.format("%.1f", expectedMaxDistance),
                                    String.format("%.4f", prevPosition.getEndLat()),
                                    String.format("%.4f", prevPosition.getEndLon()),
                                    String.format("%.4f", track.getStartPosition().getLat()),
                                    String.format("%.4f", track.getStartPosition().getLon()));
                        }
                    }
                }

                if (isAbnormal) {
                    log.warn("비정상 궤적 감지 [{}]: vessel={}, avg_speed={}, distance={}",
                        abnormalReason, track.getVesselKey(), track.getAvgSpeed(), track.getDistanceNm());
                    saveAbnormalTrack(track, abnormalReason);
                } else {
                    filteredTracks.add(track);

                    // 정상 궤적의 종료 위치 저장 (캐시 업데이트용)
                    if (track.getEndPosition() != null) {
                        currentBucketEndPositions.put(track.getMmsi(), VesselBucketPositionDto.builder()
                                .mmsi(track.getMmsi())
                                .endLon(track.getEndPosition().getLon())
                                .endLat(track.getEndPosition().getLat())
                                .endTime(track.getEndPosition().getTime())
                                .build());
                    }
                }
            }

            return filteredTracks.isEmpty() ? null : filteredTracks;
        };
    }

    /**
     * Haversine 거리 계산 (해리)
     */
    private double calculateDistance(double lat1, double lon1, double lat2, double lon2) {
        final double R = 3440.065; // 지구 반경 (해리)
        double dLat = Math.toRadians(lat2 - lat1);
        double dLon = Math.toRadians(lon2 - lon1);
        double a = Math.sin(dLat/2) * Math.sin(dLat/2) +
                Math.cos(Math.toRadians(lat1)) * Math.cos(Math.toRadians(lat2)) *
                Math.sin(dLon/2) * Math.sin(dLon/2);
        double c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1-a));
        return R * c;
    }
    
    private void saveAbnormalTrack(VesselTrack track, String abnormalReason) {
        try {
            // Job 이름 설정
            abnormalTrackWriter.setJobName("vesselTrackAggregationJob");
            
            List<AbnormalTrackDetector.AbnormalSegment> segments = new ArrayList<>();
            Map<String, Object> details = new ConcurrentHashMap<>();
            details.put("avgSpeed", track.getAvgSpeed());
            details.put("distanceNm", track.getDistanceNm());
            details.put("timeBucket", track.getTimeBucket());
            
            // S&P AIS API는 선박 전용
            double speedLimit = 100.0;
            double distanceLimit = 10.0;
            
            // 비정상 유형 결정
            String abnormalType = "abnormal_5min";
            double actualValue = 0.0;
            double threshold = 0.0;
            String description = "";
            
            if (track.getAvgSpeed() != null && track.getAvgSpeed().doubleValue() >= speedLimit) {
                abnormalType = "extreme_avg_speed_5min";
                actualValue = track.getAvgSpeed().doubleValue();
                threshold = speedLimit;
                description = String.format("5분 비정상 평균속도: %.1f knots", actualValue);
            } else if (track.getDistanceNm() != null && track.getDistanceNm().doubleValue() >= distanceLimit) {
                abnormalType = "extreme_distance_5min";
                actualValue = track.getDistanceNm().doubleValue();
                threshold = distanceLimit;
                description = String.format("5분 비정상 이동거리: %.1f nm", actualValue);
            }
            
            segments.add(AbnormalTrackDetector.AbnormalSegment.builder()
                .type(abnormalType)
                .startIndex(0)
                .endIndex(track.getPointCount() - 1)
                .actualValue(actualValue)
                .threshold(threshold)
                .description(description)
                .details(details)
                .build());
            
            AbnormalDetectionResult result = AbnormalTrackDetector.AbnormalDetectionResult.builder()
                .originalTrack(track)
                .correctedTrack(null)
                .abnormalSegments(segments)
                .hasAbnormalities(true)
                .build();
            
            abnormalTrackWriter.write(new Chunk<>(Arrays.asList(result)));
        } catch (Exception e) {
            log.error("Failed to save 5min abnormal track", e);
        }
    }

    // CompositeItemWriter로 3개 테이블에 동시 저장
    @Bean
    @StepScope
    public ItemWriter<List<VesselTrack>> compositeTrackWriter() {
        CompositeItemWriter<List<VesselTrack>> compositeWriter = new CompositeItemWriter<>();
        compositeWriter.setDelegates(Arrays.asList(
                vesselTrackWriter(),
                gridTrackWriter(),
                areaTrackWriter()
        ));
        return compositeWriter;
    }

    @Bean
    @StepScope
    public ItemWriter<List<VesselTrack>> vesselTrackWriter() {
        // Bulk Writer + 캐시 업데이트를 래핑
        return chunk -> {
            // 1. 기존 Writer로 DB 저장
            vesselTrackBulkWriter.write(chunk);

            // 2. 캐시 업데이트 (현재 버킷 종료 위치)
            if (!currentBucketEndPositions.isEmpty()) {
                List<VesselBucketPositionDto> positions = new ArrayList<>(currentBucketEndPositions.values());
                previousBucketCache.putAll(positions);

                log.debug("Updated previous bucket cache with {} vessel positions",
                        currentBucketEndPositions.size());

                // 다음 청크를 위해 초기화
                currentBucketEndPositions.clear();
            }
        };
    }

    @Bean
    @StepScope
    public ItemWriter<List<VesselTrack>> gridTrackWriter() {
        return chunk -> {
            // 각 track을 해구별로 분할
            List<VesselTrack> allClippedTracks = new ArrayList<>();

            for (List<VesselTrack> trackList : chunk.getItems()) {
                for (VesselTrack track : trackList) {
                    List<VesselTrack> clippedTracks = trackClippingUtils.clipTracksByHaegu(track);
                    allClippedTracks.addAll(clippedTracks);
                }
            }

            if (allClippedTracks.isEmpty()) {
                log.debug("No tracks to write to grid table after clipping");
                return;
            }

            String sql = """
                INSERT INTO signal.t_grid_vessel_tracks (
                    haegu_no, mmsi, time_bucket,
                    distance_nm, avg_speed, point_count, entry_time, exit_time
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT (haegu_no, mmsi, time_bucket) DO NOTHING
            """;

            List<Object[]> args = allClippedTracks.stream()
                    .map(track -> new Object[] {
                            track.getHaeguNo(),
                            track.getMmsi(),
                            Timestamp.valueOf(track.getTimeBucket()),
                            track.getDistanceNm(),
                            track.getAvgSpeed(),
                            track.getPointCount(),
                            track.getEntryTime() != null ? Timestamp.valueOf(track.getEntryTime()) : null,
                            track.getExitTime() != null ? Timestamp.valueOf(track.getExitTime()) : null
                    })
                    .collect(Collectors.toList());

            int[] results = new JdbcTemplate(queryDataSource).batchUpdate(sql, args);
            log.info("Inserted {} clipped records to t_grid_vessel_tracks", results.length);
        };
    }

    @Bean
    @StepScope
    public ItemWriter<List<VesselTrack>> areaTrackWriter() {
        return chunk -> {
            // 각 track을 area별로 분할
            List<VesselTrack> allClippedTracks = new ArrayList<>();

            for (List<VesselTrack> trackList : chunk.getItems()) {
                for (VesselTrack track : trackList) {
                    List<VesselTrack> clippedTracks = trackClippingUtils.clipTracksByArea(track);
                    allClippedTracks.addAll(clippedTracks);
                }
            }

            if (allClippedTracks.isEmpty()) {
                log.debug("No tracks to write to area table after clipping");
                return;
            }

            String sql = """
                INSERT INTO signal.t_area_vessel_tracks (
                    area_id, mmsi, time_bucket,
                    distance_nm, avg_speed, point_count, metrics
                ) VALUES (?, ?, ?, ?, ?, ?, ?::jsonb)
                ON CONFLICT (area_id, mmsi, time_bucket) DO NOTHING
            """;

            List<Object[]> args = allClippedTracks.stream()
                    .map(track -> new Object[] {
                            track.getAreaId(),
                            track.getMmsi(),
                            Timestamp.valueOf(track.getTimeBucket()),
                            track.getDistanceNm(),
                            track.getAvgSpeed(),
                            track.getPointCount(),
                            "{}"
                    })
                    .collect(Collectors.toList());

            int[] results = new JdbcTemplate(queryDataSource).batchUpdate(sql, args);
            log.info("Inserted {} clipped records to t_area_vessel_tracks", results.length);
        };
    }

    // 집계 스텝들 (변경 없음)
    @Bean
    public Step gridTrackSummaryStep() {
        return new StepBuilder("gridTrackSummaryStep", jobRepository)
                .tasklet((contribution, chunkContext) -> {
                    JdbcTemplate jdbcTemplate = new JdbcTemplate(queryDataSource);

                    String sql = """
                        INSERT INTO signal.t_grid_tracks_summary 
                        (haegu_no, time_bucket, total_vessels, total_distance_nm, avg_speed, vessel_list)
                        SELECT 
                            haegu_no,
                            time_bucket,
                            COUNT(DISTINCT mmsi) as total_vessels,
                            SUM(distance_nm) as total_distance_nm,
                            AVG(avg_speed) as avg_speed,
                            jsonb_agg(jsonb_build_object(
                                'mmsi', mmsi,
                                'distance_nm', distance_nm,
                                'avg_speed', avg_speed
                            )) as vessel_list
                        FROM signal.t_grid_vessel_tracks
                        WHERE time_bucket = ?
                        GROUP BY haegu_no, time_bucket
                        ON CONFLICT (haegu_no, time_bucket) 
                        DO UPDATE SET
                            total_vessels = EXCLUDED.total_vessels,
                            total_distance_nm = EXCLUDED.total_distance_nm,
                            avg_speed = EXCLUDED.avg_speed,
                            vessel_list = EXCLUDED.vessel_list
                    """;

                    String timeBucketStr = (String) chunkContext.getStepContext()
                            .getJobParameters().get("timeBucket");
                    Timestamp timeBucket = Timestamp.valueOf(timeBucketStr);

                    int updated = jdbcTemplate.update(sql, timeBucket);
                    log.info("Updated {} grid track summaries for time_bucket: {}", updated, timeBucket);

                    return org.springframework.batch.repeat.RepeatStatus.FINISHED;
                }, transactionManager)
                .build();
    }

    @Bean
    public Step areaTrackSummaryStep() {
        return new StepBuilder("areaTrackSummaryStep", jobRepository)
                .tasklet((contribution, chunkContext) -> {
                    JdbcTemplate jdbcTemplate = new JdbcTemplate(queryDataSource);

                    String sql = """
                        INSERT INTO signal.t_area_tracks_summary 
                        (area_id, time_bucket, total_vessels, total_distance_nm, avg_speed, vessel_list)
                        SELECT 
                            area_id,
                            time_bucket,
                            COUNT(DISTINCT mmsi) as total_vessels,
                            SUM(distance_nm) as total_distance_nm,
                            AVG(avg_speed) as avg_speed,
                            jsonb_agg(jsonb_build_object(
                                'mmsi', mmsi,
                                'distance_nm', distance_nm,
                                'avg_speed', avg_speed
                            )) as vessel_list
                        FROM signal.t_area_vessel_tracks
                        WHERE time_bucket = ?
                        GROUP BY area_id, time_bucket
                        ON CONFLICT (area_id, time_bucket) 
                        DO UPDATE SET
                            total_vessels = EXCLUDED.total_vessels,
                            total_distance_nm = EXCLUDED.total_distance_nm,
                            avg_speed = EXCLUDED.avg_speed,
                            vessel_list = EXCLUDED.vessel_list
                    """;

                    String timeBucketStr = (String) chunkContext.getStepContext()
                            .getJobParameters().get("timeBucket");
                    Timestamp timeBucket = Timestamp.valueOf(timeBucketStr);

                    int updated = jdbcTemplate.update(sql, timeBucket);
                    log.info("Updated {} area track summaries for time_bucket: {}", updated, timeBucket);

                    return org.springframework.batch.repeat.RepeatStatus.FINISHED;
                }, transactionManager)
                .build();
    }
}