package gc.mda.signal_batch.batch.job;

import gc.mda.signal_batch.batch.reader.AisTargetCacheManager;
import gc.mda.signal_batch.domain.vessel.model.AisTargetEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.*;

/**
 * HourlyJob 편승: 정적 정보 COALESCE + CDC → t_vessel_static INSERT
 *
 * 전략:
 * 1. COALESCE: 캐시에서 직전 1시간 데이터 → 필드별 lastNonEmpty 조합
 * 2. CDC: 이전 저장 레코드와 비교 → 변경 시에만 INSERT
 *
 * 조회: WHERE mmsi=? AND time_bucket <= ? ORDER BY time_bucket DESC LIMIT 1
 */
@Slf4j
@Configuration
@Profile("!query")
@ConditionalOnProperty(name = "vessel.batch.scheduler.enabled", havingValue = "true", matchIfMissing = true)
public class VesselStaticStepConfig {

    private final JobRepository jobRepository;
    private final DataSource queryDataSource;
    private final PlatformTransactionManager transactionManager;
    private final AisTargetCacheManager cacheManager;

    public VesselStaticStepConfig(
            JobRepository jobRepository,
            @Qualifier("queryDataSource") DataSource queryDataSource,
            @Qualifier("queryTransactionManager") PlatformTransactionManager transactionManager,
            AisTargetCacheManager cacheManager) {
        this.jobRepository = jobRepository;
        this.queryDataSource = queryDataSource;
        this.transactionManager = transactionManager;
        this.cacheManager = cacheManager;
    }

    @Bean
    public Step vesselStaticSyncStep() {
        return new StepBuilder("vesselStaticSyncStep", jobRepository)
                .tasklet((contribution, chunkContext) -> {
                    // 1. 캐시에서 전체 데이터 → MMSI별 그룹
                    Collection<AisTargetEntity> allEntities = cacheManager.getAllValues();

                    if (allEntities.isEmpty()) {
                        log.debug("캐시에 데이터 없음 — t_vessel_static 동기화 스킵");
                        return org.springframework.batch.repeat.RepeatStatus.FINISHED;
                    }

                    // 시간 버킷: 현재 시각의 정각
                    LocalDateTime hourBucket = LocalDateTime.now()
                            .withMinute(0).withSecond(0).withNano(0);

                    // MMSI별 최신 데이터 (필드별 COALESCE)
                    Map<String, AisTargetEntity> coalesced = coalesceByMmsi(allEntities);

                    JdbcTemplate jdbcTemplate = new JdbcTemplate(queryDataSource);

                    // 2. CDC: 이전 레코드와 비교 → 변경 시에만 INSERT
                    String selectPrevSql = """
                        SELECT imo, name, callsign, vessel_type, extra_info,
                               length, width, draught, destination, status,
                               signal_kind_code, class_type
                        FROM signal.t_vessel_static
                        WHERE mmsi = ? AND time_bucket <= ?
                        ORDER BY time_bucket DESC
                        LIMIT 1
                    """;

                    String insertSql = """
                        INSERT INTO signal.t_vessel_static (
                            mmsi, time_bucket, imo, name, callsign,
                            vessel_type, extra_info, length, width, draught,
                            destination, eta, status, signal_kind_code, class_type
                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        ON CONFLICT (mmsi, time_bucket) DO UPDATE SET
                            imo = EXCLUDED.imo,
                            name = EXCLUDED.name,
                            callsign = EXCLUDED.callsign,
                            vessel_type = EXCLUDED.vessel_type,
                            extra_info = EXCLUDED.extra_info,
                            length = EXCLUDED.length,
                            width = EXCLUDED.width,
                            draught = EXCLUDED.draught,
                            destination = EXCLUDED.destination,
                            eta = EXCLUDED.eta,
                            status = EXCLUDED.status,
                            signal_kind_code = EXCLUDED.signal_kind_code,
                            class_type = EXCLUDED.class_type
                    """;

                    Timestamp hourBucketTs = Timestamp.valueOf(hourBucket);
                    int inserted = 0;
                    int skipped = 0;

                    List<Object[]> batchArgs = new ArrayList<>();

                    for (Map.Entry<String, AisTargetEntity> entry : coalesced.entrySet()) {
                        String mmsi = entry.getKey();
                        AisTargetEntity current = entry.getValue();

                        // 이전 레코드 조회
                        boolean changed;
                        try {
                            Map<String, Object> prev = jdbcTemplate.queryForMap(
                                    selectPrevSql, mmsi, hourBucketTs);
                            changed = hasStaticInfoChanged(current, prev);
                        } catch (org.springframework.dao.EmptyResultDataAccessException e) {
                            // 이전 레코드 없음 → 첫 INSERT
                            changed = true;
                        }

                        if (changed) {
                            Timestamp etaTs = current.getEta() != null
                                    ? Timestamp.from(current.getEta().toInstant())
                                    : null;

                            batchArgs.add(new Object[] {
                                    mmsi, hourBucketTs,
                                    current.getImo(), current.getName(), current.getCallsign(),
                                    current.getVesselType(), current.getExtraInfo(),
                                    current.getLength(), current.getWidth(), current.getDraught(),
                                    current.getDestination(), etaTs, current.getStatus(),
                                    current.getSignalKindCode(), current.getClassType()
                            });
                            inserted++;
                        } else {
                            skipped++;
                        }
                    }

                    if (!batchArgs.isEmpty()) {
                        jdbcTemplate.batchUpdate(insertSql, batchArgs);
                    }

                    log.info("t_vessel_static 동기화 완료: 총 {} 선박, INSERT {} 건, CDC 스킵 {} 건",
                            coalesced.size(), inserted, skipped);

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

    /**
     * MMSI별 필드 COALESCE: 각 필드별 마지막 non-empty 값 조합
     */
    private Map<String, AisTargetEntity> coalesceByMmsi(Collection<AisTargetEntity> entities) {
        Map<String, AisTargetEntity> result = new LinkedHashMap<>();

        for (AisTargetEntity entity : entities) {
            if (entity.getMmsi() == null) continue;

            result.merge(entity.getMmsi(), entity, (existing, incoming) -> {
                // 더 최신 타임스탬프 기준, 각 필드별 non-empty 우선
                return AisTargetEntity.builder()
                        .mmsi(existing.getMmsi())
                        .imo(coalesce(incoming.getImo(), existing.getImo()))
                        .name(coalesceStr(incoming.getName(), existing.getName()))
                        .callsign(coalesceStr(incoming.getCallsign(), existing.getCallsign()))
                        .vesselType(coalesceStr(incoming.getVesselType(), existing.getVesselType()))
                        .extraInfo(coalesceStr(incoming.getExtraInfo(), existing.getExtraInfo()))
                        .length(coalesce(incoming.getLength(), existing.getLength()))
                        .width(coalesce(incoming.getWidth(), existing.getWidth()))
                        .draught(coalesce(incoming.getDraught(), existing.getDraught()))
                        .destination(coalesceStr(incoming.getDestination(), existing.getDestination()))
                        .eta(coalesce(incoming.getEta(), existing.getEta()))
                        .status(coalesceStr(incoming.getStatus(), existing.getStatus()))
                        .signalKindCode(coalesceStr(incoming.getSignalKindCode(), existing.getSignalKindCode()))
                        .classType(coalesceStr(incoming.getClassType(), existing.getClassType()))
                        .messageTimestamp(coalesce(incoming.getMessageTimestamp(), existing.getMessageTimestamp()))
                        .build();
            });
        }

        return result;
    }

    /**
     * CDC: 정적 정보 변경 여부 비교
     */
    private boolean hasStaticInfoChanged(AisTargetEntity current, Map<String, Object> prev) {
        return !Objects.equals(current.getImo(), toLong(prev.get("imo")))
                || !Objects.equals(current.getName(), prev.get("name"))
                || !Objects.equals(current.getCallsign(), prev.get("callsign"))
                || !Objects.equals(current.getVesselType(), prev.get("vessel_type"))
                || !Objects.equals(current.getExtraInfo(), prev.get("extra_info"))
                || !Objects.equals(current.getLength(), toInt(prev.get("length")))
                || !Objects.equals(current.getWidth(), toInt(prev.get("width")))
                || !Objects.equals(current.getDraught(), toDouble(prev.get("draught")))
                || !Objects.equals(current.getDestination(), prev.get("destination"))
                || !Objects.equals(current.getStatus(), prev.get("status"))
                || !Objects.equals(current.getSignalKindCode(), prev.get("signal_kind_code"))
                || !Objects.equals(current.getClassType(), prev.get("class_type"));
    }

    private <T> T coalesce(T a, T b) {
        return a != null ? a : b;
    }

    private String coalesceStr(String a, String b) {
        return (a != null && !a.isBlank()) ? a : b;
    }

    private Long toLong(Object val) {
        if (val == null) return null;
        if (val instanceof Long l) return l;
        if (val instanceof Number n) return n.longValue();
        return null;
    }

    private Integer toInt(Object val) {
        if (val == null) return null;
        if (val instanceof Integer i) return i;
        if (val instanceof Number n) return n.intValue();
        return null;
    }

    private Double toDouble(Object val) {
        if (val == null) return null;
        if (val instanceof Double d) return d;
        if (val instanceof Number n) return n.doubleValue();
        return null;
    }
}
