package gc.mda.signal_batch.global.util;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Profile;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import jakarta.annotation.PostConstruct;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.IntStream;

@Slf4j
@Component
@Profile("!query")  // query 프로파일에서는 파티션 관리 비활성화
@ConditionalOnProperty(name = "vessel.batch.scheduler.enabled", havingValue = "true", matchIfMissing = true)
public class PartitionManager {

    private final JdbcTemplate queryJdbcTemplate;
    private final gc.mda.signal_batch.global.config.PartitionRetentionConfig retentionConfig;

    public PartitionManager(
            @Qualifier("queryJdbcTemplate") JdbcTemplate queryJdbcTemplate,
            gc.mda.signal_batch.global.config.PartitionRetentionConfig retentionConfig) {
        this.queryJdbcTemplate = queryJdbcTemplate;
        this.retentionConfig = retentionConfig;
    }

    private static final DateTimeFormatter PARTITION_DATE_FORMAT = DateTimeFormatter.ofPattern("yyMMdd");
    private static final DateTimeFormatter PARTITION_MONTH_FORMAT = DateTimeFormatter.ofPattern("yyyy_MM");

    // 일별 파티션 테이블 목록
    private static final List<String> QUERY_DAILY_PARTITION_TABLES = List.of(
            "t_vessel_tracks_5min",
            "t_grid_vessel_tracks",
            "t_grid_tracks_summary",
            "t_area_vessel_tracks",
            "t_area_tracks_summary",
            "t_tile_summary",
            "t_area_statistics"
    );

    // 월별 파티션 테이블 목록
    private static final List<String> MONTHLY_PARTITION_TABLES = List.of(
            "t_vessel_tracks_hourly",
            "t_vessel_tracks_daily",
            "t_grid_tracks_summary_hourly",
            "t_grid_tracks_summary_daily",
            "t_area_tracks_summary_hourly",
            "t_area_tracks_summary_daily",
            "t_abnormal_tracks"
    );

    /**
     * 애플리케이션 시작 시 파티션 초기화
     */
    @PostConstruct
    public void initialize() {
        log.info("========== PartitionManager Initialization ==========");

        // DataSource 정보 로깅
        log.info("=== Query DataSource Info ===");
        DataSourceLogger.logJdbcTemplateInfo("PartitionManager-Query", queryJdbcTemplate);

        if (!retentionConfig.isEnableAutoManagement()) {
            log.info("Partition auto-management is disabled");
            log.info("========== End of Initialization ==========");
            return;
        }

        log.info("Initializing partition management...");
        try {
            // 현재 존재하는 테이블 확인
            checkExistingTables();

            // 일별 파티션 생성
            createDailyPartitions(LocalDate.now(), retentionConfig.getFutureDays());

            // 월별 파티션 생성
            createMonthlyPartitions(LocalDate.now());

            // 파티션 상태 리포트
            reportPartitionStatus();
        } catch (Exception e) {
            log.error("Failed to initialize partitions", e);
        }

        log.info("========== End of Initialization ==========");
    }

    /**
     * 매일 자정에 실행되는 파티션 관리 작업
     */
    @Scheduled(cron = "0 0 0 * * *")
    @Transactional
    public void managePartitions() {
        if (!retentionConfig.isEnableAutoManagement()) {
            return;
        }

        log.info("Starting scheduled partition management");

        try {
            LocalDate today = LocalDate.now();

            // 1. 일별 파티션 생성 (CollectDB + QueryDB)
            createDailyPartitions(today, retentionConfig.getFutureDays());

            // 2. 월별 파티션 생성 (매월 1일에 다음달 파티션 생성)
            if (today.getDayOfMonth() == 1) {
                createMonthlyPartitions(today.plusMonths(1));
            }

            // 3. 오래된 파티션 삭제 (테이블별 retention period 적용)
            dropOldPartitionsByTable(today);

            // 4. 파티션 통계 업데이트
            updatePartitionStatistics();

            // 5. 상태 리포트
            reportPartitionStatus();

            log.info("Partition management completed successfully");

        } catch (Exception e) {
            log.error("Failed to manage partitions", e);
        }
    }

    /**
     * 현재 존재하는 테이블 확인
     */
    private void checkExistingTables() {
        log.info("Checking existing tables...");

        // 일별 파티션 테이블 확인
        log.info("--- Daily Partition Tables ---");
        for (String table : QUERY_DAILY_PARTITION_TABLES) {
            checkTableExists("signal", table);
        }

        // 월별 파티션 테이블 확인
        log.info("--- Monthly Partition Tables ---");
        for (String table : MONTHLY_PARTITION_TABLES) {
            checkTableExists("signal", table);
        }
    }

    private void checkTableExists(String schema, String table) {
        String sql = "SELECT EXISTS (SELECT 1 FROM pg_tables WHERE schemaname = ? AND tablename = ?)";
        Boolean exists = queryJdbcTemplate.queryForObject(sql, Boolean.class, schema, table);

        if (Boolean.TRUE.equals(exists)) {
            log.info("Table exists: {}.{}", schema, table);
        } else {
            log.warn("Table NOT found: {}.{}", schema, table);
        }
    }

    /**
     * 일별 파티션 생성 (CollectDB + QueryDB)
     */
    private void createDailyPartitions(LocalDate startDate, int days) {
        log.info("Creating daily partitions for {} days starting from {}", days, startDate);

        List<PartitionTask> tasks = new ArrayList<>();

        IntStream.range(0, days).forEach(offset -> {
            LocalDate targetDate = startDate.plusDays(offset);

            for (String table : QUERY_DAILY_PARTITION_TABLES) {
                tasks.add(new PartitionTask("signal", table, targetDate, queryJdbcTemplate, "daily", "DB"));
            }
        });

        // 병렬 처리
        tasks.parallelStream().forEach(this::createPartition);
    }

    /**
     * 월별 파티션 생성 (QueryDB만)
     */
    private void createMonthlyPartitions(LocalDate targetMonth) {
        log.info("Creating monthly partitions for {}", targetMonth.format(DateTimeFormatter.ofPattern("yyyy-MM")));

        List<PartitionTask> tasks = new ArrayList<>();

        for (String table : MONTHLY_PARTITION_TABLES) {
            tasks.add(new PartitionTask("signal", table, targetMonth, queryJdbcTemplate, "monthly", "DB"));
        }

        // 병렬 처리
        tasks.parallelStream().forEach(this::createPartition);
    }

    /**
     * 개별 파티션 생성
     */
    private void createPartition(PartitionTask task) {
        String partitionName;
        String createSql;

        if (task.partitionType.equals("daily")) {
            partitionName = task.baseTable + "_" + task.date.format(PARTITION_DATE_FORMAT);
            createSql = String.format("""
                CREATE TABLE IF NOT EXISTS %s.%s PARTITION OF %s.%s
                FOR VALUES FROM ('%s') TO ('%s')
                """,
                    task.schema, partitionName, task.schema, task.baseTable,
                    task.date, task.date.plusDays(1)
            );
        } else {
            // monthly
            partitionName = task.baseTable + "_" + task.date.format(PARTITION_MONTH_FORMAT);
            LocalDate firstDayOfMonth = task.date.withDayOfMonth(1);
            LocalDate firstDayOfNextMonth = firstDayOfMonth.plusMonths(1);

            createSql = String.format("""
                CREATE TABLE IF NOT EXISTS %s.%s PARTITION OF %s.%s
                FOR VALUES FROM ('%s') TO ('%s')
                """,
                    task.schema, partitionName, task.schema, task.baseTable,
                    firstDayOfMonth, firstDayOfNextMonth
            );
        }

        try {
            // 파티션 존재 확인
            if (partitionExists(task.schema, partitionName, task.jdbcTemplate)) {
                log.debug("[{}] Partition already exists: {}.{}", task.dbType, task.schema, partitionName);
                return;
            }

            // 파티션 생성
            task.jdbcTemplate.execute(createSql);
            log.info("[{}] Created partition: {}.{}", task.dbType, task.schema, partitionName);

            // 파티션별 인덱스 생성
            createPartitionIndexes(task.schema, partitionName, task.baseTable, task.jdbcTemplate);

        } catch (Exception e) {
            log.error("[{}] Failed to create partition: {}.{}", task.dbType, task.schema, partitionName, e);
        }
    }

    /**
     * 파티션별 인덱스 생성
     */
    private void createPartitionIndexes(String schema, String partitionName,
                                        String baseTable, JdbcTemplate jdbcTemplate) {
        List<String> indexSqls = new ArrayList<>();

        // 5분 궤적 테이블
        if (baseTable.equals("t_vessel_tracks_5min")) {
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_idx ON %s.%s (time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_vessel_idx ON %s.%s (mmsi, time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_geom_idx ON %s.%s USING GIST (track_geom)",
                    partitionName, schema, partitionName
            ));
            // 성능 최적화를 위한 복합 인덱스 (WebSocket API 개선)
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_vessel_include_idx ON %s.%s (time_bucket, mmsi) " +
                    "INCLUDE (distance_nm, avg_speed, max_speed, point_count) WHERE track_geom IS NOT NULL",
                    partitionName, schema, partitionName
            ));
        }
        // 해구별 궤적 테이블
        else if (baseTable.contains("t_grid_vessel_tracks") || baseTable.contains("t_grid_tracks_summary")) {
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_haegu_time_idx ON %s.%s (haegu_no, time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_idx ON %s.%s (time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            
            // 선박별 진입 이력 조회를 위한 인덱스 (순서 있는 다중 해구 진입 체크)
            if (baseTable.equals("t_grid_vessel_tracks")) {
                indexSqls.add(String.format(
                        "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_vessel_time_idx ON %s.%s (mmsi, time_bucket DESC)",
                        partitionName, schema, partitionName
                ));
                // 진입/퇴출 시간 인덱스
                indexSqls.add(String.format(
                        "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_entry_exit_idx ON %s.%s (entry_time, exit_time) WHERE entry_time IS NOT NULL",
                        partitionName, schema, partitionName
                ));
            }
        }
        // 구역별 궤적 테이블
        else if (baseTable.contains("t_area_vessel_tracks") || baseTable.contains("t_area_tracks_summary")) {
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_area_time_idx ON %s.%s (area_id, time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_idx ON %s.%s (time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            
            // 선박별 진입 이력 조회를 위한 인덱스 (순서 있는 다중 구역 진입 체크)
            if (baseTable.equals("t_area_vessel_tracks")) {
                indexSqls.add(String.format(
                        "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_vessel_time_idx ON %s.%s (mmsi, time_bucket DESC)",
                        partitionName, schema, partitionName
                ));
                // 다중 구역 순차 진입 체크를 위한 복합 인덱스
                indexSqls.add(String.format(
                        "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_area_vessel_time_idx ON %s.%s (area_id, mmsi, time_bucket DESC)",
                        partitionName, schema, partitionName
                ));
            }
        }
        // 타일 집계 테이블
        else if (baseTable.contains("t_tile_summary")) {
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_tile_time_idx ON %s.%s (tile_id, time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_bucket_idx ON %s.%s (time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
        }
        // 구역 통계 테이블
        else if (baseTable.contains("t_area_statistics")) {
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_area_time_idx ON %s.%s (area_id, time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
        }
        // 시간별 궤적 테이블
        else if (baseTable.contains("hourly")) {
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_idx ON %s.%s (time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            if (baseTable.contains("vessel_tracks")) {
                indexSqls.add(String.format(
                        "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_vessel_idx ON %s.%s (mmsi, time_bucket DESC)",
                        partitionName, schema, partitionName
                ));
                // 성능 최적화를 위한 복합 인덱스 (WebSocket API 개선)
                if (baseTable.equals("t_vessel_tracks_hourly")) {
                    indexSqls.add(String.format(
                            "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_vessel_include_idx ON %s.%s (time_bucket, mmsi) " +
                            "INCLUDE (distance_nm, avg_speed, max_speed, point_count) WHERE track_geom IS NOT NULL",
                            partitionName, schema, partitionName
                    ));
                    indexSqls.add(String.format(
                            "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_geom_idx ON %s.%s USING GIST (track_geom)",
                            partitionName, schema, partitionName
                    ));
                }
            }
        }
        // 일별 궤적 테이블
        else if (baseTable.contains("daily")) {
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_idx ON %s.%s (time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            if (baseTable.contains("vessel_tracks")) {
                indexSqls.add(String.format(
                        "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_vessel_idx ON %s.%s (mmsi, time_bucket DESC)",
                        partitionName, schema, partitionName
                ));
                // 성능 최적화를 위한 복합 인덱스 (WebSocket API 개선)
                if (baseTable.equals("t_vessel_tracks_daily")) {
                    indexSqls.add(String.format(
                            "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_vessel_include_idx ON %s.%s (time_bucket, mmsi) " +
                            "INCLUDE (distance_nm, avg_speed, max_speed, point_count) WHERE track_geom IS NOT NULL",
                            partitionName, schema, partitionName
                    ));
                    indexSqls.add(String.format(
                            "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_geom_idx ON %s.%s USING GIST (track_geom)",
                            partitionName, schema, partitionName
                    ));
                }
            }
        }
        // 비정상 궤적 테이블
        else if (baseTable.equals("t_abnormal_tracks")) {
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_time_idx ON %s.%s (time_bucket DESC)",
                    partitionName, schema, partitionName
            ));
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_vessel_idx ON %s.%s (mmsi)",
                    partitionName, schema, partitionName
            ));
            indexSqls.add(String.format(
                    "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s_type_idx ON %s.%s (abnormal_type)",
                    partitionName, schema, partitionName
            ));
        }

        // 인덱스 생성 실행
        for (String indexSql : indexSqls) {
            try {
                jdbcTemplate.execute(indexSql);
                log.debug("Created index for partition {}", partitionName);
            } catch (Exception e) {
                log.warn("Failed to create index for partition {}: {}", partitionName, e.getMessage());
            }
        }
    }

    /**
     * 테이블별로 다른 retention period를 적용하여 오래된 파티션 삭제
     * 일별 파티션: 일(day) 단위 계산
     * 월별 파티션: 월(month) 단위 계산
     */
    private void dropOldPartitionsByTable(LocalDate today) {
        if (!retentionConfig.isEnableAutoCleanup()) {
            log.info("Partition auto cleanup is disabled. Skipping partition cleanup.");
            return;
        }

        log.info("========== Starting Partition Cleanup ==========");
        log.info("Cleanup Date: {}", today);

        try {
            // 일별 파티션 테이블 정리
            log.info("--- Daily Partition Tables ---");
            for (String tableName : QUERY_DAILY_PARTITION_TABLES) {
                int retentionDays = retentionConfig.getRetentionDays(tableName);

                if (retentionDays <= 0) {
                    log.info("[{}] Unlimited retention (days={}). Skipping cleanup.", tableName, retentionDays);
                    continue;
                }

                LocalDate cutoffDate = today.minusDays(retentionDays);
                log.info("[{}] Retention: {} days, Cutoff: {}", tableName, retentionDays, cutoffDate);

                dropPartitionsForTable(tableName, cutoffDate, queryJdbcTemplate, "DB");
            }

            // QueryDB 월별 파티션 테이블 정리
            log.info("");
            log.info("--- QueryDB Monthly Partition Tables ---");
            for (String tableName : MONTHLY_PARTITION_TABLES) {
                int retentionMonths = retentionConfig.getRetentionMonths(tableName);

                if (retentionMonths <= 0) {
                    log.info("[{}] Unlimited retention (months={}). Skipping cleanup.", tableName, retentionMonths);
                    continue;
                }

                // 월 단위 계산: N개월 전의 1일
                LocalDate cutoffDate = today.minusMonths(retentionMonths).withDayOfMonth(1);
                log.info("[{}] Retention: {} months, Cutoff: {}", tableName, retentionMonths, cutoffDate);

                dropPartitionsForTable(tableName, cutoffDate, queryJdbcTemplate, "DB");
            }

            log.info("========== Partition Cleanup Completed ==========");

        } catch (Exception e) {
            log.error("Failed to clean old partitions", e);
        }
    }

    /**
     * 개별 테이블의 파티션 삭제
     */
    private void dropPartitionsForTable(String tableName, LocalDate cutoffDate, JdbcTemplate jdbcTemplate, String dbType) {
        int droppedCount = 0;
        int keptCount = 0;
        int skippedCount = 0;

        // 월별 파티션 테이블인지 확인
        boolean isMonthlyTable = MONTHLY_PARTITION_TABLES.contains(tableName);

        try {
            List<PartitionInfo> partitions = findPartitions("signal", tableName + "_%", jdbcTemplate);

            for (PartitionInfo partition : partitions) {
                if (partition.partitionDate == null) {
                    log.warn("  [{}:{}] Cannot determine partition date, skipping: {}",
                            dbType, tableName, partition.tableName);
                    skippedCount++;
                    continue;
                }

                // 파티션 패턴 검증: 일별 vs 월별
                boolean isDailyPattern = partition.tableName.matches(".*_\\d{6}$");
                boolean isMonthlyPattern = partition.tableName.matches(".*_\\d{4}_\\d{2}$");

                // 일별 파티션 정리 시: 월별 파티션은 건너뛰기
                if (!isMonthlyTable && !isDailyPattern) {
                    log.debug("  [{}:{}] Skipping non-daily partition during daily cleanup: {}",
                            dbType, tableName, partition.tableName);
                    skippedCount++;
                    continue;
                }

                // 월별 파티션 정리 시: 일별 파티션은 건너뛰기
                if (isMonthlyTable && !isMonthlyPattern) {
                    log.debug("  [{}:{}] Skipping non-monthly partition during monthly cleanup: {}",
                            dbType, tableName, partition.tableName);
                    skippedCount++;
                    continue;
                }

                if (partition.partitionDate.isBefore(cutoffDate)) {
                    dropPartition(partition, jdbcTemplate);
                    droppedCount++;
                    log.info("  [{}:{}] Dropped: {} (date: {})",
                            dbType, tableName, partition.tableName, partition.partitionDate);
                } else {
                    keptCount++;
                }
            }

            log.info("  [{}:{}] Result: dropped={}, kept={}, skipped={}",
                    dbType, tableName, droppedCount, keptCount, skippedCount);

        } catch (Exception e) {
            log.error("  [{}:{}] Failed to clean partitions: {}", dbType, tableName, e.getMessage(), e);
        }
    }


    /**
     * 파티션 정보 조회
     */
    private List<PartitionInfo> findPartitions(String schema, String tablePattern, JdbcTemplate jdbcTemplate) {
        String sql = """
            SELECT schemaname, tablename 
            FROM pg_tables 
            WHERE schemaname = ? AND tablename LIKE ?
            ORDER BY tablename
        """;

        return jdbcTemplate.query(sql, (rs, rowNum) -> {
            PartitionInfo info = new PartitionInfo();
            info.schema = rs.getString("schemaname");
            info.tableName = rs.getString("tablename");

            // 날짜 추출
            String tableName = info.tableName;

            // 일별 파티션 (테이블명_YYMMDD 형식)
            if (tableName.matches(".*_\\d{6}$")) {
                String dateStr = tableName.substring(tableName.lastIndexOf('_') + 1);
                try {
                    info.partitionDate = LocalDate.parse("20" + dateStr, DateTimeFormatter.ofPattern("yyyyMMdd"));
                } catch (Exception e) {
                    log.debug("Failed to parse daily date from partition name: {}", tableName);
                }
            }
            // 월별 파티션 (테이블명_YYYY_MM 형식)
            else if (tableName.matches(".*_\\d{4}_\\d{2}$")) {
                String yearMonth = tableName.substring(tableName.lastIndexOf('_', tableName.lastIndexOf('_') - 1) + 1);
                try {
                    info.partitionDate = LocalDate.parse(yearMonth + "_01", DateTimeFormatter.ofPattern("yyyy_MM_dd"));
                } catch (Exception e) {
                    log.debug("Failed to parse monthly date from partition name: {}", tableName);
                }
            }

            return info;
        }, schema, tablePattern);
    }

    /**
     * 파티션 삭제 (개선된 예외 처리)
     */
    private void dropPartition(PartitionInfo partition, JdbcTemplate jdbcTemplate) {
        Integer rowCount = null;

        try {
            // 파티션 존재 여부 확인
            if (!partitionExists(partition.schema, partition.tableName, jdbcTemplate)) {
                log.debug("Partition {}.{} does not exist, skipping drop",
                        partition.schema, partition.tableName);
                return;
            }

            // 삭제 전 데이터 건수 확인
            try {
                String countSql = String.format("SELECT COUNT(*) FROM %s.%s", partition.schema, partition.tableName);
                rowCount = jdbcTemplate.queryForObject(countSql, Integer.class);
            } catch (Exception e) {
                log.warn("Failed to count rows in partition {}.{}, proceeding with drop",
                        partition.schema, partition.tableName);
            }

            // 파티션 삭제 (DROP TABLE IF EXISTS로 안전하게 처리)
            String dropSql = String.format("DROP TABLE IF EXISTS %s.%s CASCADE", partition.schema, partition.tableName);
            jdbcTemplate.execute(dropSql);

            if (rowCount != null) {
                log.info("Dropped partition: {}.{} (contained {} rows)",
                        partition.schema, partition.tableName, rowCount);
            } else {
                log.info("Dropped partition: {}.{}", partition.schema, partition.tableName);
            }

        } catch (org.springframework.dao.DataAccessException e) {
            // PostgreSQL 관련 에러
            log.error("Database error while dropping partition {}.{}: {}",
                    partition.schema, partition.tableName, e.getMessage());
        } catch (Exception e) {
            // 기타 예외
            log.error("Unexpected error while dropping partition {}.{}: {}",
                    partition.schema, partition.tableName, e.getMessage(), e);
        }
    }

    /**
     * 파티션 통계 업데이트
     */
    private void updatePartitionStatistics() {
        try {
            // 주요 테이블만 ANALYZE 실행
            queryJdbcTemplate.execute("ANALYZE signal.t_vessel_tracks_5min");
            queryJdbcTemplate.execute("ANALYZE signal.t_vessel_tracks_hourly");
            queryJdbcTemplate.execute("ANALYZE signal.t_vessel_tracks_daily");
            queryJdbcTemplate.execute("ANALYZE signal.t_tile_summary");
            queryJdbcTemplate.execute("ANALYZE signal.t_area_statistics");

            log.info("Updated partition statistics");
        } catch (Exception e) {
            log.error("Failed to update partition statistics", e);
        }
    }

    /**
     * 파티션 상태 리포트 (queryDB만)
     */
    private void reportPartitionStatus() {
        try {
            // 현재 파티션 상태 조회
            String sql = """
                SELECT
                    schemaname,
                    tablename,
                    pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as size,
                    (SELECT COUNT(*) FROM pg_indexes WHERE tablename = t.tablename) as index_count
                FROM pg_tables t
                WHERE schemaname = 'signal'
                  AND tablename LIKE '%\\_%'
                  AND (tablename LIKE 't_vessel_tracks%'
                       OR tablename LIKE 't_grid%'
                       OR tablename LIKE 't_area%'
                       OR tablename LIKE 't_tile%'
                       OR tablename LIKE 't_abnormal%')
                ORDER BY tablename
            """;

            log.info("=== Partition Status Report (QueryDB) ===");

            List<Map<String, Object>> queryStatus = queryJdbcTemplate.queryForList(sql);

            // 일별 파티션
            log.info("--- Daily Partitions ---");
            queryStatus.stream()
                    .filter(row -> ((String)row.get("tablename")).matches(".*_\\d{6}$"))
                    .forEach(row -> {
                        log.info("  - {}: {} (indexes: {})",
                                row.get("tablename"),
                                row.get("size"),
                                row.get("index_count"));
                    });

            // 월별 파티션
            log.info("--- Monthly Partitions ---");
            queryStatus.stream()
                    .filter(row -> ((String)row.get("tablename")).matches(".*_\\d{4}_\\d{2}$"))
                    .forEach(row -> {
                        log.info("  - {}: {} (indexes: {})",
                                row.get("tablename"),
                                row.get("size"),
                                row.get("index_count"));
                    });

        } catch (Exception e) {
            log.error("Failed to report partition status", e);
        }
    }

    /**
     * 파티션 존재 확인
     */
    private boolean partitionExists(String schema, String partitionName, JdbcTemplate jdbcTemplate) {
        String sql = "SELECT EXISTS (SELECT 1 FROM pg_tables WHERE schemaname = ? AND tablename = ?)";
        Boolean exists = jdbcTemplate.queryForObject(sql, Boolean.class, schema, partitionName);

        if (!Boolean.TRUE.equals(exists)) {
            String dsInfo = DataSourceLogger.getDataSourceInfo(jdbcTemplate.getDataSource());
            log.debug("Partition {}.{} does not exist in: {}", schema, partitionName, dsInfo);
        }

        return Boolean.TRUE.equals(exists);
    }

    // 내부 클래스들
    private static class PartitionTask {
        String schema;
        String baseTable;
        LocalDate date;
        JdbcTemplate jdbcTemplate;
        String partitionType; // "daily" or "monthly"
        String dbType; // "CollectDB" or "QueryDB"

        PartitionTask(String schema, String baseTable, LocalDate date, JdbcTemplate jdbcTemplate, String partitionType, String dbType) {
            this.schema = schema;
            this.baseTable = baseTable;
            this.date = date;
            this.jdbcTemplate = jdbcTemplate;
            this.partitionType = partitionType;
            this.dbType = dbType;
        }
    }

    private static class PartitionInfo {
        String schema;
        String tableName;
        LocalDate partitionDate;
    }
}