package gc.mda.signal_batch.batch.writer;

import gc.mda.signal_batch.domain.vessel.model.VesselTrack;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.postgresql.copy.CopyManager;
import org.postgresql.core.BaseConnection;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;

import javax.sql.DataSource;
import java.io.*;
import java.sql.Connection;
import java.sql.Timestamp;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
import java.util.LinkedHashMap;
import java.util.stream.Collectors;

@Slf4j
@Component
@ConditionalOnProperty(name = "vessel.batch.scheduler.enabled", havingValue = "true", matchIfMissing = true)
public class VesselTrackBulkWriter implements ItemWriter<List<VesselTrack>> {

    private final DataSource queryDataSource;
    private final JdbcTemplate queryJdbcTemplate;

    public VesselTrackBulkWriter(
            @Qualifier("queryDataSource") DataSource queryDataSource,
            @Qualifier("queryJdbcTemplate") JdbcTemplate queryJdbcTemplate) {
        this.queryDataSource = queryDataSource;
        this.queryJdbcTemplate = queryJdbcTemplate;
    }

    private static final DateTimeFormatter TIMESTAMP_FORMATTER =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    // DataSource 확인을 위한 초기화 메서드
    @jakarta.annotation.PostConstruct
    public void init() {
        try {
            java.sql.Connection conn = queryDataSource.getConnection();
            String url = conn.getMetaData().getURL();
            String user = conn.getMetaData().getUserName();
            conn.close();
            log.warn("========================================");
            log.warn("!!! VesselTrackBulkWriter DataSource INFO !!!");
            log.warn("!!! URL: {}", url);
            log.warn("!!! User: {}", user);
            log.warn("!!! DataSource Class: {}", queryDataSource.getClass().getName());
            log.warn("========================================");
        } catch (Exception e) {
            log.error("Failed to get DataSource info", e);
        }
    }

    private final ObjectMapper objectMapper = new ObjectMapper()
            .registerModule(new JavaTimeModule())
            .setDateFormat(new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));

    @Override
    public void write(Chunk<? extends List<VesselTrack>> chunk) throws Exception {
        List<VesselTrack> allTracks = chunk.getItems().stream()
                .flatMap(List::stream)
                .collect(Collectors.toList());

        if (allTracks.isEmpty()) {
            return;
        }

        try {
            bulkInsertTracks(allTracks, "signal.t_vessel_tracks_5min");
        } catch (Exception e) {
            log.error("Bulk insert failed, using fallback", e);
            fallbackInsert(allTracks, "signal.t_vessel_tracks_5min");
        }
    }

    public void writeHourlyTracks(List<VesselTrack> tracks) throws Exception {
        if (tracks.isEmpty()) {
            return;
        }

        try {
            bulkInsertTracks(tracks, "signal.t_vessel_tracks_hourly");
        } catch (Exception e) {
            log.error("Hourly bulk insert failed, using fallback", e);
            fallbackInsert(tracks, "signal.t_vessel_tracks_hourly");
        }
    }

    public void writeDailyTracks(List<VesselTrack> tracks) throws Exception {
        if (tracks.isEmpty()) {
            return;
        }

        try {
            bulkInsertTracks(tracks, "signal.t_vessel_tracks_daily");
        } catch (Exception e) {
            log.error("Daily bulk insert failed, using fallback", e);
            fallbackInsert(tracks, "signal.t_vessel_tracks_daily");
        }
    }

    // 임시 테이블 + COPY 패턴을 사용한 Bulk Insert
    private void bulkInsertTracks(List<VesselTrack> tracks, String tableName) throws Exception {
        try (Connection conn = queryDataSource.getConnection()) {
            conn.setAutoCommit(false);

            try {
                BaseConnection baseConn = conn.unwrap(BaseConnection.class);
                CopyManager copyManager = new CopyManager(baseConn);

                // 1. 임시 테이블 생성 (UNLOGGED for performance)
                String tempTableName = "temp_vessel_tracks_" + Thread.currentThread().getId();
                try (var stmt = conn.createStatement()) {
                    stmt.execute(String.format("""
                        CREATE TEMP TABLE IF NOT EXISTS %s (
                            mmsi VARCHAR(20),
                            time_bucket TIMESTAMP,
                            track_geom GEOMETRY,
                            distance_nm NUMERIC,
                            avg_speed NUMERIC,
                            max_speed NUMERIC,
                            point_count INTEGER,
                            start_position JSONB,
                            end_position JSONB
                        ) ON COMMIT DROP
                    """, tempTableName));

                    // 임시 테이블 비우기 (이전 실행에서 남은 데이터 제거)
                    stmt.execute("TRUNCATE " + tempTableName);
                }

                // 2. COPY로 임시 테이블에 bulk insert
                String copySql = String.format("""
                    COPY %s (
                        mmsi, time_bucket, track_geom,
                        distance_nm, avg_speed, max_speed, point_count,
                        start_position, end_position
                    ) FROM STDIN
                """, tempTableName);

                StringWriter writer = new StringWriter();
                for (VesselTrack track : tracks) {
                    writer.write(formatTrackLine(track));
                    writer.write('\n');
                }

                long rowsCopied = copyManager.copyIn(copySql, new StringReader(writer.toString()));

                // 3. 임시 테이블에서 최종 테이블로 INSERT (중복 키 무시)
                String insertSql = String.format("""
                    INSERT INTO %s (
                        mmsi, time_bucket, track_geom,
                        distance_nm, avg_speed, max_speed, point_count,
                        start_position, end_position
                    )
                    SELECT
                        mmsi, time_bucket, track_geom,
                        distance_nm, avg_speed, max_speed, point_count,
                        start_position, end_position
                    FROM %s
                    ON CONFLICT (mmsi, time_bucket) DO NOTHING
                """, tableName, tempTableName);

                int rowsInserted;
                try (var stmt = conn.createStatement()) {
                    rowsInserted = stmt.executeUpdate(insertSql);
                }

                conn.commit();
                log.info("Bulk inserted {} vessel tracks to {} (copied: {}, inserted: {})",
                        tracks.size(), tableName, rowsCopied, rowsInserted);

            } catch (Exception e) {
                conn.rollback();
                throw e;
            }
        }
    }

    private String formatTrackLine(VesselTrack track) {
        StringBuilder sb = new StringBuilder();

        sb.append(track.getMmsi()).append('\t');
        sb.append(Timestamp.valueOf(track.getTimeBucket())).append('\t');

        // track_geom
        if (track.getTrackGeom() != null && !track.getTrackGeom().isEmpty()) {
            sb.append(track.getTrackGeom());
        } else {
            sb.append("\\N");
        }
        sb.append('\t');

        // distance_nm
        if (track.getDistanceNm() != null) {
            sb.append(track.getDistanceNm());
        } else {
            sb.append("\\N");
        }
        sb.append('\t');

        // avg_speed
        if (track.getAvgSpeed() != null) {
            sb.append(track.getAvgSpeed());
        } else {
            sb.append("\\N");
        }
        sb.append('\t');

        // max_speed
        if (track.getMaxSpeed() != null) {
            sb.append(track.getMaxSpeed());
        } else {
            sb.append("\\N");
        }
        sb.append('\t');

        // point_count
        sb.append(track.getPointCount()).append('\t');

        // start_position (JSON)
        if (track.getStartPosition() != null) {
            sb.append(formatPositionJson(track.getStartPosition()));
        } else {
            sb.append("\\N");
        }
        sb.append('\t');

        // end_position (JSON)
        if (track.getEndPosition() != null) {
            sb.append(formatPositionJson(track.getEndPosition()));
        } else {
            sb.append("\\N");
        }

        return sb.toString();
    }

    private String formatPositionJson(VesselTrack.TrackPosition position) {
        Map<String, Object> jsonMap = new LinkedHashMap<>();
        jsonMap.put("lat", position.getLat());
        jsonMap.put("lon", position.getLon());
        jsonMap.put("time", position.getTime().format(TIMESTAMP_FORMATTER));
        if (position.getSog() != null) {
            jsonMap.put("sog", position.getSog());
        }

        try {
            return objectMapper.writeValueAsString(jsonMap);
        } catch (Exception e) {
            log.error("Failed to format position JSON", e);
            return "{}";
        }
    }

    private void fallbackInsert(List<VesselTrack> tracks, String tableName) {
        String sql = String.format("""
            INSERT INTO %s (
                mmsi, time_bucket, track_geom,
                distance_nm, avg_speed, max_speed, point_count,
                start_position, end_position
            ) VALUES (?, ?, public.ST_GeomFromText(?), ?, ?, ?, ?, ?::jsonb, ?::jsonb)
            ON CONFLICT (mmsi, time_bucket) DO NOTHING
        """, tableName);

        int inserted = 0;
        for (VesselTrack track : tracks) {
            try {
                queryJdbcTemplate.update(sql,
                    track.getMmsi(),
                    Timestamp.valueOf(track.getTimeBucket()),
                    track.getTrackGeom(),
                    track.getDistanceNm(),
                    track.getAvgSpeed(),
                    track.getMaxSpeed(),
                    track.getPointCount(),
                    track.getStartPosition() != null ? formatPositionJson(track.getStartPosition()) : null,
                    track.getEndPosition() != null ? formatPositionJson(track.getEndPosition()) : null
                );
                inserted++;
            } catch (Exception e) {
                log.error("Failed to insert track for vessel: {} to {}", track.getMmsi(), tableName, e);
            }
        }
        log.info("Fallback inserted {} / {} vessel tracks to {}", inserted, tracks.size(), tableName);
    }
}
