package com.easyagents.flow.core.test;

import com.easyagents.flow.core.chain.runtime.InMemoryTriggerStore;
import com.easyagents.flow.core.chain.runtime.NonRetryableTriggerException;
import com.easyagents.flow.core.chain.runtime.RetryableTriggerException;
import com.easyagents.flow.core.chain.runtime.Trigger;
import com.easyagents.flow.core.chain.runtime.TriggerScheduler;
import com.easyagents.flow.core.chain.runtime.TriggerStore;
import org.junit.Assert;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.lang.reflect.Field;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * {@link TriggerScheduler} 认领、确认和失败释放语义回归测试。
 */
public class TriggerSchedulerReliabilityTest {

    /**
     * 验证消费者尚未注册时主动触发不会丢失待执行任务。
     */
    @Test
    public void shouldKeepTriggerPendingWhenConsumerIsUnavailable() {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        Trigger trigger = futureTrigger("consumer-unavailable");
        store.save(trigger);
        TriggerScheduler scheduler = scheduler(store);
        try {
            Assert.assertFalse(scheduler.fire(trigger.getId()));
            Assert.assertNotNull(store.find(trigger.getId()));
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证执行失败后触发器被释放，并能再次认领成功。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldReleaseFailedTriggerForRetry() throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        Trigger trigger = futureTrigger("retry-after-failure");
        store.save(trigger);
        TriggerScheduler scheduler = scheduler(store);
        AtomicInteger attempts = new AtomicInteger();
        CountDownLatch firstAttempt = new CountDownLatch(1);
        CountDownLatch succeeded = new CountDownLatch(1);
        scheduler.registerConsumer((current, worker) -> {
            if (attempts.incrementAndGet() == 1) {
                firstAttempt.countDown();
                throw new IllegalStateException("expected first failure");
            }
            succeeded.countDown();
        });

        try {
            Assert.assertTrue(scheduler.fire(trigger.getId()));
            Assert.assertTrue(firstAttempt.await(2, TimeUnit.SECONDS));
            awaitPending(store, trigger.getId());
            Assert.assertTrue(scheduler.fire(trigger.getId()));
            Assert.assertTrue(succeeded.await(2, TimeUnit.SECONDS));
            Assert.assertEquals(2, attempts.get());
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证并发认领同一触发器时只有一个消费者获得执行权。
     *
     * @throws Exception 并发任务等待失败时抛出
     */
    @Test
    public void shouldDispatchClaimedTriggerOnlyOnce() throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        Trigger trigger = futureTrigger("single-claim");
        store.save(trigger);
        TriggerScheduler scheduler = scheduler(store);
        CountDownLatch consumed = new CountDownLatch(1);
        AtomicInteger executions = new AtomicInteger();
        scheduler.registerConsumer((current, worker) -> {
            executions.incrementAndGet();
            consumed.countDown();
        });
        ExecutorService callers = Executors.newFixedThreadPool(2);

        try {
            List<java.util.concurrent.Future<Boolean>> results = callers.invokeAll(Arrays.asList(
                    () -> scheduler.fire(trigger.getId()),
                    () -> scheduler.fire(trigger.getId())));
            int accepted = 0;
            for (java.util.concurrent.Future<Boolean> result : results) {
                if (result.get()) {
                    accepted++;
                }
            }
            Assert.assertEquals(1, accepted);
            Assert.assertTrue(consumed.await(2, TimeUnit.SECONDS));
            Assert.assertEquals(1, executions.get());
        } finally {
            callers.shutdownNow();
            scheduler.shutdown();
        }
    }

    /**
     * 验证缺少认领对象时旧 ID 接口不会静默执行不安全确认。
     */
    @Test
    public void shouldRejectClaimMutationWithoutClaimedTrigger() {
        InMemoryTriggerStore store = new InMemoryTriggerStore();

        try {
            store.renewClaim("unsafe-id-only", 1000L);
            Assert.fail("ID-only renewal must be rejected");
        } catch (UnsupportedOperationException expected) {
            Assert.assertTrue(expected.getMessage().contains("token"));
        }

        try {
            store.acknowledge("unsafe-id-only");
            Assert.fail("ID-only acknowledge must be rejected");
        } catch (UnsupportedOperationException expected) {
            Assert.assertTrue(expected.getMessage().contains("trigger"));
        }
    }

    /**
     * 验证不可重试错误会移出待执行集合，避免形成永久热重放。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldDeadLetterNonRetryableTrigger() throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        Trigger trigger = futureTrigger("non-retryable");
        store.save(trigger);
        TriggerScheduler scheduler = scheduler(store);
        CountDownLatch attempted = new CountDownLatch(1);
        scheduler.registerConsumer((current, worker) -> {
            attempted.countDown();
            throw new NonRetryableTriggerException("invalid definition");
        });

        try {
            Assert.assertTrue(scheduler.fire(trigger.getId()));
            Assert.assertTrue(attempted.await(2, TimeUnit.SECONDS));
            long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
            while (store.find(trigger.getId()) != null && System.nanoTime() < deadline) {
                Thread.sleep(10L);
            }
            Assert.assertNull(store.find(trigger.getId()));
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证瞬时基础设施冲突即使超过普通投递上限也继续保留，避免幂等 owner 存活期内误死信。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldNotDeadLetterTransientContention()
            throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        Trigger trigger = futureTrigger("transient-contention");
        trigger.setDeliveryAttempt(100);
        store.save(trigger);
        TriggerScheduler scheduler = scheduler(store);
        CountDownLatch attempted = new CountDownLatch(1);
        scheduler.registerConsumer((current, worker) -> {
            attempted.countDown();
            throw new RetryableTriggerException(
                    "operation still owned", null);
        });

        try {
            Assert.assertTrue(scheduler.fire(trigger.getId()));
            Assert.assertTrue(attempted.await(2, TimeUnit.SECONDS));
            awaitPending(store, trigger.getId());
            Assert.assertEquals(
                    101,
                    store.find(trigger.getId())
                            .getDeliveryAttempt());
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证业务终态尚未持久化时不得先移除触发器。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldRetryTerminalConvergenceBeforeDeadLetter()
            throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        Trigger trigger = futureTrigger(
                "terminal-convergence-retry");
        store.save(trigger);
        TriggerScheduler scheduler = scheduler(store);
        AtomicInteger convergenceAttempts = new AtomicInteger();
        AtomicInteger executions = new AtomicInteger();
        CountDownLatch consumed = new CountDownLatch(1);
        scheduler.registerConsumer((current, worker) -> {
            executions.incrementAndGet();
            consumed.countDown();
            throw new NonRetryableTriggerException(
                    "invalid definition");
        });
        scheduler.registerFailureListener((current, failure) ->
                convergenceAttempts.incrementAndGet() >= 2);

        try {
            Assert.assertTrue(scheduler.fire(trigger.getId()));
            awaitPending(store, trigger.getId());
            Assert.assertNotNull(store.find(trigger.getId()));
            Assert.assertTrue(scheduler.fire(trigger.getId()));
            Assert.assertTrue(consumed.await(2, TimeUnit.SECONDS));
            long deadline = System.nanoTime()
                    + TimeUnit.SECONDS.toNanos(2);
            while ((store.find(trigger.getId()) != null
                    || convergenceAttempts.get() < 2)
                    && System.nanoTime() < deadline) {
                Thread.sleep(10L);
            }
            Assert.assertNull(store.find(trigger.getId()));
            Assert.assertEquals(2, convergenceAttempts.get());
            Assert.assertEquals(1, executions.get());
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证死信写入失败后只重放终态协议，不重复执行业务消费者。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldResumeDeadLetterFinalizationWithoutBusinessReplay()
            throws Exception {
        FailingTerminalStore store =
                new FailingTerminalStore(
                        TerminalOperation.DEAD_LETTER);
        Trigger trigger = futureTrigger(
                "dead-letter-finalization-retry");
        store.save(trigger);
        TriggerScheduler scheduler = scheduler(store);
        AtomicInteger executions = new AtomicInteger();
        AtomicInteger convergenceAttempts =
                new AtomicInteger();
        CountDownLatch convergenceCompleted =
                new CountDownLatch(2);
        scheduler.registerConsumer((current, worker) -> {
            executions.incrementAndGet();
            throw new NonRetryableTriggerException(
                    "invalid definition");
        });
        scheduler.registerFailureListener(
                (current, failure) -> {
                    convergenceAttempts.incrementAndGet();
                    convergenceCompleted.countDown();
                    return true;
                });

        try {
            Assert.assertTrue(scheduler.fire(
                    trigger.getId()));
            Assert.assertTrue(store.terminalAttempted.await(
                    2, TimeUnit.SECONDS));
            awaitPending(store, trigger.getId());
            Assert.assertTrue(store.find(trigger.getId())
                    .isDeadLetterPending());

            store.fail = false;
            Assert.assertTrue(scheduler.fire(
                    trigger.getId()));
            Assert.assertTrue(convergenceCompleted.await(
                    2, TimeUnit.SECONDS));
            long deadline = System.nanoTime()
                    + TimeUnit.SECONDS.toNanos(2);
            while (store.find(trigger.getId()) != null
                    && System.nanoTime() < deadline) {
                Thread.sleep(10L);
            }

            Assert.assertNull(store.find(trigger.getId()));
            Assert.assertEquals(1, executions.get());
            Assert.assertEquals(
                    2, convergenceAttempts.get());
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证旧 owner 失去 claim 后不能提交业务失败终态或移动新 owner 的触发器。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldRejectDeadLetterFinalizationAfterClaimTransfer()
            throws Exception {
        LostClaimStore store = new LostClaimStore();
        Trigger trigger = futureTrigger(
                "claim-transferred");
        store.save(trigger);
        TriggerScheduler scheduler = scheduler(store);
        AtomicInteger convergenceAttempts =
                new AtomicInteger();
        CountDownLatch executed = new CountDownLatch(1);
        scheduler.registerConsumer((current, worker) -> {
            executed.countDown();
            throw new NonRetryableTriggerException(
                    "stale owner");
        });
        scheduler.registerFailureListener(
                (current, failure) -> {
                    convergenceAttempts.incrementAndGet();
                    return true;
                });

        try {
            Assert.assertTrue(scheduler.fire(
                    trigger.getId()));
            Assert.assertTrue(executed.await(
                    2, TimeUnit.SECONDS));
            Assert.assertTrue(store.finalizationRejected.await(
                    2, TimeUnit.SECONDS));
            Assert.assertEquals(
                    0, convergenceAttempts.get());
            Assert.assertEquals(
                    0, store.deadLetterAttempts.get());
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证远期触发器只在固定容量内建立精确定时，溢出任务继续由持久仓储保留。
     *
     * @throws Exception 反射读取调度状态失败时抛出
     */
    @Test
    public void shouldBoundFutureTriggerLocalSchedules() throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        TriggerScheduler scheduler = scheduler(store);
        try {
            for (int index = 0; index < 2000; index++) {
                scheduler.schedule(nearFutureTrigger(
                        "future-" + index,
                        TimeUnit.SECONDS.toMillis(50)));
            }
            Field field = TriggerScheduler.class.getDeclaredField(
                    "scheduledFutures");
            field.setAccessible(true);
            @SuppressWarnings("unchecked")
            Map<String, ?> localFutures =
                    (Map<String, ?>) field.get(scheduler);
            Assert.assertEquals(1024, localFutures.size());
            Assert.assertEquals(2000, store.findAllPending().size());
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证并发登记远期触发器时本地定时表仍严格受容量上限约束。
     *
     * @throws Exception 并发调度或反射读取状态失败时抛出
     */
    @Test
    public void shouldBoundConcurrentFutureTriggerLocalSchedules()
            throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        TriggerScheduler scheduler = scheduler(store);
        ExecutorService callers =
                Executors.newFixedThreadPool(16);
        List<java.util.concurrent.Callable<Void>> tasks =
                new ArrayList<>();
        for (int index = 0; index < 2000; index++) {
            int triggerIndex = index;
            tasks.add(() -> {
                scheduler.schedule(nearFutureTrigger(
                        "concurrent-future-" + triggerIndex,
                        TimeUnit.SECONDS.toMillis(50)));
                return null;
            });
        }
        try {
            callers.invokeAll(tasks).forEach(result -> {
                try {
                    result.get();
                } catch (Exception error) {
                    throw new AssertionError(
                            "concurrent schedule failed",
                            error);
                }
            });
            Field field = TriggerScheduler.class.getDeclaredField(
                    "scheduledFutures");
            field.setAccessible(true);
            @SuppressWarnings("unchecked")
            Map<String, ?> localFutures =
                    (Map<String, ?>) field.get(scheduler);
            Assert.assertEquals(1024, localFutures.size());
            Assert.assertEquals(
                    2000,
                    store.findAllPending().size());
        } finally {
            callers.shutdownNow();
            scheduler.shutdown();
        }
    }

    /**
     * 验证高并发零延迟触发完成后 Future 与时间索引同时清空。
     *
     * @throws Exception 等待任务完成或反射读取索引失败时抛出
     */
    @Test
    public void shouldNotLeakImmediateTriggerTimeIndex()
            throws Exception {
        InMemoryTriggerStore store =
                new InMemoryTriggerStore();
        TriggerScheduler scheduler =
                scheduler(store);
        int triggerCount = 1000;
        CountDownLatch consumed =
                new CountDownLatch(triggerCount);
        scheduler.registerConsumer(
                (current, worker) ->
                        consumed.countDown());

        try {
            for (int index = 0;
                 index < triggerCount;
                 index++) {
                scheduler.schedule(
                        dueTrigger(
                                "immediate-" + index));
            }
            Assert.assertTrue(
                    consumed.await(
                            5L,
                            TimeUnit.SECONDS));
            awaitLocalScheduleIndexEmpty(
                    scheduler,
                    "scheduledFutures");
            awaitLocalScheduleIndexEmpty(
                    scheduler,
                    "scheduledTriggerTimes");
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证近未来触发器按目标时间执行，不附加完整扫描周期。
     *
     * @throws Exception 等待异步触发被中断时抛出
     */
    @Test
    public void shouldDispatchFutureTriggerBeforeNextScan() throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        TriggerScheduler scheduler = scheduler(store);
        CountDownLatch consumed = new CountDownLatch(1);
        scheduler.registerConsumer((current, worker) ->
                consumed.countDown());
        Trigger trigger = new Trigger();
        trigger.setId("future-exact");
        trigger.setTriggerAt(
                System.currentTimeMillis() + 200L);

        try {
            scheduler.schedule(trigger);
            Assert.assertTrue(consumed.await(
                    700L, TimeUnit.MILLISECONDS));
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证较晚任务占满本地定时容量时，更早到期的任务仍获得精确定时槽位。
     *
     * @throws Exception 等待异步触发被中断时抛出
     */
    @Test
    public void shouldPreferNearTriggerWhenLocalScheduleIsFull()
            throws Exception {
        InMemoryTriggerStore store =
                new InMemoryTriggerStore();
        TriggerScheduler scheduler = scheduler(store);
        CountDownLatch consumed = new CountDownLatch(1);
        scheduler.registerConsumer((current, worker) -> {
            if ("near-deadline".equals(
                    current.getId())) {
                consumed.countDown();
            }
        });
        try {
            for (int index = 0; index < 1024; index++) {
                scheduler.schedule(nearFutureTrigger(
                        "later-" + index,
                        TimeUnit.SECONDS.toMillis(50)));
            }
            scheduler.schedule(nearFutureTrigger(
                    "near-deadline", 200L));

            Assert.assertTrue(
                    "near trigger should replace a later local timer",
                    consumed.await(
                            700L,
                            TimeUnit.MILLISECONDS));
            Assert.assertNull(
                    store.find("near-deadline"));
            Assert.assertEquals(
                    1024,
                    store.findAllPending().size());
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证两个调度器并发登记同一稳定入口时，仓储只创建一份待执行触发器。
     *
     * @throws Exception 并发调用失败时抛出
     */
    @Test
    public void shouldAtomicallyScheduleStableTriggerOnce()
            throws Exception {
        InMemoryTriggerStore store =
                new InMemoryTriggerStore();
        TriggerScheduler first = scheduler(store);
        TriggerScheduler second = scheduler(store);
        ExecutorService callers =
                Executors.newFixedThreadPool(2);
        CountDownLatch ready = new CountDownLatch(2);
        CountDownLatch start = new CountDownLatch(1);
        try {
            java.util.concurrent.Future<Trigger>
                    firstResult = callers.submit(() -> {
                        ready.countDown();
                        start.await();
                        return first.scheduleIfAbsent(
                                futureTrigger(
                                        "stable-entry"));
                    });
            java.util.concurrent.Future<Trigger>
                    secondResult = callers.submit(() -> {
                        ready.countDown();
                        start.await();
                        return second.scheduleIfAbsent(
                                futureTrigger(
                                        "stable-entry"));
                    });
            Assert.assertTrue(
                    ready.await(1L, TimeUnit.SECONDS));
            start.countDown();
            for (java.util.concurrent.Future<Trigger> result
                    : Arrays.asList(
                            firstResult,
                            secondResult)) {
                Assert.assertEquals(
                        "stable-entry",
                        result.get().getId());
            }
            Assert.assertEquals(
                    1,
                    store.findAllPending().size());
        } finally {
            start.countDown();
            callers.shutdownNow();
            first.shutdown();
            second.shutdown();
        }
    }

    /**
     * 验证工作线程容量暂时耗尽后，持久化触发器可由后续扫描自动恢复执行。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldRescheduleTriggerAfterDispatchCapacityRecovers() throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(2);
        ThreadPoolExecutor worker = new ThreadPoolExecutor(
                1,
                1,
                0L,
                TimeUnit.MILLISECONDS,
                new SynchronousQueue<>());
        TriggerScheduler scheduler =
                new TriggerScheduler(store, scheduledExecutor, worker, 1000L);
        CountDownLatch firstStarted = new CountDownLatch(1);
        CountDownLatch releaseFirst = new CountDownLatch(1);
        CountDownLatch secondConsumed = new CountDownLatch(1);
        scheduler.registerConsumer((current, executor) -> {
            if ("capacity-first".equals(current.getId())) {
                firstStarted.countDown();
                try {
                    releaseFirst.await(2, TimeUnit.SECONDS);
                } catch (InterruptedException exception) {
                    Thread.currentThread().interrupt();
                    throw new IllegalStateException("test consumer interrupted", exception);
                }
            } else if ("capacity-second".equals(current.getId())) {
                secondConsumed.countDown();
            }
        });

        try {
            scheduler.schedule(dueTrigger("capacity-first"));
            Assert.assertTrue(firstStarted.await(2, TimeUnit.SECONDS));
            scheduler.schedule(dueTrigger("capacity-second"));
            Thread.sleep(100L);
            Assert.assertNotNull(store.find("capacity-second"));

            releaseFirst.countDown();
            Assert.assertTrue(
                    "pending trigger should be rescheduled after capacity recovers",
                    secondConsumed.await(3, TimeUnit.SECONDS));
        } finally {
            releaseFirst.countDown();
            scheduler.shutdown();
        }
    }

    /**
     * 验证本地调度表被已完成占位填满后，会先清扫并继续接收新任务。
     *
     * @throws Exception 反射或异步等待失败时抛出
     */
    @Test
    public void shouldPruneCompletedSchedulesBeforeCapacityCheck() throws Exception {
        InMemoryTriggerStore store = new InMemoryTriggerStore();
        ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(2);
        ExecutorService worker = Executors.newFixedThreadPool(2);
        TriggerScheduler scheduler =
                new TriggerScheduler(store, scheduledExecutor, worker, 1000L);
        CountDownLatch consumed = new CountDownLatch(1);
        scheduler.registerConsumer((current, executor) -> consumed.countDown());

        Field field = TriggerScheduler.class.getDeclaredField("scheduledFutures");
        field.setAccessible(true);
        @SuppressWarnings("unchecked")
        Map<String, ScheduledFuture<?>> localFutures =
                (Map<String, ScheduledFuture<?>>) field.get(scheduler);
        ScheduledFuture<?> completed = scheduledExecutor.schedule(
                () -> {
                },
                0L,
                TimeUnit.MILLISECONDS);
        completed.get(1, TimeUnit.SECONDS);
        for (int index = 0; index < 1024; index++) {
            localFutures.put("completed-" + index, completed);
        }

        try {
            scheduler.schedule(dueTrigger("after-completed-capacity"));
            Assert.assertTrue(
                    "completed local schedules should not block new due triggers",
                    consumed.await(2, TimeUnit.SECONDS));
            Assert.assertTrue(localFutures.size() < 1024);
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证确认成功状态写入异常时仍归还调度容量许可。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldReturnDispatchPermitWhenAcknowledgeFails() throws Exception {
        assertTerminalStoreFailureDoesNotLeakPermit(TerminalOperation.ACKNOWLEDGE);
    }

    /**
     * 验证失败重投状态写入异常时仍归还调度容量许可。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldReturnDispatchPermitWhenReleaseFails() throws Exception {
        assertTerminalStoreFailureDoesNotLeakPermit(TerminalOperation.RELEASE);
    }

    /**
     * 验证死信状态写入异常时仍归还调度容量许可。
     *
     * @throws Exception 等待异步执行被中断时抛出
     */
    @Test
    public void shouldReturnDispatchPermitWhenDeadLetterFails() throws Exception {
        assertTerminalStoreFailureDoesNotLeakPermit(TerminalOperation.DEAD_LETTER);
    }

    /**
     * 执行终态仓储异常后的容量许可回归场景。
     *
     * @param operation 需要模拟失败的终态操作
     * @throws Exception 等待异步执行被中断时抛出
     */
    private void assertTerminalStoreFailureDoesNotLeakPermit(
            TerminalOperation operation) throws Exception {
        FailingTerminalStore store = new FailingTerminalStore(operation);
        ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(2);
        ThreadPoolExecutor worker = new ThreadPoolExecutor(
                1,
                1,
                0L,
                TimeUnit.MILLISECONDS,
                new SynchronousQueue<>());
        TriggerScheduler scheduler =
                new TriggerScheduler(store, scheduledExecutor, worker, 1000L);
        CountDownLatch secondConsumed = new CountDownLatch(1);
        scheduler.registerConsumer((current, executor) -> {
            if ("second".equals(current.getId())) {
                secondConsumed.countDown();
                return;
            }
            if (operation == TerminalOperation.DEAD_LETTER) {
                throw new NonRetryableTriggerException("expected dead-letter failure");
            }
            if (operation == TerminalOperation.RELEASE) {
                throw new IllegalStateException("expected release failure");
            }
        });

        try {
            Trigger first = futureTrigger("first");
            store.save(first);
            Assert.assertTrue(scheduler.fire(first.getId()));
            Assert.assertTrue(store.terminalAttempted.await(2, TimeUnit.SECONDS));

            store.fail = false;
            Trigger second = futureTrigger("second");
            store.save(second);
            boolean accepted = false;
            long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
            while (!accepted && System.nanoTime() < deadline) {
                accepted = scheduler.fire(second.getId());
                if (!accepted) {
                    Thread.sleep(10L);
                }
            }
            Assert.assertTrue("dispatch permit should be returned", accepted);
            Assert.assertTrue(secondConsumed.await(2, TimeUnit.SECONDS));
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 创建测试调度器。
     *
     * @param store 触发器仓储
     * @return 测试调度器
     */
    private TriggerScheduler scheduler(TriggerStore store) {
        ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(2);
        ExecutorService worker = Executors.newFixedThreadPool(2);
        return new TriggerScheduler(store, scheduledExecutor, worker, 1000L);
    }

    /**
     * 等待指定本地调度索引清空。
     *
     * @param scheduler 调度器
     * @param fieldName 索引字段名
     * @throws Exception 反射读取失败或等待超时时抛出
     */
    private void awaitLocalScheduleIndexEmpty(
            TriggerScheduler scheduler,
            String fieldName) throws Exception {
        Field field =
                TriggerScheduler.class.getDeclaredField(
                        fieldName);
        field.setAccessible(true);
        @SuppressWarnings("unchecked")
        Map<String, ?> index =
                (Map<String, ?>) field.get(scheduler);
        long deadline =
                System.nanoTime()
                        + TimeUnit.SECONDS.toNanos(2L);
        while (!index.isEmpty()
                && System.nanoTime() < deadline) {
            Thread.sleep(10L);
        }
        Assert.assertTrue(
                fieldName + " should be empty",
                index.isEmpty());
    }

    /**
     * 创建远期触发器，避免定时任务干扰主动认领测试。
     *
     * @param id 触发器 ID
     * @return 测试触发器
     */
    private Trigger futureTrigger(String id) {
        return nearFutureTrigger(
                id,
                TimeUnit.MINUTES.toMillis(5));
    }

    /**
     * 创建指定延迟的测试触发器。
     *
     * @param id 触发器 ID
     * @param delayMillis 延迟毫秒数
     * @return 测试触发器
     */
    private Trigger nearFutureTrigger(
            String id,
            long delayMillis) {
        Trigger trigger = new Trigger();
        trigger.setId(id);
        trigger.setTriggerAt(
                System.currentTimeMillis()
                        + delayMillis);
        return trigger;
    }

    /**
     * 创建立即到期的测试触发器。
     *
     * @param id 触发器 ID
     * @return 测试触发器
     */
    private Trigger dueTrigger(String id) {
        Trigger trigger = new Trigger();
        trigger.setId(id);
        trigger.setTriggerAt(System.currentTimeMillis());
        return trigger;
    }

    /**
     * 等待失败触发器重新进入待执行状态。
     *
     * @param store 触发器仓储
     * @param triggerId 触发器 ID
     * @throws InterruptedException 等待被中断时抛出
     */
    private void awaitPending(InMemoryTriggerStore store, String triggerId) throws InterruptedException {
        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
        while (store.find(triggerId) == null && System.nanoTime() < deadline) {
            Thread.sleep(10L);
        }
        Assert.assertNotNull(store.find(triggerId));
    }

    /**
     * 需要模拟失败的触发器终态操作。
     */
    private enum TerminalOperation {
        ACKNOWLEDGE,
        RELEASE,
        DEAD_LETTER
    }

    /**
     * 在指定终态操作上抛出异常的进程内触发器仓储。
     */
    private static final class FailingTerminalStore extends InMemoryTriggerStore {

        private final TerminalOperation operation;
        private final CountDownLatch terminalAttempted = new CountDownLatch(1);
        private volatile boolean fail = true;

        /**
         * 创建失败仓储。
         *
         * @param operation 需要失败的终态操作
         */
        private FailingTerminalStore(TerminalOperation operation) {
            this.operation = operation;
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public void acknowledge(Trigger trigger) {
            failIfNeeded(TerminalOperation.ACKNOWLEDGE);
            super.acknowledge(trigger);
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public void release(Trigger trigger) {
            failIfNeeded(TerminalOperation.RELEASE);
            super.release(trigger);
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public void deadLetter(Trigger trigger, String reason) {
            failIfNeeded(TerminalOperation.DEAD_LETTER);
            super.deadLetter(trigger, reason);
        }

        /**
         * 在当前测试操作上抛出预期异常。
         *
         * @param current 当前终态操作
         */
        private void failIfNeeded(TerminalOperation current) {
            if (fail && operation == current) {
                terminalAttempted.countDown();
                throw new IllegalStateException("expected terminal store failure: " + current);
            }
        }
    }

    /**
     * 模拟 claim 已转移给其他 owner 的触发器仓储。
     */
    private static final class LostClaimStore
            extends InMemoryTriggerStore {

        private final CountDownLatch finalizationRejected =
                new CountDownLatch(1);
        private final AtomicInteger deadLetterAttempts =
                new AtomicInteger();

        /**
         * {@inheritDoc}
         */
        @Override
        public boolean renewClaim(
                Trigger trigger, long leaseMillis) {
            finalizationRejected.countDown();
            return false;
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public void deadLetter(
                Trigger trigger, String reason) {
            deadLetterAttempts.incrementAndGet();
        }
    }
}
