/**
 * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
 * <p>
 * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.gnu.org/licenses/lgpl-3.0.txt
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.easyagents.flow.core.chain.runtime;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 *
 * 功能:
 * - schedule trigger (持久化到 TriggerStore 并 schedule)
 * - cancel trigger
 * - fire(triggerId) 用于 webhook/event/manual 主动触发
 * - recoverAndSchedulePending() 启动时恢复未执行的 trigger
 * - periodical scan findDue(upto) 以保证重启/宕机后补偿触发
 * <p>
 * 注意: 分布式环境下需要在 TriggerStore 层提供抢占/锁逻辑（例如 lease/owner 字段）。
 */
public class TriggerScheduler {

    private static final Logger log = LoggerFactory.getLogger(TriggerScheduler.class);
    private static final long CLAIM_LEASE_MS = TimeUnit.MINUTES.toMillis(1);
    private static final long CLAIM_RENEW_INTERVAL_MS = TimeUnit.SECONDS.toMillis(20);
    private static final int MAX_DELIVERY_ATTEMPTS = 20;
    private static final long MAX_REDELIVERY_DELAY_MS = TimeUnit.MINUTES.toMillis(1);
    /**
     * 本地仅缓存一批定时任务；容量溢出时持久仓储仍是恢复与补偿来源。
     */
    private static final int MAX_LOCAL_SCHEDULED = 1024;
    /**
     * 精确定时只预热宽松近期限窗口，远期任务继续由持久仓储保管。
     */
    private static final long MIN_LOCAL_SCHEDULE_HORIZON_MS =
            TimeUnit.MINUTES.toMillis(1);
    private final TriggerStore store;
    private final ScheduledExecutorService scheduler;
    private final ExecutorService worker;
    private final Semaphore dispatchPermits;
    private final ConcurrentMap<String, ExecutorService> laneWorkers =
            new ConcurrentHashMap<>();
    private final ConcurrentMap<String, Semaphore> laneDispatchPermits =
            new ConcurrentHashMap<>();
    private final AtomicBoolean closed = new AtomicBoolean(false);

    // map 用于管理取消：triggerId -> ScheduledFuture
    private final ConcurrentMap<String, ScheduledFuture<?>> scheduledFutures = new ConcurrentHashMap<>();
    /**
     * 本地 Future 对应的绝对触发时间，用于容量满时保留更早到期任务。
     */
    private final ConcurrentMap<String, Long> scheduledTriggerTimes =
            new ConcurrentHashMap<>();
    private final ConcurrentMap<Trigger, ScheduledFuture<?>> claimRenewals = new ConcurrentHashMap<>();
    /**
     * 串行化本地 Future 的容量检查与登记，确保并发调度时仍严格受容量上限约束。
     */
    private final Object localScheduleMonitor = new Object();

    // consumer 来把 trigger 交给 ChainExecutor（或 ChainRuntime）去处理
    private volatile TriggerConsumer consumer;
    private volatile TriggerFailureListener failureListener;

    // 周期扫查间隔（ms）
    private final long scanIntervalMs;
    private final long localScheduleHorizonMs;

    // 扫描任务 future
    private ScheduledFuture<?> scanFuture;

    public interface TriggerConsumer {
        void accept(Trigger trigger, ExecutorService worker);
    }

    /**
     * 触发器不可恢复失败监听器。
     */
    public interface TriggerFailureListener {
        /**
         * 在触发器 claim 仍有效时通知业务运行时收敛实例终态。
         *
         * @param trigger 已死信触发器
         * @param failure 最后一次执行失败
         */
        boolean onDeadLetter(Trigger trigger, Throwable failure);
    }

    public TriggerScheduler(TriggerStore store, ScheduledExecutorService scheduler, ExecutorService worker, long scanIntervalMs) {
        this.store = Objects.requireNonNull(store, "TriggerStore required");
        this.scheduler = Objects.requireNonNull(scheduler, "ScheduledExecutorService required");
        this.worker = Objects.requireNonNull(worker, "ExecutorService required");
        this.dispatchPermits = createDispatchPermits(worker);
        this.scanIntervalMs = Math.max(1000, scanIntervalMs);
        long scanHorizon = this.scanIntervalMs
                > Long.MAX_VALUE / 3L
                ? Long.MAX_VALUE
                : this.scanIntervalMs * 3L;
        this.localScheduleHorizonMs = Math.max(
                MIN_LOCAL_SCHEDULE_HORIZON_MS,
                scanHorizon);

        // 启动时只恢复已经到期的任务，避免把全部远期任务复制到本机 DelayQueue。
        recoverAndSchedulePending();

        // 启动周期扫查 findDue
        startPeriodicScan();
    }


    public void registerConsumer(TriggerConsumer consumer) {
        this.consumer = consumer;
    }

    /**
     * 注册触发器不可恢复失败监听器。
     *
     * @param listener 失败监听器
     */
    public void registerFailureListener(TriggerFailureListener listener) {
        this.failureListener = listener;
    }

    /**
     * 注册独立执行通道。
     *
     * @param lane 通道名
     * @param laneWorker 通道工作线程池
     */
    public void registerWorker(
            String lane, ExecutorService laneWorker) {
        if (lane == null || lane.isBlank()) {
            throw new IllegalArgumentException("lane required");
        }
        ExecutorService workerToRegister =
                Objects.requireNonNull(laneWorker, "laneWorker required");
        ExecutorService previous =
                laneWorkers.putIfAbsent(lane, workerToRegister);
        if (previous != null && previous != workerToRegister) {
            throw new IllegalStateException(
                    "Trigger worker lane already registered: " + lane);
        }
        Semaphore permits = createDispatchPermits(workerToRegister);
        if (permits != null) {
            laneDispatchPermits.putIfAbsent(lane, permits);
        }
    }

    /**
     * schedule a trigger: persist -> schedule (单机语义)
     */
    public Trigger schedule(Trigger trigger) {
        if (closed.get()) throw new IllegalStateException("TriggerScheduler closed");
        if (trigger.getId() == null) {
            trigger.setId(UUID.randomUUID().toString());
        }
        if (trigger.getLogicalExecutionId() == null || trigger.getLogicalExecutionId().isBlank()) {
            trigger.setLogicalExecutionId(trigger.getId());
        }
        store.save(trigger);
        scheduleInternal(trigger);
        return trigger;
    }

    /**
     * 仅在持久仓储中不存在同 ID 触发器时保存并调度。
     *
     * <p>调用方需持有工作流实例锁；该方法用于可重放启动协议，稳定 ID 可避免
     * READY 到入口触发器持久化之间的崩溃窗口产生重复待执行任务。</p>
     *
     * @param trigger 带稳定 ID 的触发器
     * @return 已存在或新保存的触发器
     */
    public Trigger scheduleIfAbsent(Trigger trigger) {
        if (closed.get()) {
            throw new IllegalStateException("TriggerScheduler closed");
        }
        if (trigger == null || trigger.getId() == null
                || trigger.getId().isBlank()) {
            throw new IllegalArgumentException(
                    "Stable trigger ID required");
        }
        if (trigger.getLogicalExecutionId() == null
                || trigger.getLogicalExecutionId().isBlank()) {
            trigger.setLogicalExecutionId(
                    trigger.getId());
        }
        if (store.saveIfAbsent(trigger)) {
            scheduleInternal(trigger);
            return trigger;
        }
        Trigger existing = store.find(
                trigger.getId());
        // 已有触发器可能刚好被其他 owner 认领；稳定入口仍视为已成功登记。
        return existing == null
                ? trigger
                : existing;
    }

    /**
     * cancel trigger (从 store 删除并尝试取消已 schedule 的 future)
     */
    public boolean cancel(String triggerId) {
        boolean removed = store.remove(triggerId);
        removeLocalSchedule(triggerId, true);
        return removed;
    }

    /**
     * 主动触发（webhook/event/manual 场景）
     */
    public boolean fire(String triggerId) {
        if (closed.get()) {
            return false;
        }
        Trigger candidate = store.find(triggerId);
        if (candidate == null) {
            return false;
        }
        removeLocalSchedule(triggerId, true);
        return claimAndDispatch(candidate);
    }

    /**
     * internal scheduling for a trigger (单机 scheduled semantics)
     */
    private void scheduleInternal(Trigger trigger) {
        if (closed.get()) {
            return;
        }
        long now = System.currentTimeMillis();
        if (trigger.getTriggerAt()
                > scheduleHorizonTimestamp(now)) {
            return;
        }
        synchronized (localScheduleMonitor) {
            if (closed.get()) {
                return;
            }
            ScheduledFuture<?> existing;
            while ((existing = scheduledFutures.get(trigger.getId())) != null) {
                if (!existing.isDone() && !existing.isCancelled()) {
                    return;
                }
                // 已完成或取消的占位必须先原子移除，否则 putIfAbsent 会永久阻断重新调度。
                if (!scheduledFutures.remove(trigger.getId(), existing)) {
                    continue;
                }
                scheduledTriggerTimes.remove(
                        trigger.getId());
            }
            if (!hasLocalScheduleCapacity(
                    trigger.getTriggerAt())) {
                return;
            }
            long delayMillis = Math.max(
                    0L,
                    trigger.getTriggerAt()
                            - System.currentTimeMillis());
            ScheduledFuture<?> future = scheduler.schedule(
                    () -> {
                        // 回调先在同一登记锁下同时移除两张索引，消除零延迟任务
                        // 在 Future 与时间索引分步登记之间完成所造成的孤儿记录。
                        removeLocalSchedule(
                                trigger.getId(), false);
                        claimAndDispatch(trigger);
                    },
                    delayMillis,
                    TimeUnit.MILLISECONDS);
            ScheduledFuture<?> concurrent =
                    scheduledFutures.putIfAbsent(
                            trigger.getId(), future);
            if (concurrent != null) {
                future.cancel(false);
                return;
            }
            scheduledTriggerTimes.put(
                    trigger.getId(),
                    trigger.getTriggerAt());
            // 零延迟任务可能在 put 前完成；完成态二次清理避免残留无效 Future。
            if (future.isDone() || future.isCancelled()) {
                scheduledFutures.remove(
                        trigger.getId(), future);
                scheduledTriggerTimes.remove(
                        trigger.getId());
            }
        }
    }

    /**
     * 检查本地调度容量，并在容量耗尽时清理已完成或已取消的占位。
     *
     * <p>正常路径只执行常量时间判断；达到上限时最多扫描
     * {@link #MAX_LOCAL_SCHEDULED} 个条目，避免极窄竞态残留导致永久停摆。</p>
     *
     * @param triggerAt 待登记任务的绝对触发时间
     * @return 仍可接收本地到期任务时为 {@code true}
     */
    private boolean hasLocalScheduleCapacity(
            long triggerAt) {
        if (scheduledFutures.size() < MAX_LOCAL_SCHEDULED) {
            return true;
        }
        for (Map.Entry<String, ScheduledFuture<?>> entry : scheduledFutures.entrySet()) {
            ScheduledFuture<?> future = entry.getValue();
            if (future.isDone() || future.isCancelled()) {
                if (scheduledFutures.remove(
                        entry.getKey(), future)) {
                    scheduledTriggerTimes.remove(
                            entry.getKey());
                }
            }
        }
        if (scheduledFutures.size()
                < MAX_LOCAL_SCHEDULED) {
            return true;
        }
        Map.Entry<String, Long> latest = null;
        for (Map.Entry<String, Long> entry
                : scheduledTriggerTimes.entrySet()) {
            if (latest == null
                    || entry.getValue()
                    > latest.getValue()) {
                latest = entry;
            }
        }
        if (latest == null
                || latest.getValue() <= triggerAt) {
            return false;
        }
        removeLocalSchedule(
                latest.getKey(), true);
        return scheduledFutures.size()
                < MAX_LOCAL_SCHEDULED;
    }

    /**
     * 计算本轮应预热到本机的最远触发时间。
     *
     * @return 当前时间加本地近期限窗口；溢出时为最大时间戳
     */
    private long scheduleHorizonTimestamp() {
        return scheduleHorizonTimestamp(
                System.currentTimeMillis());
    }

    /**
     * 基于给定时间计算本机预热边界。
     *
     * @param now 当前时间戳
     * @return 当前时间加近期限窗口；溢出时为最大时间戳
     */
    private long scheduleHorizonTimestamp(
            long now) {
        return now > Long.MAX_VALUE
                - localScheduleHorizonMs
                ? Long.MAX_VALUE
                : now + localScheduleHorizonMs;
    }

    /**
     * 同时移除本地 Future 与其触发时间索引。
     *
     * @param triggerId 触发器 ID
     * @param cancel 是否取消尚未完成的 Future
     * @return 被移除的 Future；不存在时为 {@code null}
     */
    private ScheduledFuture<?> removeLocalSchedule(
            String triggerId,
            boolean cancel) {
        synchronized (localScheduleMonitor) {
            ScheduledFuture<?> scheduled =
                    scheduledFutures.remove(triggerId);
            scheduledTriggerTimes.remove(triggerId);
            if (cancel && scheduled != null) {
                scheduled.cancel(false);
            }
            return scheduled;
        }
    }

    private void recoverAndSchedulePending() {
        try {
            List<Trigger> list = store.findDue(
                    scheduleHorizonTimestamp());
            if (list == null || list.isEmpty()) return;
            for (Trigger t : list) {
                scheduleInternal(t);
            }
        } catch (Throwable t) {
            log.error("Failed to recover pending workflow triggers", t);
        }
    }

    private void startPeriodicScan() {
        if (closed.get()) return;
        scanFuture = scheduler.scheduleAtFixedRate(() -> {
            try {
                long upto = scheduleHorizonTimestamp();
                List<Trigger> due = store.findDue(upto);
                if (due == null || due.isEmpty()) return;
                for (Trigger t : due) {
                    // 如果已被 schedule 到未来（scheduledFutures 包含且尚未到期），跳过
                    ScheduledFuture<?> sf = scheduledFutures.get(t.getId());
                    if (sf != null && !sf.isDone() && !sf.isCancelled()) {
                        continue;
                    }
                    scheduleInternal(t);
                }
            } catch (Throwable tt) {
                log.error("Failed to scan due workflow triggers", tt);
            }
        }, scanIntervalMs, scanIntervalMs, TimeUnit.MILLISECONDS);
    }

    /**
     * 原子认领触发器并提交给工作线程。
     *
     * @param candidate 已加载的候选触发器
     * @return 成功认领并提交时为 true
     */
    private boolean claimAndDispatch(Trigger candidate) {
        if (candidate == null || candidate.getId() == null) {
            return false;
        }
        String triggerId = candidate.getId();
        ExecutorService dispatchWorker = resolveWorker(candidate);
        Semaphore selectedPermits = resolveDispatchPermits(candidate);
        if (selectedPermits != null && !selectedPermits.tryAcquire()) {
            // 保留持久化触发器并清理本地占位，让后续扫描能够重新调度。
            removeLocalSchedule(triggerId, false);
            return false;
        }
        Trigger claimed;
        try {
            claimed = store.claim(candidate, CLAIM_LEASE_MS);
        } catch (RuntimeException | Error error) {
            removeLocalSchedule(triggerId, false);
            releaseDispatchPermit(selectedPermits);
            throw error;
        }
        if (claimed == null) {
            releaseDispatchPermit(selectedPermits);
            removeLocalSchedule(triggerId, false);
            return false;
        }
        TriggerConsumer currentConsumer = consumer;
        if (currentConsumer == null) {
            releaseClaimBestEffort(claimed, "consumer is unavailable");
            releaseDispatchPermit(selectedPermits);
            removeLocalSchedule(triggerId, false);
            return false;
        }

        ScheduledFuture<?> renewal;
        try {
            renewal = scheduler.scheduleAtFixedRate(
                    () -> renewClaim(claimed),
                    CLAIM_RENEW_INTERVAL_MS,
                    CLAIM_RENEW_INTERVAL_MS,
                    TimeUnit.MILLISECONDS);
        } catch (RejectedExecutionException error) {
            releaseClaimBestEffort(claimed, "claim renewal scheduling was rejected");
            releaseDispatchPermit(selectedPermits);
            removeLocalSchedule(triggerId, false);
            return false;
        }
        ScheduledFuture<?> existingRenewal = claimRenewals.put(claimed, renewal);
        if (existingRenewal != null) {
            existingRenewal.cancel(false);
        }

        try {
            dispatchWorker.submit(() -> consumeClaimedTrigger(
                    claimed,
                    currentConsumer,
                    dispatchWorker,
                    selectedPermits));
            return true;
        } catch (RejectedExecutionException error) {
            cancelClaimRenewal(claimed);
            releaseClaimBestEffort(claimed, "worker submission was rejected");
            releaseDispatchPermit(selectedPermits);
            removeLocalSchedule(triggerId, false);
            log.error("Workflow trigger worker rejected task, triggerId={}", triggerId, error);
            return false;
        }
    }

    /**
     * 执行已认领触发器，并按执行结果确认或释放。
     *
     * @param trigger 已认领触发器
     * @param currentConsumer 本次执行使用的消费者快照
     */
    private void consumeClaimedTrigger(
            Trigger trigger,
            TriggerConsumer currentConsumer,
            ExecutorService dispatchWorker,
            Semaphore selectedPermits) {
        boolean succeeded = false;
        boolean deadLetter = false;
        boolean retryWithoutDeadLetter = false;
        boolean rescheduleReleasedTrigger = false;
        Throwable failure = null;
        try {
            TriggerContext.setCurrentTrigger(trigger);
            if (trigger.isDeadLetterPending()) {
                deadLetter = true;
                failure = new NonRetryableTriggerException(
                        trigger.getDeadLetterReason());
            } else {
                currentConsumer.accept(trigger, dispatchWorker);
                succeeded = true;
            }
        } catch (NonRetryableTriggerException error) {
            failure = error;
            deadLetter = true;
            log.error("Workflow trigger is not retryable, triggerId={}", trigger.getId(), error);
        } catch (RetryableTriggerException error) {
            failure = error;
            retryWithoutDeadLetter = true;
            log.warn(
                    "Workflow trigger hit transient infrastructure contention, triggerId={}",
                    trigger.getId());
        } catch (Throwable error) {
            failure = error;
            log.error("Workflow trigger execution failed, triggerId={}", trigger.getId(), error);
        } finally {
            try {
                if (succeeded) {
                    store.acknowledge(trigger);
                } else {
                    int deliveryAttempt = trigger.isDeadLetterPending()
                            ? Math.max(1, trigger.getDeliveryAttempt())
                            : incrementDeliveryAttempt(trigger);
                    if (retryWithoutDeadLetter) {
                        trigger.setTriggerAt(
                                System.currentTimeMillis()
                                        + redeliveryDelayMillis(
                                                deliveryAttempt));
                        store.release(trigger);
                        rescheduleReleasedTrigger = true;
                    } else if (deadLetter
                            || deliveryAttempt
                            >= MAX_DELIVERY_ATTEMPTS) {
                        String reason = failure == null
                                ? "delivery attempts exhausted"
                                : failure.getClass().getName()
                                + ": "
                                + failure.getMessage();
                        if (trigger.isDeadLetterPending()
                                && trigger.getDeadLetterReason() != null) {
                            reason = trigger.getDeadLetterReason();
                        } else {
                            trigger.setDeadLetterPending(true);
                            trigger.setDeadLetterReason(reason);
                            store.markDeadLetterPending(
                                    trigger);
                        }
                        if (!store.renewClaim(
                                trigger, CLAIM_LEASE_MS)) {
                            throw new TriggerClaimLostException(
                                    trigger.getId());
                        }
                        TriggerFailureListener currentFailureListener =
                                failureListener;
                        if (currentFailureListener != null
                                && !currentFailureListener.onDeadLetter(
                                        trigger, failure)) {
                            trigger.setTriggerAt(
                                    System.currentTimeMillis()
                                            + MAX_REDELIVERY_DELAY_MS);
                            store.release(trigger);
                            rescheduleReleasedTrigger = true;
                            return;
                        }
                        store.deadLetter(trigger, reason);
                    } else {
                        trigger.setTriggerAt(
                                System.currentTimeMillis()
                                        + redeliveryDelayMillis(
                                        deliveryAttempt));
                        store.release(trigger);
                        rescheduleReleasedTrigger = true;
                    }
                }
            } catch (Throwable terminalStoreError) {
                if (trigger.isDeadLetterPending()) {
                    releaseDeadLetterPendingBestEffort(trigger);
                }
                log.error(
                        "Failed to finalize workflow trigger, triggerId={}",
                        trigger.getId(),
                        terminalStoreError);
            } finally {
                TriggerContext.clearCurrentTrigger();
                cancelClaimRenewal(trigger);
                removeLocalSchedule(
                        trigger.getId(), false);
                releaseDispatchPermit(selectedPermits);
                if (rescheduleReleasedTrigger) {
                    scheduleInternal(trigger);
                }
            }
        }
    }

    /**
     * 在仍持有 claim 时保存待补写死信标记，避免业务终态成功后被普通 ACK 吞掉。
     *
     * @param trigger 待补写死信的触发器
     */
    private void releaseDeadLetterPendingBestEffort(
            Trigger trigger) {
        try {
            if (!store.renewClaim(trigger, CLAIM_LEASE_MS)) {
                return;
            }
            trigger.setTriggerAt(
                    System.currentTimeMillis()
                            + MAX_REDELIVERY_DELAY_MS);
            store.release(trigger);
        } catch (Throwable releaseError) {
            log.error(
                    "Failed to persist pending dead-letter finalization, triggerId={}",
                    trigger.getId(),
                    releaseError);
        }
    }

    /**
     * 尽力释放已认领触发器。
     *
     * <p>释放失败时保留 Redis 中的触发器和租约，等待租约自然过期恢复。调用方仍可继续
     * 清理本地调度状态和归还容量许可。</p>
     *
     * @param trigger 已认领触发器
     * @param reason 释放原因
     */
    private void releaseClaimBestEffort(Trigger trigger, String reason) {
        try {
            store.release(trigger);
        } catch (Throwable error) {
            log.error(
                    "Failed to release workflow trigger, triggerId={}, reason={}",
                    trigger == null ? null : trigger.getId(),
                    reason,
                    error);
        }
    }

    /**
     * 增加基础设施投递失败次数。
     *
     * @param trigger 当前触发器
     * @return 增加后的次数
     */
    private int incrementDeliveryAttempt(Trigger trigger) {
        int attempt = Math.max(0, trigger.getDeliveryAttempt()) + 1;
        trigger.setDeliveryAttempt(attempt);
        return attempt;
    }

    /**
     * 计算带上限的指数退避，避免缺失定义或短暂冲突形成热循环。
     *
     * @param attempt 已失败次数
     * @return 下次投递延迟毫秒数
     */
    private long redeliveryDelayMillis(int attempt) {
        int shift = Math.min(16, Math.max(0, attempt - 1));
        long delay = 1_000L << shift;
        return Math.min(MAX_REDELIVERY_DELAY_MS, delay);
    }

    /**
     * 根据工作线程池的真实容量建立领取前背压。
     *
     * @param executor 工作线程池
     * @return 容量信号量；无法识别容量时返回 null
     */
    private Semaphore createDispatchPermits(ExecutorService executor) {
        if (!(executor instanceof ThreadPoolExecutor)) {
            return null;
        }
        ThreadPoolExecutor pool = (ThreadPoolExecutor) executor;
        long capacity = (long) pool.getMaximumPoolSize() + pool.getQueue().remainingCapacity();
        return new Semaphore((int) Math.max(1L, Math.min(Integer.MAX_VALUE, capacity)));
    }

    /**
     * 释放一个工作线程容量许可。
     */
    private void releaseDispatchPermit(Semaphore permits) {
        if (permits != null) {
            permits.release();
        }
    }

    /**
     * 解析触发器执行线程池。
     *
     * @param trigger 触发器
     * @return 默认或独立通道线程池
     */
    private ExecutorService resolveWorker(Trigger trigger) {
        String lane = trigger == null ? null : trigger.getExecutionLane();
        return lane == null ? worker : laneWorkers.getOrDefault(lane, worker);
    }

    /**
     * 解析所选线程池对应的领取前背压许可。
     *
     * @param trigger 触发器
     * @return 容量许可；无法识别时为 {@code null}
     */
    private Semaphore resolveDispatchPermits(Trigger trigger) {
        String lane = trigger == null ? null : trigger.getExecutionLane();
        return lane == null
                ? dispatchPermits
                : laneDispatchPermits.getOrDefault(lane, dispatchPermits);
    }

    /**
     * 续期触发器认领租约。
     *
     * @param trigger 已认领触发器
     */
    private void renewClaim(Trigger trigger) {
        try {
            if (!store.renewClaim(trigger, CLAIM_LEASE_MS)) {
                log.warn("Workflow trigger claim renewal lost ownership, triggerId={}", trigger.getId());
                cancelClaimRenewal(trigger);
            }
        } catch (Throwable error) {
            log.error("Workflow trigger claim renewal failed, triggerId={}", trigger.getId(), error);
        }
    }

    /**
     * 在业务状态提交前验证具体触发器仍由当前工作线程持有。
     *
     * @param trigger 已认领触发器
     * @throws TriggerClaimLostException 租约已失效或已转移
     */
    public void assertClaimOwned(Trigger trigger) {
        if (trigger != null && !store.renewClaim(trigger, CLAIM_LEASE_MS)) {
            throw new TriggerClaimLostException(trigger.getId());
        }
    }

    /**
     * 取消触发器租约续期任务。
     *
     * @param trigger 已认领触发器
     */
    private void cancelClaimRenewal(Trigger trigger) {
        ScheduledFuture<?> renewal = claimRenewals.remove(trigger);
        if (renewal != null) {
            renewal.cancel(false);
        }
    }

    public void shutdown() {
        if (closed.compareAndSet(false, true)) {
            if (scanFuture != null) scanFuture.cancel(false);
            // cancel scheduled futures
            for (Map.Entry<String, ScheduledFuture<?>> e : scheduledFutures.entrySet()) {
                try {
                    e.getValue().cancel(false);
                } catch (Throwable ignored) {
                }
            }
            scheduledFutures.clear();
            scheduledTriggerTimes.clear();
            for (ScheduledFuture<?> renewal : claimRenewals.values()) {
                renewal.cancel(false);
            }
            claimRenewals.clear();

            try {
                scheduler.shutdownNow();
            } catch (Throwable ignored) {
            }
            try {
                worker.shutdownNow();
            } catch (Throwable ignored) {
            }
            for (ExecutorService laneWorker :
                    new java.util.HashSet<>(laneWorkers.values())) {
                if (laneWorker == worker) {
                    continue;
                }
                try {
                    laneWorker.shutdownNow();
                } catch (Throwable ignored) {
                }
            }
            laneWorkers.clear();
            laneDispatchPermits.clear();
        }
    }
}
