package study.batchquartz;

import java.util.List;
import java.util.Map;

import org.junit.jupiter.api.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.job.Job;
import org.springframework.batch.core.job.JobExecution;
import org.springframework.batch.core.job.parameters.JobParameters;
import org.springframework.batch.core.job.parameters.JobParametersBuilder;
import org.springframework.batch.core.launch.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.launch.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.launch.JobRestartException;
import org.springframework.batch.core.step.StepExecution;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

/// U46 判定实验之二：Spring Batch 6.0.6 的 JobInstance 身份、重启校验链、chunk 事务与
/// checkpoint、fault-tolerant 默认值、abandon/recover、并发启动互斥。全部落在真实 MySQL 元数据表。
class BatchPersistenceTest {

    @Test
    void jobIdentityIsNamePlusIdentifyingParameters() throws Exception {
        try (BatchFixture fixture = new BatchFixture()) {
            Job job = fixture.recordingTaskletJob("identityJob");
            JobParameters runA = BatchFixture.jobParameters("runDate", "2026-10-29", true);
            JobExecution first = fixture.jobOperator.start(job, runA);
            assertEquals(BatchStatus.COMPLETED, first.getStatus());

            // 相同 identifying 参数再次启动：实例已完成 -> 拒绝。
            assertThrows(JobInstanceAlreadyCompleteException.class,
                    () -> fixture.jobOperator.start(job, BatchFixture.jobParameters("runDate", "2026-10-29", true)));

            // 不同 identifying 参数：新实例。
            JobExecution second = fixture.jobOperator.start(job, BatchFixture.jobParameters("runDate", "2026-10-30", true));
            assertNotEquals(first.getJobInstance().getId(), second.getJobInstance().getId());

            // 同名 job、仅非 identifying 参数不同：仍是同一实例 -> 也是“已存在且完成”。
            JobParameters sameIdentityOtherParam = new JobParametersBuilder()
                    .addString("runDate", "2026-10-29", true)
                    .addString("traceId", "t-1", false)
                    .toJobParameters();
            assertThrows(JobInstanceAlreadyCompleteException.class,
                    () -> fixture.jobOperator.start(job, sameIdentityOtherParam));

            assertEquals(2L, fixture.db.count("SELECT COUNT(*) FROM BATCH_JOB_INSTANCE WHERE JOB_NAME='identityJob'"));
            assertEquals(2L, fixture.db.count("SELECT COUNT(*) FROM BATCH_JOB_EXECUTION E JOIN BATCH_JOB_INSTANCE I"
                    + " ON E.JOB_INSTANCE_ID=I.JOB_INSTANCE_ID WHERE I.JOB_NAME='identityJob'"));
            List<Map<String, Object>> keys = fixture.db.rows(
                    "SELECT JOB_KEY FROM BATCH_JOB_INSTANCE WHERE JOB_NAME='identityJob' ORDER BY JOB_INSTANCE_ID");
            assertNotEquals(keys.get(0).get("JOB_KEY"), keys.get(1).get("JOB_KEY"));
            System.out.println("U46 batchIdentity instances=2 executions=2 jobKeyDiffers=true"
                    + " nonIdentifyingIgnored=true");
        }
    }

    @Test
    void failedChunkRestartResumesFromCheckpointAndReplaysUncommittedWork() throws Exception {
        try (BatchFixture fixture = new BatchFixture()) {
            // 6 条数据、chunk=2：第 3 条写入时失败（第 1 块已提交），重跑应从未提交块继续。
            fixture.failing.failOnItem = 3;
            var reader = new BatchFixture.SyntheticReader(6, fixture.recorder);
            var writer = new BatchFixture.RecordingWriter(fixture.db, fixture.recorder, fixture.failing);
            Job job = fixture.jobBuilder("chunkJob")
                    .start(fixture.step("chunkStep")
                            .<Integer, Integer>chunk(2)
                            // 6.0 的 chunk(int) 不再接受 transactionManager 参数；不显式设置时
                            // 构建器默认 ResourcelessTransactionManager——业务事务根本不存在。
                            .transactionManager(fixture.transactionManager())
                            .reader(reader)
                            .writer(writer)
                            .build())
                    .build();
            fixture.register(job);
            JobParameters params = BatchFixture.jobParameters("runDate", "2026-10-29", true);
            JobExecution failed = fixture.jobOperator.start(job, params);
            assertEquals(BatchStatus.FAILED, failed.getStatus(), "未提交块写失败导致 step 失败");
            long rowsAfterFailure = fixture.db.count("SELECT COUNT(*) FROM u46_business");
            assertEquals(2L, rowsAfterFailure, "失败块整体回滚，只有前一块的 2 行在业务表");

            JobExecution restarted = fixture.jobOperator.restart(failed);
            assertEquals(BatchStatus.COMPLETED, restarted.getStatus());
            assertEquals(1L, fixture.db.count("SELECT COUNT(*) FROM u46_business WHERE item_no=3"),
                    "重跑后第 3 条恰好写入一次");
            assertEquals(6L, fixture.db.count("SELECT COUNT(*) FROM u46_business"));
            assertEquals(2L, fixture.db.count("SELECT COUNT(*) FROM BATCH_JOB_EXECUTION E JOIN BATCH_JOB_INSTANCE I"
                    + " ON E.JOB_INSTANCE_ID=I.JOB_INSTANCE_ID WHERE I.JOB_NAME='chunkJob'"),
                    "同一实例第二次执行，不是新实例");
            long readsAfterRestart = fixture.recorder.count("read:");
            assertTrue(readsAfterRestart > 6,
                    "重跑会重读未提交块（checkpoint 只回退到第 2 条之后）：reads=" + readsAfterRestart);
            Map<String, Object> lastStep = fixture.db.rows("SELECT STATUS, READ_COUNT, WRITE_COUNT, COMMIT_COUNT,"
                    + " ROLLBACK_COUNT FROM BATCH_STEP_EXECUTION WHERE STEP_NAME='chunkStep'"
                    + " ORDER BY STEP_EXECUTION_ID DESC LIMIT 1").get(0);
            assertEquals("COMPLETED", lastStep.get("STATUS"));
            System.out.println("U46 batchRestart firstStatus=FAILED rowsAfterFailure=" + rowsAfterFailure
                    + " finalRows=6 reads=" + readsAfterRestart + " commits=" + lastStep.get("COMMIT_COUNT")
                    + " rollbacks=" + lastStep.get("ROLLBACK_COUNT"));
        }
    }

    @Test
    void skipLimitWithoutFaultTolerantDoesNotSkip() throws Exception {
        try (BatchFixture fixture = new BatchFixture()) {
            fixture.failing.alwaysFail = true;
            var reader = new BatchFixture.SyntheticReader(4, fixture.recorder);
            var writer = new BatchFixture.RecordingWriter(fixture.db, fixture.recorder, fixture.failing);
            Job job = fixture.jobBuilder("noFtJob")
                    .start(fixture.step("noFtStep")
                            .<Integer, Integer>chunk(2)
                            .reader(reader)
                            .writer(writer)
                            .skipLimit(10)
                            .build())
                    .build();
            fixture.register(job);
            JobExecution execution = fixture.jobOperator.start(job,
                    BatchFixture.jobParameters("runDate", "2026-10-29", true));
            assertEquals(BatchStatus.FAILED, execution.getStatus(),
                    "只设 skipLimit 而不开 faultTolerant，异常仍然让 step 失败");
            long skipped = (Long) fixture.db.rows("SELECT PROCESS_SKIP_COUNT, WRITE_SKIP_COUNT FROM BATCH_STEP_EXECUTION"
                    + " WHERE STEP_NAME='noFtStep'").get(0).get("WRITE_SKIP_COUNT");
            assertEquals(0L, skipped, "未开 faultTolerant 时 WRITE_SKIP_COUNT 保持 0");
            System.out.println("U46 batchSkipWithoutFt status=FAILED writeSkip=0");
        }
    }

    @Test
    void faultTolerantSkipsBadItemAndKeepsOthers() throws Exception {
        try (BatchFixture fixture = new BatchFixture()) {
            fixture.failing.alwaysFail = false;
            var reader = new BatchFixture.SyntheticReader(4, fixture.recorder);
            var failingWriter = new BatchFixture.RecordingWriter(fixture.db, fixture.recorder, fixture.failing) {
                @Override
                public void write(org.springframework.batch.infrastructure.item.Chunk<? extends Integer> chunk)
                        throws Exception {
                    for (Integer item : chunk) {
                        if (item == 2) {
                            throw new IllegalStateException("permanent failure on 2");
                        }
                        super.write(org.springframework.batch.infrastructure.item.Chunk.of(item));
                    }
                }
            };
            Job job = fixture.jobBuilder("ftJob")
                    .start(fixture.step("ftStep")
                            .<Integer, Integer>chunk(2)
                            .transactionManager(fixture.transactionManager())
                            .reader(reader)
                            .writer(failingWriter)
                            .faultTolerant()
                            .skip(IllegalStateException.class)
                            .skipLimit(5)
                            .build())
                    .build();
            fixture.register(job);
            JobExecution execution = fixture.jobOperator.start(job,
                    BatchFixture.jobParameters("runDate", "2026-10-29", true));
            assertEquals(BatchStatus.COMPLETED, execution.getStatus(), "可跳过异常不失败 step");
            assertEquals(3L, fixture.db.count("SELECT COUNT(*) FROM u46_business"), "2 被跳过，1/3/4 写入");
            assertEquals(0L, fixture.db.count("SELECT COUNT(*) FROM u46_business WHERE item_no=2"));
            long writeSkips = ((Number) fixture.db.rows(
                    "SELECT WRITE_SKIP_COUNT FROM BATCH_STEP_EXECUTION WHERE STEP_NAME='ftStep'")
                    .get(0).get("WRITE_SKIP_COUNT")).longValue();
            assertEquals(1L, writeSkips, "WRITE_SKIP_COUNT 记录被跳过条目");
            System.out.println("U46 batchFaultTolerant status=COMPLETED rows=3 writeSkips=" + writeSkips);
        }
    }

    @Test
    void abandonMakesExecutionUnrestartableAndRecoverUnsticksStarted() throws Exception {
        try (BatchFixture fixture = new BatchFixture()) {
            Job job = fixture.jobBuilder("abandonJob")
                    .start(fixture.step("abandonStep")
                            .tasklet((contribution, chunkContext) -> {
                                throw new IllegalStateException("boom");
                            }, fixture.transactionManager())
                            .build())
                    .build();
            fixture.register(job);
            JobExecution failed = fixture.jobOperator.start(job,
                    BatchFixture.jobParameters("runDate", "2026-10-29", true));
            assertEquals(BatchStatus.FAILED, failed.getStatus());

            JobExecution abandoned = fixture.jobOperator.abandon(failed);
            assertEquals(BatchStatus.ABANDONED, abandoned.getStatus());
            assertThrows(JobInstanceAlreadyCompleteException.class,
                    () -> fixture.jobOperator.start(job, BatchFixture.jobParameters("runDate", "2026-10-29", true)),
                    "ABANDONED 与 COMPLETED 一样阻止同实例再启动");

            // 模拟“进程在执行中突然死亡”：直接把 execution 改成 STARTED，重启被拒。
            JobExecution other = fixture.jobOperator.start(job,
                    BatchFixture.jobParameters("runDate", "2026-10-31", true));
            assertEquals(BatchStatus.FAILED, other.getStatus());
            fixture.db.execute("UPDATE BATCH_JOB_EXECUTION SET STATUS='STARTED', END_TIME=NULL, EXIT_CODE='UNKNOWN'"
                    + " WHERE JOB_EXECUTION_ID=" + other.getId());
            fixture.db.execute("UPDATE BATCH_STEP_EXECUTION SET STATUS='STARTED', END_TIME=NULL"
                    + " WHERE JOB_EXECUTION_ID=" + other.getId() + " AND STEP_NAME='abandonStep'");
            JobExecution reloaded = fixture.jobRepository.getJobExecution(other.getId());
            assertEquals(BatchStatus.STARTED, reloaded.getStatus());
            assertThrows(JobExecutionAlreadyRunningException.class,
                    () -> fixture.jobOperator.start(job, BatchFixture.jobParameters("runDate", "2026-10-31", true)),
                    "卡在 STARTED 的执行让同参数启动报 AlreadyRunning");

            JobExecution recovered = fixture.jobOperator.recover(reloaded);
            assertEquals(BatchStatus.FAILED, recovered.getStatus(), "recover 把卡死执行标为 FAILED 使其可重启");
            JobExecution restarted = fixture.jobOperator.restart(recovered);
            assertEquals(BatchStatus.FAILED, restarted.getStatus(),
                    "重启后 step 仍抛异常，所以再次失败——但重启本身被允许");
            assertNotEquals(recovered.getId(), restarted.getId(), "restart 产生新的 JobExecution 行");
            System.out.println("U46 batchAbandonRecover abandoned=true recoverUnstuck=true restartNewExecution=true");
        }
    }

    @Test
    void completedStepIsSkippedOnRestartUnlessAllowStartIfComplete() throws Exception {
        try (BatchFixture fixture = new BatchFixture()) {
            var countingStep1 = new CountingTasklet(fixture.recorder, "step1");
            var failingStep2 = new CountingTasklet(fixture.recorder, "step2");
            failingStep2.failOnce = true;
            Job job = fixture.jobBuilder("multiStepJob")
                    .start(fixture.step("multiStep1")
                            .tasklet(countingStep1, fixture.transactionManager())
                            .build())
                    .next(fixture.step("multiStep2")
                            .tasklet(failingStep2, fixture.transactionManager())
                            .build())
                    .build();
            fixture.register(job);
            JobExecution first = fixture.jobOperator.start(job,
                    BatchFixture.jobParameters("runDate", "2026-10-29", true));
            assertEquals(BatchStatus.FAILED, first.getStatus());
            long step1RunsBefore = fixture.recorder.count("step1:");
            assertEquals(1L, step1RunsBefore);

            JobExecution restarted = fixture.jobOperator.restart(first);
            assertEquals(BatchStatus.COMPLETED, restarted.getStatus());
            assertEquals(1L, fixture.recorder.count("step1:"), "已完成 step 在重启时被跳过");
            assertEquals(2L, fixture.recorder.count("step2:"), "失败 step 重跑一次");

            // 第二次执行中，step1 的 StepExecution 计数应为 1（未重跑）。
            List<Map<String, Object>> step1Rows = fixture.db.rows(
                    "SELECT STATUS FROM BATCH_STEP_EXECUTION WHERE STEP_NAME='multiStep1'");
            assertEquals(1L, (long) step1Rows.size(), "重启没有为已完成 step 建新 StepExecution");
            System.out.println("U46 batchStepSkip step1Runs=" + fixture.recorder.count("step1:")
                    + " step2Runs=" + fixture.recorder.count("step2:") + " step1Executions=1");
        }
    }

    @Test
    void concurrentLaunchOfSameInstanceOnlyOneWins() throws Exception {
        try (BatchFixture fixture = new BatchFixture()) {
            // 慢 job：让第一次执行在窗口内保持运行。
            Job job = fixture.jobBuilder("slowJob")
                    .start(fixture.step("slowStep")
                            .tasklet((contribution, chunkContext) -> {
                                Thread.sleep(1_500);
                                return org.springframework.batch.infrastructure.repeat.RepeatStatus.FINISHED;
                            }, fixture.transactionManager())
                            .build())
                    .build();
            fixture.register(job);
            // 默认 launcher 走 SyncTaskExecutor：start() 会等 job 跑完才返回。要观察“运行中并发启动被拒”，
            // 必须显式换成异步执行器（生产语义分两半：running 判定 + create* 的 SERIALIZABLE 事务）。
            org.springframework.core.task.SimpleAsyncTaskExecutor async =
                    new org.springframework.core.task.SimpleAsyncTaskExecutor();
            async.setThreadNamePrefix("u46-launch-");
            fixture.jobOperator.setTaskExecutor(async);
            JobParameters params = BatchFixture.jobParameters("runDate", "2026-10-29", true);
            JobExecution running = fixture.jobOperator.start(job, params);
            assertEquals(BatchStatus.STARTING, running.getStatus(),
                    "异步启动返回的 execution 处于 STARTING，job 在别的线程里跑");
            assertThrows(JobExecutionAlreadyRunningException.class,
                    () -> fixture.jobOperator.start(job, params), "同一实例的并发启动被拒绝（last execution 仍 running）");
            Thread.sleep(2_500);
            JobExecution finished = fixture.jobRepository.getJobExecution(running.getId());
            assertEquals(BatchStatus.COMPLETED, finished.getStatus());
            assertEquals(1L, fixture.db.count("SELECT COUNT(*) FROM BATCH_JOB_EXECUTION"), "只产生一条执行记录");
            System.out.println("U46 batchConcurrentLaunch rejected=true executions=1");
        }
    }

    static final class CountingTasklet implements org.springframework.batch.core.step.tasklet.Tasklet {
        private final BatchFixture.ListRecorder recorder;
        private final String name;
        volatile boolean failOnce;

        CountingTasklet(BatchFixture.ListRecorder recorder, String name) {
            this.recorder = recorder;
            this.name = name;
        }

        @Override
        public org.springframework.batch.infrastructure.repeat.RepeatStatus execute(
                org.springframework.batch.core.step.StepContribution contribution,
                org.springframework.batch.core.scope.context.ChunkContext chunkContext) {
            this.recorder.record(this.name + ":run");
            if (this.failOnce) {
                this.failOnce = false;
                throw new IllegalStateException("synthetic step failure");
            }
            return org.springframework.batch.infrastructure.repeat.RepeatStatus.FINISHED;
        }
    }
}