package gc.mda.signal_batch.batch.job;

import gc.mda.signal_batch.batch.processor.AisTargetDataProcessor;
import gc.mda.signal_batch.batch.reader.AisTargetDataReader;
import gc.mda.signal_batch.batch.writer.AisTargetCacheWriter;
import gc.mda.signal_batch.domain.vessel.dto.AisTargetDto;
import gc.mda.signal_batch.domain.vessel.model.AisTargetEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
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.beans.factory.annotation.Value;
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.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;

/**
 * AIS Target Import Job 설정
 *
 * 매 1분 실행: S&P AIS API → DTO 변환 → 캐시 저장
 * Chunk Size: 50,000 (API 한 번 호출에 ~33,000건)
 *
 * DB 저장 없음 — 캐시만 업데이트.
 * t_ais_position UPSERT는 Phase 3의 5분 집계 Job에서 편승.
 */
@Slf4j
@Configuration
@Profile("!query")
@ConditionalOnProperty(name = "vessel.batch.scheduler.enabled", havingValue = "true", matchIfMissing = true)
public class AisTargetImportJobConfig {

    private final JobRepository jobRepository;
    private final PlatformTransactionManager transactionManager;
    private final AisTargetDataProcessor processor;
    private final AisTargetCacheWriter writer;
    private final WebClient aisApiWebClient;

    @Value("${app.ais-api.since-seconds:60}")
    private int sinceSeconds;

    @Value("${app.ais-api.chunk-size:50000}")
    private int chunkSize;

    public AisTargetImportJobConfig(
            JobRepository jobRepository,
            @Qualifier("batchTransactionManager") PlatformTransactionManager transactionManager,
            AisTargetDataProcessor processor,
            AisTargetCacheWriter writer,
            @Qualifier("aisApiWebClient") WebClient aisApiWebClient) {
        this.jobRepository = jobRepository;
        this.transactionManager = transactionManager;
        this.processor = processor;
        this.writer = writer;
        this.aisApiWebClient = aisApiWebClient;
    }

    @Bean(name = "aisTargetImportStep")
    public Step aisTargetImportStep() {
        return new StepBuilder("aisTargetImportStep", jobRepository)
                .<AisTargetDto, AisTargetEntity>chunk(chunkSize, transactionManager)
                .reader(new AisTargetDataReader(aisApiWebClient, sinceSeconds))
                .processor(processor)
                .writer(writer)
                .build();
    }

    @Bean(name = "aisTargetImportJob")
    public Job aisTargetImportJob() {
        return new JobBuilder("aisTargetImportJob", jobRepository)
                .start(aisTargetImportStep())
                .listener(new JobExecutionListener() {
                    @Override
                    public void beforeJob(JobExecution jobExecution) {
                        log.info("[aisTargetImportJob] Job 시작");
                    }

                    @Override
                    public void afterJob(JobExecution jobExecution) {
                        log.info("[aisTargetImportJob] Job 완료 - 상태: {}, 처리: {} 건",
                                jobExecution.getStatus(),
                                jobExecution.getStepExecutions().stream()
                                        .mapToLong(se -> se.getWriteCount())
                                        .sum());
                    }
                })
                .build();
    }
}
