package study.batchquartz;

import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.batch.core.configuration.support.MapJobRegistry;
import org.springframework.batch.core.job.Job;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.job.parameters.JobParameters;
import org.springframework.batch.core.job.parameters.JobParametersBuilder;
import org.springframework.batch.core.launch.support.SimpleJobOperator;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JdbcJobRepositoryFactoryBean;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.infrastructure.item.ItemProcessor;
import org.springframework.batch.infrastructure.item.ItemWriter;
import org.springframework.batch.infrastructure.item.support.AbstractItemCountingItemStreamItemReader;
import org.springframework.batch.infrastructure.repeat.RepeatStatus;
import org.springframework.transaction.PlatformTransactionManager;

/// U46 的 Batch 运行基座：真实 MySQL JobRepository（官方 MySQL schema 由 LabDatabase 执行）、
/// 可测的 JobOperator、内存记录器与按编号失败的合成服务。
final class BatchFixture implements AutoCloseable {

    final LabDatabase db;
    final JobRepository jobRepository;
    final SimpleJobOperator jobOperator;
    final MapJobRegistry jobRegistry = new MapJobRegistry();
    final ListRecorder recorder = new ListRecorder();
    final FailingService failing = new FailingService();

    BatchFixture() throws Exception {
        this.db = new LabDatabase("org/springframework/batch/core/schema-mysql.sql");
        this.db.execute("CREATE TABLE u46_business (id BIGINT AUTO_INCREMENT PRIMARY KEY, item_no INT NOT NULL)");
        JdbcJobRepositoryFactoryBean factory = new JdbcJobRepositoryFactoryBean();
        factory.setDataSource(this.db.dataSource());
        factory.setTransactionManager(this.db.transactionManager());
        factory.setDatabaseType("MYSQL");
        factory.afterPropertiesSet();
        this.jobRepository = factory.getObject();
        this.jobOperator = new SimpleJobOperator();
        this.jobOperator.setJobRepository(this.jobRepository);
        this.jobOperator.setJobRegistry(this.jobRegistry);
        this.jobOperator.afterPropertiesSet();
    }

    PlatformTransactionManager transactionManager() {
        return this.db.transactionManager();
    }

    StepBuilder step(String name) {
        return new StepBuilder(name, this.jobRepository);
    }

    JobBuilder jobBuilder(String name) {
        return new JobBuilder(name, this.jobRepository);
    }

    Job recordingTaskletJob(String name) {
        Tasklet tasklet = (contribution, chunkContext) -> {
            this.recorder.record("run:" + name);
            return RepeatStatus.FINISHED;
        };
        Job job = new JobBuilder(name, this.jobRepository)
                .start(new StepBuilder(name + "-step", this.jobRepository)
                        .tasklet(tasklet, this.db.transactionManager())
                        .build())
                .build();
        register(job);
        return job;
    }

    void register(Job job) {
        try {
            this.jobRegistry.register(job);
        }
        catch (Exception e) {
            throw new IllegalStateException("job 注册失败：" + job.getName(), e);
        }
    }

    static JobParameters jobParameters(String key, String value, boolean identifying) {
        return new JobParametersBuilder().addString(key, value, identifying).toJobParameters();
    }

    @Override
    public void close() {
        this.db.close();
    }

    static final class ListRecorder {
        final CopyOnWriteArrayList<String> events = new CopyOnWriteArrayList<>();

        void record(String event) {
            this.events.add(event);
        }

        long count(String prefix) {
            return this.events.stream().filter(event -> event.startsWith(prefix)).count();
        }

        void clear() {
            this.events.clear();
        }
    }

    static final class FailingService {
        final AtomicInteger attempts = new AtomicInteger();
        final CopyOnWriteArrayList<Integer> failuresSeen = new CopyOnWriteArrayList<>();
        volatile boolean alwaysFail = false;
        /// 按业务条目号失败（-1 表示不按条目失败）；失败次数耗尽后恢复。
        volatile int failOnItem = -1;
        volatile int failTimes = 1;

        void call(int id) {
            this.attempts.incrementAndGet();
            if (this.alwaysFail || (id == this.failOnItem && this.failTimes > 0)) {
                if (id == this.failOnItem) {
                    this.failTimes--;
                }
                this.failuresSeen.add(id);
                throw new IllegalStateException("synthetic failure for item " + id);
            }
        }

        void reset() {
            this.attempts.set(0);
            this.failuresSeen.clear();
            this.alwaysFail = false;
            this.failOnItem = -1;
            this.failTimes = 1;
        }
    }

    /// 合成 reader：从 1 读到 `total`；checkpoint 语义与 Batch 自带 Counting 读者一致，
    /// 目的只是观察“重启从哪一条继续读”，不模拟外部系统。
    static final class SyntheticReader extends AbstractItemCountingItemStreamItemReader<Integer> {
        private final int total;
        private final ListRecorder recorder;

        SyntheticReader(int total, ListRecorder recorder) {
            this.total = total;
            this.recorder = recorder;
            setName("syntheticReader");
        }

        @Override
        protected Integer doRead() {
            // 基类 read() 在调用 doRead() 之前已经把 currentItemCount 自增，所以这里直接用
            // 它的值；再 +1 会跳号（这版最初就犯过这个错，读到了 2、3 而不是 1、2）。
            int current = getCurrentItemCount();
            if (current > this.total) {
                return null;
            }
            this.recorder.record("read:" + current);
            return current;
        }

        @Override
        protected void doOpen() {
        }

        @Override
        protected void doClose() {
        }
    }

    /// 合成 writer：用真实 MySQL 记录每次写入的行，便于核对“重跑后业务表里有什么”。
    static class RecordingWriter implements ItemWriter<Integer> {
        private final LabDatabase db;
        private final ListRecorder recorder;
        private final FailingService service;

        RecordingWriter(LabDatabase db, ListRecorder recorder, FailingService service) {
            this.db = db;
            this.recorder = recorder;
            this.service = service;
        }

        @Override
        public void write(org.springframework.batch.infrastructure.item.Chunk<? extends Integer> chunk)
                throws Exception {
            for (Integer item : chunk) {
                this.service.call(item);
                // 关键：必须拿事务绑定连接（DataSourceUtils），否则自取的连接在块回滚时不会回滚，
                // 实验会错误地“证明”整块失败没有留下任何行——这本身是 chunk 事务语义的一部分。
                java.sql.Connection connection = org.springframework.jdbc.datasource.DataSourceUtils
                        .getConnection(this.db.dataSource());
                try (var statement = connection.prepareStatement(
                        "INSERT INTO u46_business(item_no) VALUES (?)")) {
                    statement.setInt(1, item);
                    statement.executeUpdate();
                }
                finally {
                    org.springframework.jdbc.datasource.DataSourceUtils.releaseConnection(
                            connection, this.db.dataSource());
                }
                this.recorder.record("write:" + item);
            }
        }
    }

    static final class IdentityProcessor implements ItemProcessor<Integer, Integer> {
        @Override
        public Integer process(Integer item) {
            return item;
        }
    }
}