#Step 09 — ExecutionContext 와 스코프

ExecutionContext ctx = stepExecution.getExecutionContext();
ctx.putLong("lastOrderId", 41000L);
ctx.putString("mode", "BULK");
long v = ctx.getLong("lastOrderId");

// [9-2] Step 에서 값을 넣고, 다음 Step 에서 읽어 보려는 시도
@Bean
public Tasklet writerTasklet() {
    return (contribution, chunkContext) -> {
        ExecutionContext stepCtx = chunkContext.getStepContext()
                .getStepExecution().getExecutionContext();
        stepCtx.putLong("completedCount", 70000L);
        log.info("[writeStep] stepContext 에 completedCount=70000 저장");
        return RepeatStatus.FINISHED;
    };
}

@Bean
public Tasklet readerTasklet() {
    return (contribution, chunkContext) -> {
        ExecutionContext stepCtx = chunkContext.getStepContext()
                .getStepExecution().getExecutionContext();
        ExecutionContext jobCtx = chunkContext.getStepContext()
                .getStepExecution().getJobExecution().getExecutionContext();
        log.info("[readStep] stepContext.completedCount = {}", stepCtx.get("completedCount"));
        log.info("[readStep] jobContext.completedCount  = {}", jobCtx.get("completedCount"));
        return RepeatStatus.FINISHED;
    };
}

// [9-3] 승격 리스너
@Bean
public ExecutionContextPromotionListener promotionListener() {
    ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
    listener.setKeys(new String[]{"completedCount", "settleDate"});  // 올릴 키 목록
    listener.setStatuses(new String[]{"COMPLETED"});                 // 기본값. 이 ExitStatus 일 때만 승격
    listener.setStrict(false);                                       // 키가 없어도 예외 안 냄
    return listener;
}

@Bean
public Step writeStep(JobRepository jobRepository, PlatformTransactionManager tm) {
    return new StepBuilder("writeStep", jobRepository)
            .tasklet(writerTasklet(), tm)
            .listener(promotionListener())      // ← Step 에 붙입니다. Job 이 아닙니다.
            .build();
}

ExecutionContext jobCtx = chunkContext.getStepContext()
        .getStepExecution().getJobExecution().getExecutionContext();
log.info("[readStep] jobContext.completedCount  = {}", jobCtx.get("completedCount"));

// [9-5] 41,000번째 아이템에서 터지는 Processor
@Bean
public ItemProcessor<Order, Settlement> boobyTrapProcessor() {
    AtomicInteger seen = new AtomicInteger();
    return order -> {
        if (seen.incrementAndGet() == 41_000) {
            throw new IllegalStateException("의도적 실패 — orderId=" + order.order_id());
        }
        return toSettlement(order);
    };
}

// [9-6] StepScope + jobParameters 바인딩
@Bean
@StepScope
public JdbcPagingItemReader<Order> dailyOrderReader(
        DataSource dataSource,
        @Value("#{jobParameters['date']}") String date) {

    MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
    provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
    provider.setFromClause("FROM orders");
    provider.setWhereClause("WHERE status = 'COMPLETED' AND DATE(ordered_at) = :date");
    provider.setSortKeys(Map.of("order_id", Order.ASCENDING));

    return new JdbcPagingItemReaderBuilder<Order>()
            .name("dailyOrderReader")
            .dataSource(dataSource)
            .queryProvider(provider)
            .parameterValues(Map.of("date", date))
            .pageSize(1000)
            .rowMapper(new DataClassRowMapper<>(Order.class))
            .build();
}

// [9-6] 승격된 값을 다음 Step 의 빈이 받아 쓰기
@Bean
@StepScope
public Tasklet reportTasklet(
        @Value("#{jobExecutionContext['completedCount']}") Long completedCount) {
    return (contribution, chunkContext) -> {
        log.info("정산 대상 {}건에 대한 리포트 생성", completedCount);
        return RepeatStatus.FINISHED;
    };
}

@Bean
@StepScope
public JdbcPagingItemReader<Order> partitionedReader(
        @Value("#{stepExecutionContext['minId']}") Long minId,
        @Value("#{stepExecutionContext['maxId']}") Long maxId) { ... }

// [9-7] 잘못된 코드 — @StepScope 가 없습니다
@Bean
// @StepScope   ← 지웠습니다
public JdbcPagingItemReader<Order> brokenReader(
        DataSource dataSource,
        @Value("#{jobParameters['date']}") String date) {
    ...
}

@Scope(value = "step", proxyMode = ScopedProxyMode.TARGET_CLASS)
public @interface StepScope { }

// [9-8] 위험한 선언 — 리턴 타입이 인터페이스
@Bean
@StepScope
public ItemReader<Order> leakyReader(DataSource ds,
        @Value("#{jobParameters['date']}") String date) {
    return new JdbcCursorItemReaderBuilder<Order>()
            .name("leakyReader")
            .dataSource(ds)
            .sql("SELECT order_id, customer_id, amount, status, ordered_at FROM orders "
               + "WHERE status='COMPLETED' AND DATE(ordered_at)=?")
            .queryArguments(date)
            .rowMapper(new DataClassRowMapper<>(Order.class))
            .build();
}

// [9-8] 올바른 선언
@Bean
@StepScope
public JdbcCursorItemReader<Order> safeReader(...) { ... }   // ← 구현체 타입

// [9-9] 생성 횟수를 세는 스코프 빈
private final AtomicInteger stepScopedCount = new AtomicInteger();
private final AtomicInteger jobScopedCount  = new AtomicInteger();

@Bean
@StepScope
public StepScopedProbe stepProbe(@Value("#{stepExecution.stepName}") String stepName) {
    log.info(">>> @StepScope 빈 생성 #{} (step={})", stepScopedCount.incrementAndGet(), stepName);
    return new StepScopedProbe(stepName);
}

@Bean
@JobScope
public JobScopedProbe jobProbe(@Value("#{jobExecution.jobInstance.jobName}") String jobName) {
    log.info(">>> @JobScope  빈 생성 #{} (job={})", jobScopedCount.incrementAndGet(), jobName);
    return new JobScopedProbe(jobName);
}

package com.example.batch.step09;

/*
 * ============================================================================
 *  Step 09 — ExecutionContext 와 스코프 : 실습 파일
 * ============================================================================
 *
 *  실행 방법
 *    ./gradlew bootRun --args='--spring.batch.job.name=step09Job date=2025-03-01'
 *
 *  주의
 *    - BrokenScopeConfig(9-7) / LeakyProxyConfig(9-8) 은 @Configuration 이 주석
 *      처리되어 있습니다. 재현할 때만 주석을 푸세요.
 *      9-7 은 "애플리케이션 기동 실패"이므로 풀어 둔 채로는 다른 실습이 전부 막힙니다.
 *    - Step09Job 실행 전에 TRUNCATE TABLE settlement; 를 권합니다.
 *      (settlement.uk_settlement_order UNIQUE 때문에 재실행 시 중복 키가 납니다)
 * ============================================================================
 */

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobScope;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.listener.ExecutionContextPromotionListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.DataClassRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

public final class Practice {

    private Practice() {
    }

    // ========================================================================
    // [9-2] Job Context 와 Step Context 는 서로 안 보입니다
    //       — 승격 없이 Step Context 에만 넣으면 다음 Step 에서 null 입니다.
    // ========================================================================
    @Configuration
    static class ContextScopeConfig {

        private static final Logger log = LoggerFactory.getLogger(ContextScopeConfig.class);

        @Bean
        Job contextScopeJob(JobRepository jobRepository,
                            Step csWriteStep, Step csReadStep) {
            return new JobBuilder("contextScopeJob", jobRepository)
                    .start(csWriteStep)
                    .next(csReadStep)
                    .build();
        }

        @Bean
        Step csWriteStep(JobRepository jobRepository, PlatformTransactionManager tm) {
            return new StepBuilder("writeStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        ExecutionContext stepCtx = chunkContext.getStepContext()
                                .getStepExecution().getExecutionContext();
                        stepCtx.putLong("completedCount", 70000L);
                        log.info("[writeStep] stepContext 에 completedCount=70000 저장");
                        return RepeatStatus.FINISHED;
                    }, tm)
                    .build();
        }

        @Bean
        Step csReadStep(JobRepository jobRepository, PlatformTransactionManager tm) {
            return new StepBuilder("readStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        ExecutionContext stepCtx = chunkContext.getStepContext()
                                .getStepExecution().getExecutionContext();
                        ExecutionContext jobCtx = chunkContext.getStepContext()
                                .getStepExecution().getJobExecution().getExecutionContext();
                        // 둘 다 null 입니다. 이것이 이 절의 결론입니다.
                        log.info("[readStep] stepContext.completedCount = {}", stepCtx.get("completedCount"));
                        log.info("[readStep] jobContext.completedCount  = {}", jobCtx.get("completedCount"));
                        return RepeatStatus.FINISHED;
                    }, tm)
                    .build();
        }
    }

    // ========================================================================
    // [9-3] ExecutionContextPromotionListener — Step -> Job 승격
    // ========================================================================
    @Configuration
    static class PromotionConfig {

        private static final Logger log = LoggerFactory.getLogger(PromotionConfig.class);

        /**
         * 승격 리스너. Job 이 아니라 <b>Step</b> 에 붙여야 합니다.
         * afterStep 콜백에서 동작하므로, Step 이 끝나야 Job Context 에 반영됩니다.
         */
        @Bean
        ExecutionContextPromotionListener promotionListener() {
            ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
            listener.setKeys(new String[]{"completedCount", "settleDate"});
            listener.setStatuses(new String[]{"COMPLETED"});  // 기본값과 동일. 명시하는 습관을 권합니다.
            listener.setStrict(false);                        // 키가 없어도 예외를 내지 않습니다.
            return listener;
        }

        @Bean
        Job promotionJob(JobRepository jobRepository, Step pmWriteStep, Step pmReadStep) {
            return new JobBuilder("promotionJob", jobRepository)
                    .start(pmWriteStep)
                    .next(pmReadStep)
                    .build();
        }

        @Bean
        Step pmWriteStep(JobRepository jobRepository, PlatformTransactionManager tm,
                         ExecutionContextPromotionListener promotionListener) {
            return new StepBuilder("writeStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        ExecutionContext stepCtx = chunkContext.getStepContext()
                                .getStepExecution().getExecutionContext();
                        stepCtx.putLong("completedCount", 70000L);
                        stepCtx.putString("settleDate", "2025-03-01");
                        log.info("[writeStep] stepContext 에 completedCount=70000 저장");
                        return RepeatStatus.FINISHED;
                    }, tm)
                    .listener(promotionListener)   // ← 여기가 핵심
                    .build();
        }

        @Bean
        Step pmReadStep(JobRepository jobRepository, PlatformTransactionManager tm) {
            return new StepBuilder("readStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        ExecutionContext stepCtx = chunkContext.getStepContext()
                                .getStepExecution().getExecutionContext();
                        ExecutionContext jobCtx = chunkContext.getStepContext()
                                .getStepExecution().getJobExecution().getExecutionContext();
                        log.info("[readStep] stepContext.completedCount = {}", stepCtx.get("completedCount")); // null
                        log.info("[readStep] jobContext.completedCount  = {}", jobCtx.get("completedCount"));  // 70000
                        return RepeatStatus.FINISHED;
                    }, tm)
                    .build();
        }
    }

    // ========================================================================
    // [9-5] ExecutionContext 는 재시작의 근거입니다
    //       41,000번째 아이템에서 일부러 실패시키고, 같은 파라미터로 재실행합니다.
    // ========================================================================
    @Configuration
    static class RestartConfig {

        private static final Logger log = LoggerFactory.getLogger(RestartConfig.class);

        @Bean
        Job restartDemoJob(JobRepository jobRepository, Step settleStepWithTrap) {
            return new JobBuilder("restartDemoJob", jobRepository)
                    .start(settleStepWithTrap)
                    .build();
        }

        @Bean
        Step settleStepWithTrap(JobRepository jobRepository, PlatformTransactionManager tm,
                                JdbcPagingItemReader<Order> allCompletedReader,
                                ItemProcessor<Order, Settlement> boobyTrapProcessor,
                                JdbcBatchItemWriter<Settlement> settlementWriter) {
            return new StepBuilder("settleStep", jobRepository)
                    .<Order, Settlement>chunk(1000, tm)
                    .reader(allCompletedReader)
                    .processor(boobyTrapProcessor)
                    .writer(settlementWriter)
                    .build();
        }

        /**
         * 리턴 타입이 <b>구현체</b>입니다. 이래야 @StepScope 프록시가 ItemStream 을 구현하고,
         * SimpleStepBuilder 가 registerStream() 으로 등록해 재시작이 동작합니다. (9-8 참고)
         */
        @Bean
        @StepScope
        JdbcPagingItemReader<Order> allCompletedReader(DataSource dataSource) {
            MySqlPagingQueryProviderStub provider = new MySqlPagingQueryProviderStub();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause("WHERE status = 'COMPLETED'");
            provider.setSortKey("order_id");

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("JdbcPagingItemReader");   // ← 컨텍스트 키 접두사가 됩니다
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider.build());
            reader.setPageSize(1000);
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }

        /**
         * 41,000번째 아이템에서 터집니다.
         * 청크 1,000 이므로 40청크(=40,000건)까지 커밋되고, 41번째 청크가 롤백됩니다.
         * 따라서 컨텍스트의 read.count 는 41000 이 아니라 40000 입니다.
         */
        @Bean
        @StepScope
        ItemProcessor<Order, Settlement> boobyTrapProcessor(JdbcTemplate jdbcTemplate) {
            AtomicInteger seen = new AtomicInteger();
            return order -> {
                if (seen.incrementAndGet() == 41_000) {
                    throw new IllegalStateException("의도적 실패 — orderId=" + order.order_id());
                }
                return toSettlement(order, feeRateOf(jdbcTemplate, order.customerId()));
            };
        }
    }

    // ========================================================================
    // [9-6] @StepScope 와 늦은 바인딩
    // ========================================================================
    @Configuration
    static class ScopeConfig {

        private static final Logger log = LoggerFactory.getLogger(ScopeConfig.class);

        /**
         * #{jobParameters['date']} 는 @StepScope 가 등록한 SpEL 루트에서만 해석됩니다.
         * 이 애노테이션을 지우면 기동 시점에 EL1008E 로 실패합니다. (9-7)
         */
        @Bean
        @StepScope
        JdbcPagingItemReader<Order> dailyOrderReader(
                DataSource dataSource,
                @Value("#{jobParameters['date']}") String date) {

            log.debug("dailyOrderReader 생성 — date={}", date);

            MySqlPagingQueryProviderStub provider = new MySqlPagingQueryProviderStub();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause("WHERE status = 'COMPLETED' AND DATE(ordered_at) = :date");
            provider.setSortKey("order_id");

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("dailyOrderReader");
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider.build());
            reader.setParameterValues(Map.of("date", date));
            reader.setPageSize(1000);
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }

        /**
         * 승격된 Job Context 값을 SpEL 로 받습니다.
         * 앞 Step 에 ExecutionContextPromotionListener 가 붙어 있어야 값이 존재합니다.
         */
        @Bean
        @StepScope
        org.springframework.batch.core.step.tasklet.Tasklet reportTasklet(
                @Value("#{jobExecutionContext['completedCount']}") Long completedCount) {
            return (contribution, chunkContext) -> {
                log.info("정산 대상 {}건에 대한 리포트 생성", completedCount);
                return RepeatStatus.FINISHED;
            };
        }

        /**
         * 파티셔닝(Step 13)에서 쓰는 형태입니다.
         * stepExecutionContext 는 @StepScope 에서만 해석됩니다. @JobScope 에서는 EL1008E 입니다.
         */
        @Bean
        @StepScope
        JdbcPagingItemReader<Order> partitionedReader(
                DataSource dataSource,
                @Value("#{stepExecutionContext['minId']}") Long minId,
                @Value("#{stepExecutionContext['maxId']}") Long maxId) {

            log.debug("partitionedReader 생성 — minId={}, maxId={}", minId, maxId);

            MySqlPagingQueryProviderStub provider = new MySqlPagingQueryProviderStub();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause("WHERE status='COMPLETED' AND order_id BETWEEN :minId AND :maxId");
            provider.setSortKey("order_id");

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("partitionedReader");
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider.build());
            reader.setParameterValues(Map.of("minId", minId, "maxId", maxId));
            reader.setPageSize(1000);
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }
    }

    // ========================================================================
    // [9-7] @StepScope 누락 — 기동 시점에 EL1008E
    //
    //   재현하려면 아래 @Configuration 주석을 푸세요.
    //   푼 채로 두면 애플리케이션이 아예 뜨지 않아 다른 실습이 전부 막힙니다.
    //
    //   Caused by: org.springframework.expression.spel.SpelEvaluationException:
    //     EL1008E: Property or field 'jobParameters' cannot be found on object of
    //     type 'org.springframework.beans.factory.config.BeanExpressionContext'
    // ========================================================================
    // @Configuration
    static class BrokenScopeConfig {

        @Bean
        // @StepScope   ← 이 한 줄이 없어서 실패합니다
        JdbcPagingItemReader<Order> brokenReader(
                DataSource dataSource,
                @Value("#{jobParameters['date']}") String date) {

            MySqlPagingQueryProviderStub provider = new MySqlPagingQueryProviderStub();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause("WHERE status='COMPLETED' AND DATE(ordered_at) = :date");
            provider.setSortKey("order_id");

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("brokenReader");
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider.build());
            reader.setParameterValues(Map.of("date", date));
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }
    }

    // ========================================================================
    // [9-8] 프록시 함정 — 리턴 타입을 인터페이스로 쓰면 ItemStream 이 사라집니다
    //
    //   @StepScope == @Scope(value="step", proxyMode=ScopedProxyMode.TARGET_CLASS)
    //   프록시가 무엇을 구현할지는 @Bean 메서드의 "리턴 타입"이 정합니다.
    //
    //   재현하려면 아래 @Configuration 주석을 푸세요.
    // ========================================================================
    // @Configuration
    static class LeakyProxyConfig {

        /**
         * (A) 위험 — 리턴 타입이 인터페이스입니다.
         *     프록시가 ItemStream 을 구현하지 않아 open()/update()/close() 가 호출되지 않습니다.
         *     JdbcCursorItemReader 는 open() 에서 ResultSet 을 열므로 첫 read() 에서 터집니다.
         *
         *     org.springframework.batch.item.ReaderNotOpenException:
         *       Reader must be open before it can be read.
         */
        @Bean
        @StepScope
        ItemReader<Order> leakyReader(DataSource ds,
                                      @Value("#{jobParameters['date']}") String date) {
            JdbcCursorItemReader<Order> reader = new JdbcCursorItemReader<>();
            reader.setName("leakyReader");
            reader.setDataSource(ds);
            reader.setSql("SELECT order_id, customer_id, amount, status, ordered_at FROM orders "
                    + "WHERE status='COMPLETED' AND DATE(ordered_at)=?");
            reader.setPreparedStatementSetter(ps -> ps.setString(1, date));
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }

        /**
         * (B) 정답 — 리턴 타입을 구현체로 선언합니다.
         *     프록시가 JdbcCursorItemReader 를 상속하므로 ItemStream 도 함께 구현합니다.
         */
        @Bean
        @StepScope
        JdbcCursorItemReader<Order> safeReader(DataSource ds,
                                               @Value("#{jobParameters['date']}") String date) {
            JdbcCursorItemReader<Order> reader = new JdbcCursorItemReader<>();
            reader.setName("safeReader");
            reader.setDataSource(ds);
            reader.setSql("SELECT order_id, customer_id, amount, status, ordered_at FROM orders "
                    + "WHERE status='COMPLETED' AND DATE(ordered_at)=?");
            reader.setPreparedStatementSetter(ps -> ps.setString(1, date));
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }

        /**
         * (C) 차선책 — 인터페이스로 노출하되 .stream(...) 으로 명시 등록합니다.
         *     결국 ItemStream 타입을 알아야 하므로 (B) 가 더 낫습니다.
         */
        @Bean
        Step explicitStreamStep(JobRepository jobRepository, PlatformTransactionManager tm,
                                ItemReader<Order> leakyReader,
                                ItemProcessor<Order, Settlement> passThrough,
                                JdbcBatchItemWriter<Settlement> settlementWriter) {
            return new StepBuilder("explicitStreamStep", jobRepository)
                    .<Order, Settlement>chunk(1000, tm)
                    .reader(leakyReader)
                    .processor(passThrough)
                    .writer(settlementWriter)
                    .stream((org.springframework.batch.item.ItemStream) leakyReader)  // ← 명시 등록
                    .build();
        }
    }

    // ========================================================================
    // [9-9] 스코프 빈이 몇 번 만들어지는가
    //       @StepScope = StepExecution 마다, @JobScope = JobExecution 마다
    // ========================================================================
    @Configuration
    static class ProbeConfig {

        private static final Logger log = LoggerFactory.getLogger(ProbeConfig.class);

        // 싱글턴 설정 클래스의 필드이므로 애플리케이션 단위로 누적됩니다.
        private final AtomicInteger stepScopedCount = new AtomicInteger();
        private final AtomicInteger jobScopedCount = new AtomicInteger();

        record StepScopedProbe(String stepName) {
        }

        record JobScopedProbe(String jobName) {
        }

        @Bean
        @StepScope
        StepScopedProbe stepProbe(@Value("#{stepExecution.stepName}") String stepName) {
            log.info(">>> @StepScope 빈 생성 #{} (step={})", stepScopedCount.incrementAndGet(), stepName);
            return new StepScopedProbe(stepName);
        }

        @Bean
        @JobScope
        JobScopedProbe jobProbe(@Value("#{jobExecution.jobInstance.jobName}") String jobName) {
            log.info(">>> @JobScope  빈 생성 #{} (job={})", jobScopedCount.incrementAndGet(), jobName);
            return new JobScopedProbe(jobName);
        }

        @Bean
        Job scopeProbeJob(JobRepository jobRepository,
                          Step probeStepA, Step probeStepB, Step probeStepC) {
            return new JobBuilder("scopeProbeJob", jobRepository)
                    .start(probeStepA).next(probeStepB).next(probeStepC)
                    .build();
        }

        @Bean
        Step probeStepA(JobRepository r, PlatformTransactionManager tm,
                        StepScopedProbe stepProbe, JobScopedProbe jobProbe) {
            return probeStep("probeStepA", r, tm, stepProbe, jobProbe);
        }

        @Bean
        Step probeStepB(JobRepository r, PlatformTransactionManager tm,
                        StepScopedProbe stepProbe, JobScopedProbe jobProbe) {
            return probeStep("probeStepB", r, tm, stepProbe, jobProbe);
        }

        @Bean
        Step probeStepC(JobRepository r, PlatformTransactionManager tm,
                        StepScopedProbe stepProbe, JobScopedProbe jobProbe) {
            return probeStep("probeStepC", r, tm, stepProbe, jobProbe);
        }

        private Step probeStep(String name, JobRepository r, PlatformTransactionManager tm,
                               StepScopedProbe stepProbe, JobScopedProbe jobProbe) {
            return new StepBuilder(name, r)
                    .tasklet((contribution, chunkContext) -> {
                        // 프록시가 실제로 대상 객체를 만드는 시점은 "이 호출"입니다.
                        log.debug("{} — stepProbe={}, jobProbe={}", name, stepProbe.stepName(), jobProbe.jobName());
                        return RepeatStatus.FINISHED;
                    }, tm)
                    .build();
        }
    }

    // ========================================================================
    // [9-10] 종합 — 승격 + 스코프로 세 Step 잇기
    //
    //   prepareStep --(completedCount, settleDate)--> settleStep --(feeTotal)--> reportStep
    // ========================================================================
    @Configuration
    static class Step09Job {

        private static final Logger log = LoggerFactory.getLogger(Step09Job.class);

        @Bean
        Job step09Job(JobRepository jobRepository,
                      Step prepareStep, Step s09SettleStep, Step s09ReportStep) {
            return new JobBuilder("step09Job", jobRepository)
                    .start(prepareStep)
                    .next(s09SettleStep)
                    .next(s09ReportStep)
                    .build();
        }

        // ── 승격 리스너 두 개 (올릴 키가 다르므로 따로 만듭니다)
        @Bean
        ExecutionContextPromotionListener preparePromotion() {
            ExecutionContextPromotionListener l = new ExecutionContextPromotionListener();
            l.setKeys(new String[]{"completedCount", "settleDate"});
            l.setStatuses(new String[]{"COMPLETED"});
            return l;
        }

        @Bean
        ExecutionContextPromotionListener settlePromotion() {
            ExecutionContextPromotionListener l = new ExecutionContextPromotionListener();
            l.setKeys(new String[]{"feeTotal"});
            l.setStatuses(new String[]{"COMPLETED"});
            return l;
        }

        // ── prepareStep : COUNT(*) 를 세어 Step Context 에 넣습니다
        @Bean
        Step prepareStep(JobRepository jobRepository, PlatformTransactionManager tm,
                         JdbcTemplate jdbcTemplate,
                         ExecutionContextPromotionListener preparePromotion) {
            return new StepBuilder("prepareStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        String date = (String) chunkContext.getStepContext()
                                .getJobParameters().get("date");
                        Long cnt = jdbcTemplate.queryForObject(
                                "SELECT COUNT(*) FROM orders WHERE status='COMPLETED' AND DATE(ordered_at)=?",
                                Long.class, date);
                        ExecutionContext ctx = chunkContext.getStepContext()
                                .getStepExecution().getExecutionContext();
                        ctx.putLong("completedCount", cnt == null ? 0L : cnt);
                        ctx.putString("settleDate", date);
                        log.info("[prepare] date={}, COMPLETED 건수={}", date, cnt);
                        return RepeatStatus.FINISHED;
                    }, tm)
                    .listener(preparePromotion)
                    .build();
        }

        // ── settleStep : jobParameters 와 jobExecutionContext 를 둘 다 씁니다
        @Bean
        @StepScope
        JdbcPagingItemReader<Order> settleReader(
                DataSource dataSource,
                @Value("#{jobParameters['date']}") String date,
                @Value("#{jobExecutionContext['completedCount']}") Long expected) {

            log.debug("settleReader 생성 — date={}, 예상 건수={}", date, expected);

            MySqlPagingQueryProviderStub provider = new MySqlPagingQueryProviderStub();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause("WHERE status='COMPLETED' AND DATE(ordered_at) = :date");
            provider.setSortKey("order_id");

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("settleReader");
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider.build());
            reader.setParameterValues(Map.of("date", date));
            reader.setPageSize(1000);
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }

        @Bean
        Step s09SettleStep(JobRepository jobRepository, PlatformTransactionManager tm,
                           JdbcPagingItemReader<Order> settleReader,
                           JdbcTemplate jdbcTemplate,
                           JdbcBatchItemWriter<Settlement> settlementWriter,
                           ExecutionContextPromotionListener settlePromotion) {

            // 수수료 합계를 누적해 Step Context 에 남깁니다.
            // 주의: 누적 상태는 "빈 필드"가 아니라 ExecutionContext 에 남겨야 재시작에 안전합니다.
            return new StepBuilder("settleStep", jobRepository)
                    .<Order, Settlement>chunk(1000, tm)
                    .reader(settleReader)
                    .processor((ItemProcessor<Order, Settlement>) order ->
                            toSettlement(order, feeRateOf(jdbcTemplate, order.customerId())))
                    .writer(chunk -> {
                        settlementWriter.write(chunk);
                        // (실전에서는 StepExecutionListener 로 분리하는 편이 깔끔합니다 — Step 12)
                    })
                    .listener(new FeeTotalListener())
                    .listener(settlePromotion)
                    .build();
        }

        // ── reportStep : 승격된 값만으로 리포트
        @Bean
        Step s09ReportStep(JobRepository jobRepository, PlatformTransactionManager tm,
                           org.springframework.batch.core.step.tasklet.Tasklet s09ReportTasklet) {
            return new StepBuilder("reportStep", jobRepository)
                    .tasklet(s09ReportTasklet, tm)
                    .build();
        }

        @Bean
        @StepScope
        org.springframework.batch.core.step.tasklet.Tasklet s09ReportTasklet(
                @Value("#{jobExecutionContext['settleDate']}") String settleDate,
                @Value("#{jobExecutionContext['completedCount']}") Long completedCount,
                @Value("#{jobExecutionContext['feeTotal']}") BigDecimal feeTotal) {
            return (contribution, chunkContext) -> {
                log.info("[report] {} 정산 {}건, 수수료 합계 = {}", settleDate, completedCount, feeTotal);
                return RepeatStatus.FINISHED;
            };
        }
    }

    /**
     * [9-10] 수수료 합계를 Step ExecutionContext 에 누적하는 리스너.
     * 값을 빈 필드가 아니라 컨텍스트에 두는 이유는 9-9 의 함정 그대로입니다.
     */
    static class FeeTotalListener implements org.springframework.batch.core.StepExecutionListener {

        @Override
        public org.springframework.batch.core.ExitStatus afterStep(
                org.springframework.batch.core.StepExecution stepExecution) {
            // 실제 합계는 Writer 쪽에서 누적하거나 SQL 로 재집계합니다.
            // 여기서는 컨텍스트에 키가 존재하도록 보장하는 역할만 합니다.
            ExecutionContext ctx = stepExecution.getExecutionContext();
            if (!ctx.containsKey("feeTotal")) {
                ctx.put("feeTotal", BigDecimal.ZERO);
            }
            return stepExecution.getExitStatus();
        }
    }

    // ========================================================================
    // 공통 헬퍼
    // ========================================================================

    static Settlement toSettlement(Order order, BigDecimal feeRate) {
        BigDecimal gross = order.amount();
        BigDecimal fee = gross.multiply(feeRate).setScale(2, RoundingMode.HALF_UP);
        return new Settlement(
                order.order_id(),
                order.customerId(),
                order.orderedAt().toLocalDate(),
                gross,
                feeRate,
                fee,
                gross.subtract(fee)
        );
    }

    private static final Map<Integer, BigDecimal> FEE_CACHE = new HashMap<>();

    static BigDecimal feeRateOf(JdbcTemplate jdbcTemplate, Integer customerId) {
        return FEE_CACHE.computeIfAbsent(customerId, id -> jdbcTemplate.queryForObject(
                "SELECT fee_rate FROM customers WHERE customer_id = ?", BigDecimal.class, id));
    }

    /**
     * MySqlPagingQueryProvider 설정을 짧게 쓰기 위한 얇은 래퍼입니다.
     * 실제 프로젝트에서는 org.springframework.batch.item.database.support.MySqlPagingQueryProvider 를
     * 직접 쓰면 됩니다.
     */
    static class MySqlPagingQueryProviderStub {
        private final org.springframework.batch.item.database.support.MySqlPagingQueryProvider delegate =
                new org.springframework.batch.item.database.support.MySqlPagingQueryProvider();

        void setSelectClause(String s) { delegate.setSelectClause(s); }
        void setFromClause(String s) { delegate.setFromClause(s); }
        void setWhereClause(String s) { delegate.setWhereClause(s); }

        void setSortKey(String key) {
            delegate.setSortKeys(Map.of(key, org.springframework.batch.item.database.Order.ASCENDING));
        }

        org.springframework.batch.item.database.support.MySqlPagingQueryProvider build() {
            return delegate;
        }
    }

    // ========================================================================
    // 본문 9-4 / 9-5 / 9-8 에서 쓴 조회 SQL 모음
    //   mysql -h127.0.0.1 -P3308 -ubatch -pbatch1234 batchdb -t -e "..."
    // ========================================================================
    static final String CONTEXT_INSPECT_SQL = """
            -- [9-4] Job ExecutionContext 전문 (SHORT / SERIALIZED 둘 다)
            SELECT JOB_EXECUTION_ID, SHORT_CONTEXT, SERIALIZED_CONTEXT
              FROM BATCH_JOB_EXECUTION_CONTEXT
             ORDER BY JOB_EXECUTION_ID DESC LIMIT 1;

            -- [9-4] 항상 이 형태로 보는 습관을 들이세요
            SELECT JOB_EXECUTION_ID, COALESCE(SERIALIZED_CONTEXT, SHORT_CONTEXT) AS ctx
              FROM BATCH_JOB_EXECUTION_CONTEXT
             ORDER BY JOB_EXECUTION_ID DESC LIMIT 3;

            -- [9-5] 실패 지점과 컨텍스트의 read.count 비교
            SELECT s.STEP_NAME, s.STATUS, s.READ_COUNT, s.WRITE_COUNT, s.COMMIT_COUNT,
                   COALESCE(c.SERIALIZED_CONTEXT, c.SHORT_CONTEXT) AS ctx
              FROM BATCH_STEP_EXECUTION s
              JOIN BATCH_STEP_EXECUTION_CONTEXT c USING (STEP_EXECUTION_ID)
             ORDER BY s.STEP_EXECUTION_ID DESC LIMIT 2;

            -- [9-8] read.count 키가 남았는지 확인 (프록시 함정 점검)
            SELECT COALESCE(SERIALIZED_CONTEXT, SHORT_CONTEXT) LIKE '%read.count%' AS has_read_count
              FROM BATCH_STEP_EXECUTION_CONTEXT
             ORDER BY STEP_EXECUTION_ID DESC LIMIT 1;
            """;
}

package com.example.batch.step09;

/*
 * ============================================================================
 *  Step 09 — ExecutionContext 와 스코프 : 연습문제 (6문제)
 * ============================================================================
 *
 *  각 문제의 "// 여기에 작성:" 자리를 채우세요. 정답은 Solution.java 입니다.
 *
 *  실행 전 준비
 *    mysql -h127.0.0.1 -P3308 -ubatch -pbatch1234 batchdb -e "TRUNCATE TABLE settlement;"
 * ============================================================================
 */

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.listener.ExecutionContextPromotionListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.DataClassRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.util.Map;

public final class Exercise {

    private Exercise() {
    }

    // ========================================================================
    // 문제 1.
    //   countStep 이 "COMPLETED 주문 건수"와 "정산 대상 일자"를 Step Context 에
    //   넣습니다. 다음 Step(useStep) 이 그 값을 읽을 수 있게 만드세요.
    //
    //   요구사항
    //     (a) 승격 리스너 빈을 완성할 것 — 올릴 키는 "orderCount", "targetDate"
    //     (b) Step 이 FAILED 로 끝나도 승격되도록 상태를 명시할 것
    //     (c) 리스너를 올바른 곳(Job? Step?)에 부착할 것
    // ========================================================================
    @Configuration
    static class Q1 {

        private static final Logger log = LoggerFactory.getLogger(Q1.class);

        @Bean
        ExecutionContextPromotionListener q1Promotion() {
            ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
            // 여기에 작성: setKeys / setStatuses / setStrict 를 지정하세요
            return listener;
        }

        @Bean
        Step q1CountStep(JobRepository jobRepository, PlatformTransactionManager tm,
                         JdbcTemplate jdbcTemplate,
                         ExecutionContextPromotionListener q1Promotion) {
            return new StepBuilder("q1CountStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        Long cnt = jdbcTemplate.queryForObject(
                                "SELECT COUNT(*) FROM orders WHERE status='COMPLETED'", Long.class);
                        ExecutionContext ctx = chunkContext.getStepContext()
                                .getStepExecution().getExecutionContext();
                        ctx.putLong("orderCount", cnt == null ? 0L : cnt);
                        ctx.putString("targetDate", "2025-03-01");
                        log.info("[q1CountStep] orderCount={}", cnt);
                        return RepeatStatus.FINISHED;
                    }, tm)
                    // 여기에 작성: 승격 리스너를 부착하세요
                    .build();
        }

        @Bean
        Step q1UseStep(JobRepository jobRepository, PlatformTransactionManager tm) {
            return new StepBuilder("q1UseStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        // 여기에 작성: Job ExecutionContext 에서 orderCount 를 꺼내 로그로 찍으세요
                        // 기대 출력: [q1UseStep] orderCount = 70000
                        return RepeatStatus.FINISHED;
                    }, tm)
                    .build();
        }

        @Bean
        Job q1Job(JobRepository jobRepository, Step q1CountStep, Step q1UseStep) {
            return new JobBuilder("q1Job", jobRepository)
                    .start(q1CountStep).next(q1UseStep).build();
        }
    }

    // ========================================================================
    // 문제 2.
    //   jobParameters 의 'date' 로 하루치 COMPLETED 주문만 읽는 Reader 를 완성하세요.
    //
    //   확인 방법
    //     ./gradlew bootRun --args='--spring.batch.job.name=q2Job date=2025-03-01'
    //     → READ_COUNT 가 389 여야 정답입니다.
    // ========================================================================
    @Configuration
    static class Q2 {

        @Bean
        // 여기에 작성: 이 Reader 가 jobParameters 를 볼 수 있게 하는 애노테이션
        JdbcPagingItemReader<Order> q2Reader(
                DataSource dataSource,
                /* 여기에 작성: date 파라미터를 받는 @Value 표현식 */ String date) {

            MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            // 여기에 작성: status 와 날짜 조건을 담은 WHERE 절 (named parameter :date 사용)
            provider.setSortKeys(Map.of("order_id", org.springframework.batch.item.database.Order.ASCENDING));

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("q2Reader");
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider);
            // 여기에 작성: named parameter 값 바인딩
            reader.setPageSize(1000);
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }
    }

    // ========================================================================
    // 문제 3.  [진단 문제 — 코드를 고치기 전에 먼저 실행해 보세요]
    //
    //   아래 설정을 활성화하면 애플리케이션이 아예 뜨지 않습니다.
    //
    //   (a) 어떤 예외가 나는지 클래스명과 메시지 코드를 적으세요.
    //   (b) 왜 "Job 실행 시점"이 아니라 "기동 시점"에 터지는지 설명하세요.
    //   (c) 코드를 고치세요.
    // ========================================================================
    // @Configuration      ← 재현할 때만 주석을 푸세요
    static class Q3Broken {

        @Bean
        JdbcPagingItemReader<Order> q3Reader(
                DataSource dataSource,
                @Value("#{jobParameters['date']}") String date) {

            MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause("WHERE status='COMPLETED' AND DATE(ordered_at) = :date");
            provider.setSortKeys(Map.of("order_id", org.springframework.batch.item.database.Order.ASCENDING));

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("q3Reader");
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider);
            reader.setParameterValues(Map.of("date", date));
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }

        // (a) 예외 클래스 / 메시지 코드
        //     여기에 작성:
        //
        // (b) 왜 기동 시점인가
        //     여기에 작성:
        //
        // (c) 고친 코드
        //     여기에 작성:
    }

    // ========================================================================
    // 문제 4.  [SQL 문제]
    //
    //   가장 최근 JobExecution 의 Job ExecutionContext 에 저장된 JSON 을
    //   "잘리지 않은 전문"으로 조회하는 SQL 을 쓰세요.
    //
    //   힌트: 이 테이블에는 컨텍스트 컬럼이 두 개 있고, 작은 컨텍스트는
    //         한쪽이 NULL 입니다.
    // ========================================================================
    static final String Q4_SQL = """
            -- 여기에 작성:
            """;

    // ========================================================================
    // 문제 5.
    //   아래 Reader 는 실행하면 "정상적으로 성공"합니다. 그런데 위험합니다.
    //
    //   (a) 무엇이 조용히 망가져 있습니까?
    //   (b) 그것을 SQL 로 확인하는 방법을 쓰세요.
    //   (c) 두 가지 방법으로 고치세요. 어느 쪽이 권장인지도 적으세요.
    // ========================================================================
    @Configuration
    static class Q5 {

        @Bean
        @StepScope
        ItemReader<Order> q5LeakyReader(
                DataSource dataSource,
                @Value("#{jobParameters['date']}") String date) {

            MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause("WHERE status='COMPLETED' AND DATE(ordered_at) = :date");
            provider.setSortKeys(Map.of("order_id", org.springframework.batch.item.database.Order.ASCENDING));

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("q5Reader");
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider);
            reader.setParameterValues(Map.of("date", date));
            reader.setPageSize(1000);
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }

        // (a) 무엇이 망가졌나
        //     여기에 작성:
        //
        // (b) 확인용 SQL
        //     여기에 작성:
        //
        // (c) 해법 1 (권장) / 해법 2
        //     여기에 작성:

        @Bean
        Step q5Step(JobRepository jobRepository, PlatformTransactionManager tm,
                    ItemReader<Order> q5LeakyReader,
                    ItemProcessor<Order, Settlement> q5Processor,
                    JdbcBatchItemWriter<Settlement> settlementWriter) {
            return new StepBuilder("q5Step", jobRepository)
                    .<Order, Settlement>chunk(1000, tm)
                    .reader(q5LeakyReader)
                    .processor(q5Processor)
                    .writer(settlementWriter)
                    // 여기에 작성: (해법 2 를 택한다면) 필요한 한 줄
                    .build();
        }
    }

    // ========================================================================
    // 문제 6.  [판단 문제 — 코드 없음]
    //
    //   아래 세 가지 빈에 각각 무슨 스코프를 붙여야 합니까?
    //   (싱글턴 / @JobScope / @StepScope 중 하나) 그리고 그 이유를 쓰세요.
    //
    //   (1) jobParameters['inputFile'] 로 경로를 받는 FlatFileItemReader
    //       답:    여기에 작성:
    //       이유:  여기에 작성:
    //
    //   (2) stepExecutionContext['minId'] / ['maxId'] 로 구간을 받는
    //       파티션 전용 JdbcPagingItemReader
    //       답:    여기에 작성:
    //       이유:  여기에 작성:
    //
    //   (3) Job 이 도는 동안 전체 Step 의 처리 건수를 누적해 마지막에
    //       한 번 리포트하는 StatsCollector
    //       답:    여기에 작성:
    //       이유:  여기에 작성:
    //       주의:  @StepScope 를 고르면 어떤 일이 벌어집니까?
    //              여기에 작성:
    // ========================================================================
}

package com.example.batch.step09;

/*
 * ============================================================================
 *  Step 09 — ExecutionContext 와 스코프 : 정답과 해설
 * ============================================================================
 *  문제를 직접 풀어 본 뒤에 여세요.
 * ============================================================================
 */

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobScope;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.listener.ExecutionContextPromotionListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.DataClassRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

public final class Solution {

    private Solution() {
    }

    // ========================================================================
    // 정답 1.
    //
    //  핵심은 세 가지입니다.
    //
    //  (a) setKeys — 올릴 키를 "명시"해야 합니다. 전체 복사 같은 옵션은 없습니다.
    //      의도적으로 그렇게 설계되어 있습니다. Step Context 에는 프레임워크가
    //      넣은 batch.stepType / batch.taskletType 같은 키도 있는데, 그것까지
    //      Job Context 로 올라가면 재시작 검증이 엉킵니다.
    //
    //  (b) setStatuses — 여기가 이 문제의 진짜 함정입니다.
    //      기본값은 { "COMPLETED" } 뿐입니다. 즉 Step 이 FAILED 로 끝나면
    //      "아무 예외 없이" 승격이 생략됩니다.
    //      그러면 다음 Step 은 jobContext.get("orderCount") 로 null 을 받고,
    //      그것을 long 으로 언박싱하는 순간 전혀 상관없는 자리에서
    //      NullPointerException 이 납니다. 원인을 찾기가 아주 어렵습니다.
    //      → 부분 실패에서도 값이 필요하면 "FAILED" 를 명시하세요.
    //
    //  (c) setStrict — true 로 두면 지정한 키가 Step Context 에 없을 때
    //      IllegalArgumentException 을 던집니다. 조건부로만 넣는 키에는
    //      절대 쓰면 안 됩니다. 반대로 "반드시 있어야 하는" 키라면
    //      strict=true 가 조용한 null 보다 낫습니다.
    //
    //  (d) 리스너는 Job 이 아니라 Step 에 붙입니다.
    //      ExecutionContextPromotionListener 는 StepExecutionListener 이고,
    //      afterStep 콜백에서 동작하기 때문입니다.
    //      JobBuilder.listener(...) 에 넣으면 아무 일도 일어나지 않습니다.
    // ========================================================================
    @Configuration
    static class A1 {

        private static final Logger log = LoggerFactory.getLogger(A1.class);

        @Bean
        ExecutionContextPromotionListener a1Promotion() {
            ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
            listener.setKeys(new String[]{"orderCount", "targetDate"});
            listener.setStatuses(new String[]{"COMPLETED", "FAILED"});  // ← (b)
            listener.setStrict(false);                                  // ← (c)
            return listener;
        }

        @Bean
        Step a1CountStep(JobRepository jobRepository, PlatformTransactionManager tm,
                         JdbcTemplate jdbcTemplate,
                         ExecutionContextPromotionListener a1Promotion) {
            return new StepBuilder("a1CountStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        Long cnt = jdbcTemplate.queryForObject(
                                "SELECT COUNT(*) FROM orders WHERE status='COMPLETED'", Long.class);
                        ExecutionContext ctx = chunkContext.getStepContext()
                                .getStepExecution().getExecutionContext();
                        ctx.putLong("orderCount", cnt == null ? 0L : cnt);
                        ctx.putString("targetDate", "2025-03-01");
                        log.info("[a1CountStep] orderCount={}", cnt);   // 70000
                        return RepeatStatus.FINISHED;
                    }, tm)
                    .listener(a1Promotion)     // ← (d) Step 에 부착
                    .build();
        }

        @Bean
        Step a1UseStep(JobRepository jobRepository, PlatformTransactionManager tm) {
            return new StepBuilder("a1UseStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        ExecutionContext jobCtx = chunkContext.getStepContext()
                                .getStepExecution().getJobExecution().getExecutionContext();

                        // 방어적으로 읽습니다. containsKey 검사 없이 getLong 하면
                        // 키가 없을 때 0 이 반환되어 "0건 처리" 로 조용히 틀립니다.
                        if (!jobCtx.containsKey("orderCount")) {
                            throw new IllegalStateException("orderCount 가 승격되지 않았습니다");
                        }
                        log.info("[a1UseStep] orderCount = {}", jobCtx.getLong("orderCount"));
                        return RepeatStatus.FINISHED;
                    }, tm)
                    .build();
        }

        @Bean
        Job a1Job(JobRepository jobRepository, Step a1CountStep, Step a1UseStep) {
            return new JobBuilder("a1Job", jobRepository)
                    .start(a1CountStep).next(a1UseStep).build();
        }
    }

    // ========================================================================
    // 정답 2.
    //
    //  @StepScope 가 있어야 #{jobParameters['date']} 가 해석됩니다.
    //  그리고 리턴 타입은 JdbcPagingItemReader<Order> — 구현체여야 합니다.
    //  (인터페이스로 쓰면 정답 5 의 문제가 그대로 재현됩니다.)
    //
    //  WHERE 절에는 named parameter(:date)를 쓰고, 실제 값은
    //  setParameterValues 로 바인딩합니다.
    //  문자열을 직접 이어 붙이지 마세요 —
    //    provider.setWhereClause("... = '" + date + "'")
    //  는 SQL 인젝션이자, 바인드 변수를 못 써서 MySQL 쿼리 캐시/플랜 재사용도
    //  못 하게 만듭니다.
    //
    //  검증: READ_COUNT 가 389 여야 합니다.
    //    하루 주문 약 555건 x COMPLETED 70% = 389건.
    //    이 숫자는 프로젝트 시드가 결정론적이라 항상 같습니다.
    // ========================================================================
    @Configuration
    static class A2 {

        @Bean
        @StepScope                                     // ← 정답의 절반
        JdbcPagingItemReader<Order> a2Reader(
                DataSource dataSource,
                @Value("#{jobParameters['date']}") String date) {

            MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause("WHERE status = 'COMPLETED' AND DATE(ordered_at) = :date");
            provider.setSortKeys(Map.of("order_id", org.springframework.batch.item.database.Order.ASCENDING));

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("a2Reader");
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider);
            reader.setParameterValues(Map.of("date", date));
            reader.setPageSize(1000);
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }
    }

    /*
     * ========================================================================
     * 정답 3.
     *
     *  (a) 예외
     *      org.springframework.beans.factory.BeanExpressionException:
     *        Expression parsing failed
     *      Caused by: org.springframework.expression.spel.SpelEvaluationException:
     *        EL1008E: Property or field 'jobParameters' cannot be found on object
     *        of type 'org.springframework.beans.factory.config.BeanExpressionContext'
     *
     *      기억할 문자열은 "EL1008E" 와 "jobParameters cannot be found" 입니다.
     *
     *  (b) 왜 기동 시점인가
     *      @StepScope 가 없으면 그 빈은 싱글턴입니다. 싱글턴은 애플리케이션
     *      컨텍스트가 refresh 될 때, 즉 "Job 이 시작되기 훨씬 전"에 생성됩니다.
     *      그 시점의 SpEL 평가 루트는 BeanExpressionContext 이고,
     *      거기에는 jobParameters 라는 프로퍼티가 존재하지 않습니다.
     *
     *      jobParameters / jobExecutionContext / stepExecutionContext 는
     *      StepScope · JobScope 가 빈을 만들 때 SpEL 컨텍스트에 등록해 주는
     *      것들입니다. 스코프가 없으면 등록하는 주체 자체가 없습니다.
     *
     *      타임라인:
     *        [애플리케이션 기동] 싱글턴 생성 ── jobParameters 없음 → EL1008E
     *        [Job 실행]         파라미터 확정
     *        [Step 시작]        StepScope 가 빈 생성 ── 여기서는 있음
     *
     *  (c) 고친 코드 = A2 와 동일합니다. @StepScope 한 줄을 붙이면 됩니다.
     *
     *  참고 — 어떤 스코프가 어떤 표현식을 볼 수 있는가
     *
     *    표현식                          싱글턴   @JobScope   @StepScope
     *    #{jobParameters[...]}             X         O            O
     *    #{jobExecutionContext[...]}       X         O            O
     *    #{stepExecutionContext[...]}      X         X            O
     *    #{stepExecution...}               X         X            O
     *    #{jobExecution...}                X         O            O
     *
     *  @JobScope 빈에 #{stepExecutionContext[...]} 를 쓰면 역시 EL1008E 입니다.
     *  당연합니다 — JobScope 빈이 만들어지는 시점에는 아직 StepExecution 이
     *  없기 때문입니다.
     *
     *  마지막으로, 이 에러는 "착한 에러"입니다. 애플리케이션이 아예 뜨지 않아
     *  100% 발견됩니다. 정답 5 의 함정은 그렇지 않습니다.
     * ========================================================================
     */

    // ========================================================================
    // 정답 4.
    //
    //  BATCH_JOB_EXECUTION_CONTEXT 에는 컨텍스트 컬럼이 두 개 있습니다.
    //
    //    SHORT_CONTEXT       VARCHAR(2500)  항상 채워짐. 넘치면 앞부분만 잘려 들어감
    //    SERIALIZED_CONTEXT  TEXT           2500자를 "넘을 때만" 채워짐
    //
    //  JdbcExecutionContextDao 는 읽을 때 SERIALIZED_CONTEXT 가 있으면 그것을,
    //  없으면 SHORT_CONTEXT 를 씁니다. 따라서 사람이 조회할 때도 같은 규칙을
    //  따라야 "전문"을 보게 됩니다.
    //
    //  SERIALIZED_CONTEXT 만 보고 "비어 있네" 하는 실수가 정말 흔합니다.
    //  반대로 SHORT_CONTEXT 만 보면 큰 컨텍스트가 잘린 채로 보입니다.
    // ========================================================================
    static final String A4_SQL = """
            SELECT JOB_EXECUTION_ID,
                   COALESCE(SERIALIZED_CONTEXT, SHORT_CONTEXT) AS job_context
              FROM BATCH_JOB_EXECUTION_CONTEXT
             ORDER BY JOB_EXECUTION_ID DESC
             LIMIT 1\\G

            -- 결과 예시
            -- *************************** 1. row ***************************
            -- JOB_EXECUTION_ID: 12
            --      job_context: {"@class":"java.util.HashMap",
            --                    "completedCount":["java.lang.Long",70000],
            --                    "settleDate":"2025-03-01"}
            --
            -- 참고 1. 값에 타입이 함께 붙는 이유는 Jackson default typing 때문입니다.
            --         덕분에 역직렬화 때 Long 이 Integer 로 바뀌는 사고가 없습니다.
            -- 참고 2. Spring Batch 4.x 는 자바 직렬화(BLOB)라 사람이 읽을 수 없었습니다.
            --         5.x 는 Jackson2ExecutionContextStringSerializer 가 기본입니다.
            """;

    // ========================================================================
    // 정답 5.
    //
    //  (a) 무엇이 조용히 망가졌나 — 재시작입니다.
    //
    //      @StepScope 는 @Scope(value="step", proxyMode=ScopedProxyMode.TARGET_CLASS)
    //      입니다. TARGET_CLASS 프록시가 "무엇을 구현할지"는 @Bean 메서드의
    //      리턴 타입이 결정합니다. 리턴 타입이 ItemReader<Order> 이면
    //      프록시도 ItemReader 만 구현하고 ItemStream 은 구현하지 않습니다.
    //
    //      SimpleStepBuilder.reader(r) 는 내부적으로
    //          if (r instanceof ItemStream) registerStream((ItemStream) r);
    //      를 하는데, 위 프록시는 이 검사를 통과하지 못합니다.
    //      → open() / update() / close() 가 한 번도 호출되지 않습니다.
    //
    //      JdbcCursorItemReader 였다면 open() 이 없어 ResultSet 이 안 열리므로
    //      ReaderNotOpenException 으로 시끄럽게 죽습니다. 운이 좋은 경우입니다.
    //      그런데 JdbcPagingItemReader 는 open() 없이도 첫 페이지를 그냥 조회해
    //      옵니다. 즉 70,000건을 전부 정상 처리하고 COMPLETED 로 끝납니다.
    //      아무 에러도 없습니다.
    //
    //      사라진 것은 update() 입니다. 매 청크 커밋마다
    //      "q5Reader.read.count" 를 컨텍스트에 적어 두던 그 콜백입니다.
    //      평소에는 아무 문제가 없다가, 장애로 재시작하는 날 처음부터 다시
    //      읽습니다. 이미 정산된 건을 다시 INSERT 하려다
    //      settlement.uk_settlement_order 에 걸려 DuplicateKeyException 이 나거나,
    //      UNIQUE 가 없었다면 정산이 두 배로 찍힙니다.
    //
    //  (b) 확인용 SQL — read.count 키가 남았는지 봅니다.
    //
    //      SELECT COALESCE(SERIALIZED_CONTEXT, SHORT_CONTEXT) AS ctx
    //        FROM BATCH_STEP_EXECUTION_CONTEXT
    //       ORDER BY STEP_EXECUTION_ID DESC LIMIT 1\G
    //
    //      정상이면 "q5Reader.read.count":70000 이 보입니다.
    //      깨졌으면 batch.stepType / batch.taskletType 두 개만 있습니다.
    //
    //      운영 배포 전 체크리스트에 이 한 줄을 넣으세요.
    //
    //  (c) 해법 두 가지
    // ========================================================================
    @Configuration
    static class A5 {

        /**
         * 해법 1 (권장) — 리턴 타입을 구현체로 선언합니다.
         * 프록시가 JdbcPagingItemReader 를 상속하므로 ItemStream 도 함께
         * 구현하게 되고, reader() 가 알아서 registerStream 합니다.
         * 코드에 아무 추가 규칙이 필요 없다는 것이 이 방법의 가치입니다.
         */
        @Bean
        @StepScope
        JdbcPagingItemReader<Order> a5Reader(
                DataSource dataSource,
                @Value("#{jobParameters['date']}") String date) {

            MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
            provider.setSelectClause("SELECT order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause("WHERE status='COMPLETED' AND DATE(ordered_at) = :date");
            provider.setSortKeys(Map.of("order_id", org.springframework.batch.item.database.Order.ASCENDING));

            JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
            reader.setName("a5Reader");
            reader.setDataSource(dataSource);
            reader.setQueryProvider(provider);
            reader.setParameterValues(Map.of("date", date));
            reader.setPageSize(1000);
            reader.setRowMapper(new DataClassRowMapper<>(Order.class));
            return reader;
        }

        @Bean
        Step a5Step(JobRepository jobRepository, PlatformTransactionManager tm,
                    JdbcPagingItemReader<Order> a5Reader,
                    ItemProcessor<Order, Settlement> a5Processor,
                    JdbcBatchItemWriter<Settlement> settlementWriter) {
            return new StepBuilder("a5Step", jobRepository)
                    .<Order, Settlement>chunk(1000, tm)
                    .reader(a5Reader)          // ItemStream 이 자동 등록됩니다
                    .processor(a5Processor)
                    .writer(settlementWriter)
                    .build();
        }

        /**
         * 해법 2 (차선책) — .stream(...) 으로 명시 등록합니다.
         *
         * 왜 차선책인가:
         *   1. 등록을 잊기 쉽습니다. 잊어도 에러가 안 나므로 리뷰에서도 안 잡힙니다.
         *   2. ItemStream 으로 캐스팅해야 하는데, 그러려면 결국 구현체가
         *      ItemStream 임을 알고 있어야 합니다. 그럴 바에는 해법 1 이 낫습니다.
         *   3. Reader 를 다른 Step 에서 재사용할 때 그 Step 에서도 똑같이
         *      .stream(...) 을 써야 합니다. 규칙이 전파되어야 하는 설계는 약합니다.
         *
         * 인터페이스 리턴이 불가피한 경우(예: 조건에 따라 서로 다른 구현체를
         * 반환해야 할 때)에만 쓰세요.
         */
        Step a5StepAlternative(JobRepository jobRepository, PlatformTransactionManager tm,
                               org.springframework.batch.item.ItemReader<Order> leaky,
                               ItemProcessor<Order, Settlement> a5Processor,
                               JdbcBatchItemWriter<Settlement> settlementWriter) {
            return new StepBuilder("a5StepAlt", jobRepository)
                    .<Order, Settlement>chunk(1000, tm)
                    .reader(leaky)
                    .processor(a5Processor)
                    .writer(settlementWriter)
                    .stream((ItemStream) leaky)     // ← 명시 등록
                    .build();
        }
    }

    /*
     * ========================================================================
     * 정답 6.
     *
     *  판단 기준은 한 줄로 요약됩니다.
     *
     *    stepExecutionContext 가 필요하다        → @StepScope
     *    Job 전체에서 인스턴스가 하나여야 한다     → @JobScope
     *    실행 시점 값이 전혀 필요 없다            → 싱글턴 (스코프 없음)
     *
     *  (1) jobParameters['inputFile'] 로 경로를 받는 FlatFileItemReader
     *      답: @StepScope
     *      이유: jobParameters 를 SpEL 로 받아야 하므로 스코프가 필요합니다.
     *            @JobScope 로도 파라미터 자체는 읽히지만, Reader 는
     *            ItemStream 상태(읽은 줄 수)를 StepExecution 단위로 관리해야
     *            하므로 @StepScope 가 맞습니다.
     *            그리고 리턴 타입은 반드시 FlatFileItemReader<T> 로 —
     *            ItemReader<T> 로 쓰면 정답 5 의 함정에 그대로 빠집니다.
     *            FlatFileItemReader 는 open() 에서 파일을 열기 때문에
     *            이 경우엔 ReaderNotOpenException 으로 시끄럽게 죽습니다.
     *
     *  (2) stepExecutionContext['minId'] / ['maxId'] 를 받는 파티션 Reader
     *      답: @StepScope (다른 선택지가 없습니다)
     *      이유: stepExecutionContext 는 @StepScope 에서만 해석됩니다.
     *            @JobScope 를 붙이면 EL1008E 로 기동에 실패합니다.
     *            파티셔닝은 파티션마다 별도의 StepExecution 을 만들고 각각의
     *            Step Context 에 minId/maxId 를 넣어 주는 구조이므로,
     *            "StepExecution 마다 새 인스턴스"라는 @StepScope 의 성질이
     *            그대로 요구사항입니다. (Step 13 에서 다룹니다)
     *
     *  (3) Job 전체의 처리 건수를 누적하는 StatsCollector
     *      답: @JobScope
     *      이유: JobExecution 하나당 인스턴스가 하나여야 누적이 성립합니다.
     *            싱글턴으로 두면 애플리케이션이 여러 Job 을 돌릴 때 값이
     *            섞이고, 특히 같은 Job 을 두 번 실행하면 이전 실행의 숫자가
     *            남습니다.
     *
     *      @StepScope 를 고르면 벌어지는 일:
     *            Step 이 시작될 때마다 새 인스턴스가 만들어집니다.
     *            즉 누적 카운터가 매 Step 마다 0 으로 초기화됩니다.
     *            마지막 Step 의 건수만 리포트되는데, 그 값도 "그럴듯한 숫자"라
     *            아무도 이상함을 눈치채지 못합니다. 예외도 로그도 없습니다.
     *            이것이 이 코스가 계속 말하는 "에러 없이 조용히 틀리는 코드"의
     *            전형입니다.
     *
     *      더 나은 대안:
     *            누적값을 빈 필드가 아니라 ExecutionContext 에 두는 것입니다.
     *            그래야 DB 에 남고, 재시작 후에도 복원됩니다.
     *            빈 필드는 스코프가 무엇이든 프로세스가 죽으면 사라집니다.
     * ========================================================================
     */
    @Configuration
    static class A6 {

        /** (3) 의 정답 형태 — @JobScope 로 JobExecution 당 하나. */
        @Bean
        @JobScope
        StatsCollector statsCollector(
                @Value("#{jobExecution.jobInstance.jobName}") String jobName) {
            return new StatsCollector(jobName);
        }
    }

    static class StatsCollector {
        private final String jobName;
        private final AtomicLong total = new AtomicLong();

        StatsCollector(String jobName) {
            this.jobName = jobName;
        }

        void add(long n) {
            total.addAndGet(n);
        }

        long total() {
            return total.get();
        }

        String jobName() {
            return jobName;
        }
    }
}