package tech.easyflow.ai.easyagentsflow.repository;

import com.alicp.jetcache.support.JavaValueEncoder;
import com.easyagents.flow.core.chain.repository.LoopInputReference;
import com.easyagents.flow.core.chain.repository.LoopResultRepository;
import com.easyagents.flow.core.chain.runtime.ExecutionBudgetExceededException;
import com.easyagents.flow.core.chain.runtime.TriggerClaimLostException;
import org.springframework.stereotype.Component;
import tech.easyflow.common.cache.ApplicationClassLoaderJavaValueDecoder;
import tech.easyflow.common.cache.VersionedObjectStore;
import tech.easyflow.common.constant.CacheKey;

import javax.annotation.Resource;
import java.io.Serializable;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

/**
 * 基于分块缓存的循环累计结果仓储。
 * <p>
 * 每轮只重写当前固定大小分块，避免循环历史结果随节点热状态反复序列化。
 */
@Component
public class LoopResultRepositoryImpl extends BaseRepository implements LoopResultRepository {

    static final int CHUNK_SIZE = 128;
    private static final Duration RESULT_TTL = Duration.ofDays(7);
    private static final long REFRESH_INTERVAL_MILLIS =
            Duration.ofHours(12).toMillis();
    private static final int MAX_CACHED_INPUT_RESULTS = 256;
    private static final int MAX_CACHED_OUTPUT_RESULTS = 256;
    private static final long MAX_INPUT_CACHE_WEIGHT_BYTES =
            32L * 1024L * 1024L;
    private static final long MAX_OUTPUT_CACHE_WEIGHT_BYTES =
            64L * 1024L * 1024L;
    private static final long MAX_CACHE_ENTRY_WEIGHT_BYTES =
            8L * 1024L * 1024L;
    private static final long EMPTY_CHUNK_WEIGHT_BYTES = 256L;
    private static final ApplicationClassLoaderJavaValueDecoder
            CACHE_VALUE_DECODER =
            new ApplicationClassLoaderJavaValueDecoder();

    @Resource
    private VersionedObjectStore versionedObjectStore;
    /**
     * 每个活跃循环只保留最近读取的一块，避免同一 128 项分块重复反序列化 128 次。
     */
    private final BoundedWeightedCache<String, CachedInputChunk>
            inputChunkCache =
            new BoundedWeightedCache<>(
                    MAX_CACHED_INPUT_RESULTS,
                    MAX_INPUT_CACHE_WEIGHT_BYTES,
                    MAX_CACHE_ENTRY_WEIGHT_BYTES);
    /**
     * 每个循环只缓存最近成功提交的活动输出分块，跨分块时自动替换历史块。
     */
    private final BoundedWeightedCache<String, CachedOutputChunk>
            outputChunkCache =
            new BoundedWeightedCache<>(
                    MAX_CACHED_OUTPUT_RESULTS,
                    MAX_OUTPUT_CACHE_WEIGHT_BYTES,
                    MAX_CACHE_ENTRY_WEIGHT_BYTES);

    /**
     * {@inheritDoc}
     */
    @Override
    public int storeInput(String resultId, Iterable<?> items) {
        return storeInputInternal(
                null, 0L, null, 0L, resultId, items);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int storeInput(
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration,
            String resultId,
            Iterable<?> items,
            long maxItems) {
        return storeProducedInputInternal(
                instanceId,
                lockFencingToken,
                claimId,
                claimGeneration,
                resultId,
                sink -> items.forEach(sink),
                maxItems);
    }

    /**
     * 流式分块保存循环输入。
     *
     * @param instanceId 工作流实例 ID
     * @param lockFencingToken 当前实例锁 token
     * @param claimId 当前触发器 ID
     * @param claimGeneration 当前认领代际
     * @param resultId 循环结果 ID
     * @param items 已应用预算的输入
     * @return 输入元素数量
     */
    private int storeInputInternal(
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration,
            String resultId,
            Iterable<?> items) {
        return storeProducedInputInternal(
                instanceId,
                lockFencingToken,
                claimId,
                claimGeneration,
                resultId,
                sink -> items.forEach(sink),
                0L);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int storeProducedInput(
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration,
            String resultId,
            InputProducer producer,
            long maxItems) {
        return storeProducedInputInternal(
                instanceId,
                lockFencingToken,
                claimId,
                claimGeneration,
                resultId,
                producer,
                maxItems);
    }

    /**
     * 接收生产者推送并直接写入固定大小分块。
     *
     * @param instanceId 工作流实例 ID
     * @param lockFencingToken 当前实例锁 token
     * @param claimId 当前触发器 ID
     * @param claimGeneration 当前触发器认领代际
     * @param resultId 循环结果 ID
     * @param producer 输入生产者
     * @param maxItems 最大元素数
     * @return 已写入元素数量
     */
    private int storeProducedInputInternal(
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration,
            String resultId,
            InputProducer producer,
            long maxItems) {
        InputChunkWriter writer = new InputChunkWriter(
                instanceId,
                lockFencingToken,
                claimId,
                claimGeneration,
                resultId,
                maxItems);
        try {
            producer.produce(writer::append);
            return writer.finish();
        } catch (RuntimeException | Error error) {
            if (writer.versionedInput) {
                try {
                    // resultId 每次物化均唯一，失锁后也可安全清理本 owner 已写的孤立分块。
                    versionedObjectStore.deleteAll(
                            writer.writtenKeys);
                } catch (RuntimeException cleanupError) {
                    error.addSuppressed(cleanupError);
                }
            } else {
                for (String writtenKey
                        : writer.writtenKeys) {
                    removeCache(writtenKey);
                }
            }
            throw error;
        }
    }

    /**
     * 单次推送式输入的分块写入状态。
     */
    private final class InputChunkWriter {

        private final String instanceId;
        private final long lockFencingToken;
        private final String claimId;
        private final long claimGeneration;
        private final String resultId;
        private final long maxItems;
        private final boolean versionedInput;
        private final List<Object> chunk =
                new ArrayList<>(CHUNK_SIZE);
        private final List<String> writtenKeys =
                new ArrayList<>();
        private int count;
        private int chunkIndex;

        /**
         * 创建一次分块写入。
         *
         * @param instanceId 工作流实例 ID
         * @param lockFencingToken 当前实例锁 token
         * @param claimId 当前触发器 ID
         * @param claimGeneration 当前触发器认领代际
         * @param resultId 循环结果 ID
         * @param maxItems 最大元素数
         */
        private InputChunkWriter(
                String instanceId,
                long lockFencingToken,
                String claimId,
                long claimGeneration,
                String resultId,
                long maxItems) {
            this.instanceId = instanceId;
            this.lockFencingToken = lockFencingToken;
            this.claimId = claimId;
            this.claimGeneration = claimGeneration;
            this.resultId = resultId;
            this.maxItems = maxItems;
            this.versionedInput =
                    versionedObjectStore != null
                            && isVersionedResult(resultId);
        }

        /**
         * 接收一个元素，达到固定分块大小时立即写入。
         *
         * @param item 输入元素
         */
        private void append(Object item) {
            if (maxItems > 0L && count >= maxItems) {
                throw new ExecutionBudgetExceededException(
                        "Loop iteration budget exceeded while storing input "
                                + resultId
                                + ": more than "
                                + maxItems);
            }
            chunk.add(item);
            count++;
            if (chunk.size() == CHUNK_SIZE) {
                flushChunk();
            }
        }

        /**
         * 写入尾块和输入元数据。
         *
         * @return 输入元素数量
         */
        private int finish() {
            if (!chunk.isEmpty()) {
                flushChunk();
            }
            String sizeKey = inputSizeKey(resultId);
            write(sizeKey, count);
            if (versionedInput
                    && lockFencingToken > 0L) {
                write(
                        refreshMarkerKey(resultId),
                        new RefreshMarker(
                                0L,
                                System.currentTimeMillis()));
            }
            return count;
        }

        /**
         * 写入当前数据分块。
         */
        private void flushChunk() {
            write(
                    inputChunkKey(
                            resultId, chunkIndex++),
                    new ArrayList<>(chunk));
            chunk.clear();
        }

        /**
         * 写入一个受 claim/lock 保护的值。
         *
         * @param key 存储键
         * @param value 存储值
         */
        private void write(
                String key,
                Serializable value) {
            storeInputValue(
                    key,
                    value,
                    versionedInput,
                    instanceId,
                    lockFencingToken,
                    claimId,
                    claimGeneration);
            writtenKeys.add(key);
        }
    }

    /**
     * 保存一个循环输入分块或元数据。
     *
     * @param key 存储键
     * @param value 输入值
     * @param versioned 是否使用版本对象存储
     * @param instanceId 工作流实例 ID
     * @param lockFencingToken 当前实例锁 token
     * @param claimId 当前触发器 ID
     * @param claimGeneration 当前认领代际
     */
    private void storeInputValue(
            String key,
            Serializable value,
            boolean versioned,
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration) {
        if (versioned) {
            boolean created;
            if (lockFencingToken > 0L && instanceId != null) {
                String claimGuardKey = claimGeneration > 0L
                        ? executionGuardKey(instanceId, claimId)
                        : lockFenceKey(instanceId);
                long effectiveClaimGeneration = claimGeneration > 0L
                        ? claimGeneration
                        : lockFencingToken;
                created = versionedObjectStore.createIfAbsent(
                        key,
                        value,
                        0L,
                        lockFenceKey(instanceId),
                        lockFencingToken,
                        claimGuardKey,
                        effectiveClaimGeneration,
                        RESULT_TTL);
            } else if (claimGeneration > 0L
                    && instanceId != null
                    && claimId != null) {
                created = versionedObjectStore.createIfAbsent(
                        key,
                        value,
                        0L,
                        executionGuardKey(instanceId, claimId),
                        claimGeneration,
                        RESULT_TTL);
            } else {
                created = versionedObjectStore.createIfAbsent(
                        key, value, 0L, RESULT_TTL);
            }
            if (!created) {
                throw new TriggerClaimLostException("loop-input:" + key);
            }
            return;
        }
        putCache(key, value);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    @SuppressWarnings("unchecked")
    public Object loadInputItem(String resultId, int index) {
        if (index < 0) {
            throw new IllegalArgumentException("index must not be negative");
        }
        int chunkIndex = index / CHUNK_SIZE;
        boolean versionedInput =
                versionedObjectStore != null
                        && isVersionedResult(resultId);
        CachedInputChunk cached = versionedInput
                ? inputChunkCache.get(resultId)
                : null;
        List<Object> chunk;
        if (cached != null && cached.chunkIndex == chunkIndex) {
            chunk = cached.values;
        } else {
            String key = inputChunkKey(resultId, chunkIndex);
            chunk = versionedInput
                    ? versionedObjectStore.load(key, List.class)
                    : getCache(key, List.class);
            if (versionedInput && chunk != null) {
                inputChunkCache.put(
                        resultId,
                        new CachedInputChunk(
                                chunkIndex,
                                chunk,
                                serializedWeight(chunk)));
            }
        }
        if (chunk == null || index % CHUNK_SIZE >= chunk.size()) {
            throw new IllegalStateException("Loop input item not found: " + resultId + ":" + index);
        }
        Object value = chunk.get(index % CHUNK_SIZE);
        return versionedInput
                ? snapshotValue(value).value
                : value;
    }

    /**
     * {@inheritDoc}
     *
     * <p>版本化输入按 128 项分块批量读取并直接展平。每个分块由对象存储独立
     * 反序列化，不进入共享活动缓存，因此既保持调用方可变值隔离，也避免逐元素
     * 创建 ObjectStream。</p>
     */
    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    public List<Object> loadInput(
            LoopInputReference reference) {
        if (reference == null) {
            throw new IllegalArgumentException(
                    "loop input reference required");
        }
        int itemCount =
                reference.getItemCount();
        if (itemCount < 0) {
            throw new IllegalArgumentException(
                    "itemCount must not be negative");
        }
        if (itemCount == 0) {
            return new ArrayList<>();
        }
        String resultId =
                reference.getResultId();
        if (versionedObjectStore == null
                || !isVersionedResult(resultId)) {
            return LoopResultRepository.super
                    .loadInput(reference);
        }

        int chunkCount =
                (itemCount + CHUNK_SIZE - 1)
                        / CHUNK_SIZE;
        List<String> keys =
                new ArrayList<>(chunkCount);
        for (int chunkIndex = 0;
             chunkIndex < chunkCount;
             chunkIndex++) {
            keys.add(inputChunkKey(
                    resultId, chunkIndex));
        }
        List<List<Object>> chunks =
                (List) versionedObjectStore.loadAll(
                        keys, List.class);
        if (chunks == null
                || chunks.size() != chunkCount) {
            throw new IllegalStateException(
                    "Incomplete loop input chunks: "
                            + resultId);
        }

        List<Object> items =
                new ArrayList<>(itemCount);
        for (int chunkIndex = 0;
             chunkIndex < chunkCount;
             chunkIndex++) {
            List<Object> chunk =
                    chunks.get(chunkIndex);
            if (chunk == null) {
                throw new IllegalStateException(
                        "Loop input chunk not found: "
                                + inputChunkKey(
                                resultId,
                                chunkIndex));
            }
            int remaining =
                    itemCount - items.size();
            int expectedChunkSize =
                    Math.min(CHUNK_SIZE, remaining);
            if (chunk.size()
                    != expectedChunkSize) {
                throw new IllegalStateException(
                        "Incomplete loop input chunk: "
                                + inputChunkKey(
                                resultId,
                                chunkIndex));
            }
            items.addAll(chunk);
        }
        return items;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void removeInput(String resultId) {
        removeInput(null, 0L, null, 0L, resultId);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void removeInput(
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration,
            String resultId) {
        inputChunkCache.remove(resultId);
        outputChunkCache.remove(resultId);
        boolean versionedInput =
                versionedObjectStore != null && isVersionedResult(resultId);
        String sizeKey = inputSizeKey(resultId);
        Integer count = versionedInput
                ? versionedObjectStore.load(sizeKey, Integer.class)
                : getCache(sizeKey, Integer.class);
        if (count == null) {
            return;
        }
        int chunkCount = (count + CHUNK_SIZE - 1) / CHUNK_SIZE;
        List<String> keys = new ArrayList<>(chunkCount + 1);
        for (int chunkIndex = 0; chunkIndex < chunkCount; chunkIndex++) {
            keys.add(inputChunkKey(resultId, chunkIndex));
        }
        keys.add(sizeKey);
        if (versionedInput) {
            keys.add(refreshMarkerKey(resultId));
        }
        if (versionedInput) {
            guardedDelete(
                    keys,
                    instanceId,
                    lockFencingToken,
                    claimId,
                    claimGeneration);
            return;
        }
        for (String key : keys) {
            removeCache(key);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void releaseActiveCache(String resultId) {
        inputChunkCache.remove(resultId);
        outputChunkCache.remove(resultId);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void append(String resultId, int iterationIndex, Map<String, Object> outputValues) {
        append(null, 0L, null, 0L, resultId, iterationIndex, outputValues);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void append(
            String instanceId,
            long fencingToken,
            String resultId,
            int iterationIndex,
            Map<String, Object> outputValues) {
        append(instanceId, fencingToken, null, 0L, resultId, iterationIndex, outputValues);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void append(
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration,
            String resultId,
            int iterationIndex,
            Map<String, Object> outputValues) {
        if (iterationIndex < 0) {
            throw new IllegalArgumentException("iterationIndex must not be negative");
        }

        int chunkIndex = iterationIndex / CHUNK_SIZE;
        int chunkOffset = iterationIndex % CHUNK_SIZE;
        boolean guardedVersionedWrite =
                versionedObjectStore != null && lockFencingToken > 0L && instanceId != null;
        if (guardedVersionedWrite && chunkOffset == 0 && chunkIndex > 0) {
            // 续期与是否声明输出无关，纯副作用循环同样需要保留早期输入分块。
            refreshActiveChunkExpirations(
                    instanceId,
                    lockFencingToken,
                    claimId,
                    claimGeneration,
                    resultId,
                    chunkIndex);
        }
        if (outputValues == null || outputValues.isEmpty()) {
            return;
        }

        String key = chunkKey(resultId, chunkIndex);
        if (guardedVersionedWrite) {
            if (claimGeneration > 0L && (claimId == null || claimId.trim().isEmpty())) {
                throw new IllegalArgumentException(
                        "claimId is required with claimGeneration");
            }
            appendVersioned(
                    instanceId,
                    lockFencingToken,
                    claimId,
                    claimGeneration,
                    resultId,
                    chunkIndex,
                    key,
                    chunkOffset,
                    outputValues);
            return;
        }
        LoopResultChunk chunk = getCache(key, LoopResultChunk.class);
        if (chunk == null) {
            chunk = new LoopResultChunk();
        }

        for (Map.Entry<String, Object> entry : outputValues.entrySet()) {
            List<Object> values = chunk.values.computeIfAbsent(entry.getKey(), ignored -> new ArrayList<>());
            if (values.size() < chunkOffset) {
                throw new IllegalStateException("Loop result chunk is incomplete: " + key);
            }
            if (values.size() == chunkOffset) {
                values.add(entry.getValue());
                continue;
            }
            Object existing = values.get(chunkOffset);
            if (!java.util.Objects.deepEquals(existing, entry.getValue())) {
                throw new IllegalStateException("Conflicting loop result replay: " + key);
            }
        }
        putCache(key, chunk);
    }

    /**
     * 使用版本比较和 fencing token 原子追加循环分块。
     *
     * @param instanceId 工作流实例 ID
     * @param lockFencingToken 当前实例锁 token
     * @param claimId 当前触发器 ID
     * @param claimGeneration 当前触发器认领代际
     * @param resultId 循环结果 ID
     * @param chunkIndex 当前分块序号
     * @param key 分块键
     * @param chunkOffset 分块内偏移
     * @param outputValues 本轮输出
     */
    private void appendVersioned(
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration,
            String resultId,
            int chunkIndex,
            String key,
            int chunkOffset,
            Map<String, Object> outputValues) {
        String claimGuardKey = claimGeneration > 0L
                ? executionGuardKey(instanceId, claimId)
                : lockFenceKey(instanceId);
        long effectiveClaimGeneration = claimGeneration > 0L
                ? claimGeneration
                : lockFencingToken;
        for (int attempt = 0; attempt < 16; attempt++) {
            CachedOutputChunk cached =
                    outputChunkCache.get(resultId);
            if (cached != null
                    && cached.chunkIndex
                    != chunkIndex) {
                outputChunkCache.remove(resultId);
                cached = null;
            }
            LoopResultChunk chunk = cached == null
                    ? versionedObjectStore.load(key, LoopResultChunk.class)
                    : cached.chunk.copy();
            long chunkWeight = cached == null
                    ? serializedWeightOrDefault(
                            chunk,
                            EMPTY_CHUNK_WEIGHT_BYTES)
                    : cached.weight();
            if (chunk == null) {
                LoopResultChunk created = new LoopResultChunk();
                OutputMutation mutation = applyOutput(
                        created,
                        key,
                        chunkOffset,
                        outputValues);
                if (versionedObjectStore.createIfAbsent(
                        key,
                        created,
                        created.version,
                        lockFenceKey(instanceId),
                        lockFencingToken,
                        claimGuardKey,
                        effectiveClaimGeneration,
                        RESULT_TTL)) {
                    outputChunkCache.put(
                            resultId,
                            new CachedOutputChunk(
                                    chunkIndex,
                                    created.copy(),
                                    saturatingAdd(
                                            EMPTY_CHUNK_WEIGHT_BYTES,
                                            mutation.addedWeight)));
                    return;
                }
                outputChunkCache.remove(resultId);
                continue;
            }
            long expectedVersion = chunk.version;
            OutputMutation mutation = applyOutput(
                    chunk,
                    key,
                    chunkOffset,
                    outputValues);
            if (!mutation.changed) {
                if (cached == null) {
                    outputChunkCache.put(
                            resultId,
                            new CachedOutputChunk(
                                    chunkIndex,
                                    chunk.copy(),
                                    chunkWeight));
                }
                return;
            }
            chunk.version = expectedVersion + 1L;
            if (versionedObjectStore.compareAndSet(
                    key,
                    expectedVersion,
                    chunk,
                    chunk.version,
                    lockFenceKey(instanceId),
                    lockFencingToken,
                    claimGuardKey,
                    effectiveClaimGeneration,
                    RESULT_TTL)) {
                outputChunkCache.put(
                        resultId,
                        new CachedOutputChunk(
                                chunkIndex,
                                chunk.copy(),
                                saturatingAdd(
                                        chunkWeight,
                                        mutation.addedWeight)));
                return;
            }
            outputChunkCache.remove(resultId);
        }
        throw new TriggerClaimLostException("loop-result:" + key);
    }

    /**
     * 在进入新结果分块时批量续期既有结果和输入分块。
     *
     * <p>每 128 轮最多执行一次 Redis pipeline，保证长循环早期分块不会先于活跃实例
     * 过期，同时避免每轮逐块续期。</p>
     *
     * @param resultId 循环结果 ID
     * @param currentChunkIndex 当前新分块序号
     */
    private void refreshActiveChunkExpirations(
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration,
            String resultId,
            int currentChunkIndex) {
        String markerKey = refreshMarkerKey(resultId);
        RefreshMarker marker =
                versionedObjectStore.load(markerKey, RefreshMarker.class);
        long now = System.currentTimeMillis();
        if (marker != null
                && now - marker.refreshedAt < REFRESH_INTERVAL_MILLIS) {
            return;
        }
        String claimGuardKey = claimGeneration > 0L
                ? executionGuardKey(instanceId, claimId)
                : lockFenceKey(instanceId);
        long effectiveClaimGeneration = claimGeneration > 0L
                ? claimGeneration
                : lockFencingToken;
        boolean markerUpdated;
        if (marker == null) {
            markerUpdated = versionedObjectStore.createIfAbsent(
                    markerKey,
                    new RefreshMarker(0L, now),
                    0L,
                    lockFenceKey(instanceId),
                    lockFencingToken,
                    claimGuardKey,
                    effectiveClaimGeneration,
                    RESULT_TTL);
        } else {
            long expectedVersion = marker.version;
            marker.version = expectedVersion + 1L;
            marker.refreshedAt = now;
            markerUpdated = versionedObjectStore.compareAndSet(
                    markerKey,
                    expectedVersion,
                    marker,
                    marker.version,
                    lockFenceKey(instanceId),
                    lockFencingToken,
                    claimGuardKey,
                    effectiveClaimGeneration,
                    RESULT_TTL);
        }
        if (!markerUpdated) {
            throw new TriggerClaimLostException(
                    "loop-refresh:" + resultId);
        }
        List<String> keys = new ArrayList<>();
        for (int chunkIndex = 0; chunkIndex < currentChunkIndex; chunkIndex++) {
            keys.add(chunkKey(resultId, chunkIndex));
        }
        Integer inputSize = versionedObjectStore.load(
                inputSizeKey(resultId), Integer.class);
        if (inputSize != null) {
            int inputChunkCount = (inputSize + CHUNK_SIZE - 1) / CHUNK_SIZE;
            for (int inputChunkIndex = 0;
                 inputChunkIndex < inputChunkCount;
                 inputChunkIndex++) {
                keys.add(inputChunkKey(resultId, inputChunkIndex));
            }
            keys.add(inputSizeKey(resultId));
        }
        keys.add(markerKey);
        versionedObjectStore.refreshExpirations(keys, RESULT_TTL);
    }

    /**
     * 将一轮输出合并到分块并校验幂等重放。
     *
     * @param chunk 当前分块
     * @param key 分块键
     * @param chunkOffset 分块内偏移
     * @param outputValues 本轮输出
     * @return 本次变更及新增缓存重量
     */
    private OutputMutation applyOutput(
            LoopResultChunk chunk,
            String key,
            int chunkOffset,
            Map<String, Object> outputValues) {
        boolean changed = false;
        long addedWeight = 0L;
        for (Map.Entry<String, Object> entry : outputValues.entrySet()) {
            boolean newOutput =
                    !chunk.values.containsKey(
                            entry.getKey());
            List<Object> values =
                    chunk.values.computeIfAbsent(entry.getKey(), ignored -> new ArrayList<>());
            if (values.size() < chunkOffset) {
                throw new IllegalStateException("Loop result chunk is incomplete: " + key);
            }
            if (values.size() == chunkOffset) {
                SnapshotValue snapshot =
                        snapshotValue(entry.getValue());
                values.add(snapshot.value);
                changed = true;
                addedWeight = saturatingAdd(
                        addedWeight,
                        snapshot.weight
                                + (newOutput
                                ? estimateStringWeight(
                                        entry.getKey())
                                : 0L)
                                + 32L);
                continue;
            }
            Object existing = values.get(chunkOffset);
            if (!java.util.Objects.deepEquals(existing, entry.getValue())) {
                throw new IllegalStateException("Conflicting loop result replay: " + key);
            }
        }
        return new OutputMutation(
                changed, addedWeight);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Map<String, Object> load(String resultId, int iterationCount, List<String> outputNames) {
        Map<String, Object> result = new LinkedHashMap<>();
        if (iterationCount == 0 || outputNames == null || outputNames.isEmpty()) {
            return result;
        }
        if (iterationCount < 0) {
            throw new IllegalArgumentException("iterationCount must not be negative");
        }

        Map<String, List<Object>> collected = new LinkedHashMap<>();
        for (String outputName : outputNames) {
            collected.put(outputName, new ArrayList<>(iterationCount));
        }

        int chunkCount = (iterationCount + CHUNK_SIZE - 1) / CHUNK_SIZE;
        List<LoopResultChunk> versionedChunks = null;
        if (versionedObjectStore != null && isVersionedResult(resultId)) {
            List<String> keys = new ArrayList<>(chunkCount);
            for (int chunkIndex = 0; chunkIndex < chunkCount; chunkIndex++) {
                keys.add(chunkKey(resultId, chunkIndex));
            }
            versionedChunks = versionedObjectStore.loadAll(
                    keys, LoopResultChunk.class);
        }
        for (int chunkIndex = 0; chunkIndex < chunkCount; chunkIndex++) {
            String key = chunkKey(resultId, chunkIndex);
            LoopResultChunk chunk = versionedChunks != null
                    ? versionedChunks.get(chunkIndex)
                    : getCache(key, LoopResultChunk.class);
            if (chunk == null) {
                throw new IllegalStateException("Loop result chunk not found: " + key);
            }
            for (String outputName : outputNames) {
                List<Object> values = chunk.values.get(outputName);
                if (values == null) {
                    throw new IllegalStateException("Loop output not found: " + outputName);
                }
                collected.get(outputName).addAll(values);
            }
        }

        for (Map.Entry<String, List<Object>> entry : collected.entrySet()) {
            if (entry.getValue().size() != iterationCount) {
                throw new IllegalStateException("Incomplete loop result: " + entry.getKey());
            }
            result.put(entry.getKey(), entry.getValue());
        }
        return result;
    }

    /**
     * 构建结果分块缓存键。
     *
     * @param resultId 循环结果 ID
     * @param chunkIndex 分块序号
     * @return 缓存键
     */
    private String chunkKey(String resultId, int chunkIndex) {
        if (isVersionedResult(resultId)) {
            String instanceId = resultId.substring(0, resultId.indexOf(':'));
            String localResultId = resultId.substring(resultId.indexOf(':') + 1);
            return CacheKey.CHAIN_STATE_CAS_KEY + "{" + instanceId + "}:loop:"
                    + localResultId + ":" + chunkIndex;
        }
        return CacheKey.LOOP_RESULT_CACHE_KEY + resultId + ":" + chunkIndex;
    }

    /**
     * 判断结果 ID 是否为包含实例前缀的新格式。
     *
     * @param resultId 循环结果 ID
     * @return 新格式时为 {@code true}
     */
    private boolean isVersionedResult(String resultId) {
        return resultId != null && resultId.indexOf(':') > 0;
    }

    /**
     * 构建实例 fencing token 键。
     *
     * @param instanceId 工作流实例 ID
     * @return fencing token 键
     */
    private String executionGuardKey(String instanceId, String claimId) {
        return CacheKey.CHAIN_STATE_CAS_KEY + "{" + instanceId + "}:claim:" + claimId;
    }

    /**
     * 构建实例锁 fencing token 键。
     *
     * @param instanceId 工作流实例 ID
     * @return fencing token 键
     */
    private String lockFenceKey(String instanceId) {
        return CacheKey.CHAIN_STATE_CAS_KEY + "{" + instanceId + "}:fence";
    }

    /**
     * 构建输入分块缓存键。
     *
     * @param resultId 循环结果 ID
     * @param chunkIndex 分块序号
     * @return 缓存键
     */
    private String inputChunkKey(String resultId, int chunkIndex) {
        if (isVersionedResult(resultId)) {
            String instanceId = resultId.substring(0, resultId.indexOf(':'));
            String localResultId = resultId.substring(resultId.indexOf(':') + 1);
            return CacheKey.CHAIN_STATE_CAS_KEY + "{" + instanceId + "}:loop:"
                    + localResultId + ":input:" + chunkIndex;
        }
        return CacheKey.LOOP_RESULT_CACHE_KEY + resultId + ":input:" + chunkIndex;
    }

    /**
     * 构建输入数量缓存键。
     *
     * @param resultId 循环结果 ID
     * @return 缓存键
     */
    private String inputSizeKey(String resultId) {
        if (isVersionedResult(resultId)) {
            String instanceId = resultId.substring(0, resultId.indexOf(':'));
            String localResultId = resultId.substring(resultId.indexOf(':') + 1);
            return CacheKey.CHAIN_STATE_CAS_KEY + "{" + instanceId + "}:loop:"
                    + localResultId + ":input:size";
        }
        return CacheKey.LOOP_RESULT_CACHE_KEY + resultId + ":input:size";
    }

    /**
     * 构建循环分块生命周期刷新标记键。
     *
     * @param resultId 循环结果 ID
     * @return 刷新标记键
     */
    private String refreshMarkerKey(String resultId) {
        if (isVersionedResult(resultId)) {
            String instanceId = resultId.substring(0, resultId.indexOf(':'));
            String localResultId =
                    resultId.substring(resultId.indexOf(':') + 1);
            return CacheKey.CHAIN_STATE_CAS_KEY + "{" + instanceId + "}:loop:"
                    + localResultId + ":refresh";
        }
        return CacheKey.LOOP_RESULT_CACHE_KEY + resultId + ":refresh";
    }

    /**
     * 仅在当前双守卫仍有效时删除循环输入。
     *
     * @param keys 待删除键
     * @param instanceId 工作流实例 ID
     * @param lockFencingToken 当前实例锁 token
     * @param claimId 当前触发器 ID
     * @param claimGeneration 当前认领代际
     */
    private void guardedDelete(
            List<String> keys,
            String instanceId,
            long lockFencingToken,
            String claimId,
            long claimGeneration) {
        if (keys.isEmpty()) {
            return;
        }
        if (lockFencingToken <= 0L || instanceId == null) {
            versionedObjectStore.deleteAll(keys);
            return;
        }
        String claimGuardKey = claimGeneration > 0L
                ? executionGuardKey(instanceId, claimId)
                : lockFenceKey(instanceId);
        long effectiveClaimGeneration = claimGeneration > 0L
                ? claimGeneration
                : lockFencingToken;
        versionedObjectStore.deleteAll(
                keys,
                lockFenceKey(instanceId),
                lockFencingToken,
                claimGuardKey,
                effectiveClaimGeneration);
    }

    /**
     * 创建与调用方可变对象图隔离的缓存快照。
     *
     * <p>持久化路径本身使用相同 Java 序列化协议；这里只对可能可变的单个业务值复制，
     * 避免每轮重新复制整个 128 项分块。</p>
     *
     * @param value 业务值
     * @return 隔离值及其序列化重量
     */
    private static SnapshotValue snapshotValue(
            Object value) {
        if (isKnownImmutable(value)) {
            return new SnapshotValue(
                    value,
                    estimateImmutableWeight(value));
        }
        byte[] encoded = JavaValueEncoder.INSTANCE.apply(
                value);
        return new SnapshotValue(
                CACHE_VALUE_DECODER.apply(encoded),
                encoded.length);
    }

    /**
     * 计算完整已持久化对象的序列化重量。
     *
     * @param value 已持久化对象
     * @return 字节重量；不可序列化时返回最大值以拒绝缓存
     */
    private static long serializedWeight(
            Object value) {
        if (!(value instanceof Serializable)) {
            return Long.MAX_VALUE;
        }
        return JavaValueEncoder.INSTANCE.apply(
                value).length;
    }

    /**
     * 对可空对象计算序列化重量。
     *
     * @param value 对象
     * @param defaultWeight 空对象默认重量
     * @return 缓存重量
     */
    private static long serializedWeightOrDefault(
            Object value,
            long defaultWeight) {
        return value == null
                ? defaultWeight
                : serializedWeight(value);
    }

    /**
     * 判断无需复制即可安全共享的基础不可变值。
     *
     * @param value 值
     * @return 已知不可变时为 {@code true}
     */
    private static boolean isKnownImmutable(
            Object value) {
        return value == null
                || value instanceof String
                || value instanceof Boolean
                || value instanceof Character
                || value instanceof Byte
                || value instanceof Short
                || value instanceof Integer
                || value instanceof Long
                || value instanceof Float
                || value instanceof Double
                || value instanceof java.math.BigInteger
                || value instanceof java.math.BigDecimal
                || value instanceof Enum<?>
                || value instanceof java.util.UUID;
    }

    /**
     * 估算基础不可变值的堆重量。
     *
     * @param value 不可变值
     * @return 保守字节估算
     */
    private static long estimateImmutableWeight(
            Object value) {
        if (value == null) {
            return 8L;
        }
        if (value instanceof String) {
            return estimateStringWeight(
                    (String) value);
        }
        return 32L;
    }

    /**
     * 估算字符串及其字符存储重量。
     *
     * @param value 字符串
     * @return 保守字节估算
     */
    private static long estimateStringWeight(
            String value) {
        return value == null
                ? 8L
                : 40L + (long) value.length() * 2L;
    }

    /**
     * 饱和相加两个非负重量。
     *
     * @param left 左值
     * @param right 右值
     * @return 相加结果；溢出时为 {@link Long#MAX_VALUE}
     */
    private static long saturatingAdd(
            long left,
            long right) {
        if (left >= Long.MAX_VALUE - right) {
            return Long.MAX_VALUE;
        }
        return left + right;
    }

    /**
     * 可计量缓存值。
     */
    private interface WeightedCacheValue {

        /**
         * 获取缓存重量。
         *
         * @return 非负字节重量
         */
        long weight();
    }

    /**
     * 同时限制条目数、单条重量和总重量的 LRU 缓存。
     *
     * @param <K> 键类型
     * @param <V> 可计量值类型
     */
    private static final class BoundedWeightedCache<
            K, V extends WeightedCacheValue> {

        private final int maxEntries;
        private final long maxTotalWeight;
        private final long maxEntryWeight;
        private final LinkedHashMap<K, V> values =
                new LinkedHashMap<>(16, 0.75F, true);
        private long totalWeight;

        /**
         * 创建有界缓存。
         *
         * @param maxEntries 最大条目数
         * @param maxTotalWeight 最大总重量
         * @param maxEntryWeight 最大单条重量
         */
        private BoundedWeightedCache(
                int maxEntries,
                long maxTotalWeight,
                long maxEntryWeight) {
            this.maxEntries = maxEntries;
            this.maxTotalWeight = maxTotalWeight;
            this.maxEntryWeight = maxEntryWeight;
        }

        /**
         * 获取并提升条目热度。
         *
         * @param key 键
         * @return 缓存值；未命中时为 {@code null}
         */
        private synchronized V get(K key) {
            return values.get(key);
        }

        /**
         * 写入缓存；超大条目直接拒绝，并按 LRU 淘汰到双上限以内。
         *
         * @param key 键
         * @param value 值
         */
        private synchronized void put(
                K key,
                V value) {
            V previous = values.remove(key);
            if (previous != null) {
                totalWeight -= previous.weight();
            }
            long valueWeight =
                    Math.max(0L, value.weight());
            if (valueWeight > maxEntryWeight
                    || valueWeight
                    > maxTotalWeight) {
                return;
            }
            while (!values.isEmpty()
                    && (values.size() >= maxEntries
                    || totalWeight
                    > maxTotalWeight
                    - valueWeight)) {
                Iterator<Map.Entry<K, V>> iterator =
                        values.entrySet().iterator();
                Map.Entry<K, V> eldest =
                        iterator.next();
                totalWeight -= eldest.getValue().weight();
                iterator.remove();
            }
            values.put(key, value);
            totalWeight += valueWeight;
        }

        /**
         * 删除缓存条目。
         *
         * @param key 键
         */
        private synchronized void remove(K key) {
            V removed = values.remove(key);
            if (removed != null) {
                totalWeight -= removed.weight();
            }
        }
    }

    /**
     * 单个隔离业务值及其缓存重量。
     */
    private static final class SnapshotValue {

        private final Object value;
        private final long weight;

        /**
         * 创建业务值快照。
         *
         * @param value 隔离值
         * @param weight 字节重量
         */
        private SnapshotValue(
                Object value,
                long weight) {
            this.value = value;
            this.weight = weight;
        }
    }

    /**
     * 一次输出合并结果。
     */
    private static final class OutputMutation {

        private final boolean changed;
        private final long addedWeight;

        /**
         * 创建输出变更结果。
         *
         * @param changed 分块是否变化
         * @param addedWeight 新增重量
         */
        private OutputMutation(
                boolean changed,
                long addedWeight) {
            this.changed = changed;
            this.addedWeight = addedWeight;
        }
    }

    /**
     * 一个循环最近读取的输入分块。
     */
    private static final class CachedInputChunk
            implements WeightedCacheValue {

        private final int chunkIndex;
        private final List<Object> values;
        private final long weight;

        /**
         * 创建输入分块缓存。
         *
         * @param chunkIndex 分块序号
         * @param values 分块值
         * @param weight 缓存重量
         */
        private CachedInputChunk(
                int chunkIndex,
                List<Object> values,
                long weight) {
            this.chunkIndex = chunkIndex;
            this.values = values;
            this.weight = weight;
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public long weight() {
            return weight;
        }
    }

    /**
     * 一个循环最近成功提交的活动输出分块。
     */
    private static final class CachedOutputChunk
            implements WeightedCacheValue {

        private final int chunkIndex;
        private final LoopResultChunk chunk;
        private final long weight;

        /**
         * 创建活动输出缓存。
         *
         * @param chunkIndex 分块序号
         * @param chunk 私有分块快照
         * @param weight 缓存重量
         */
        private CachedOutputChunk(
                int chunkIndex,
                LoopResultChunk chunk,
                long weight) {
            this.chunkIndex = chunkIndex;
            this.chunk = chunk;
            this.weight = weight;
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public long weight() {
            return weight;
        }
    }

    /**
     * 循环分块批量续期节流标记。
     */
    static final class RefreshMarker implements Serializable {

        private long version;
        private long refreshedAt;

        /**
         * 创建刷新标记。
         *
         * @param version 对象版本
         * @param refreshedAt 最近刷新时间
         */
        RefreshMarker(long version, long refreshedAt) {
            this.version = version;
            this.refreshedAt = refreshedAt;
        }
    }

    /**
     * 单个循环结果分块。
     */
    static final class LoopResultChunk implements Serializable {

        private long version;
        private final Map<String, List<Object>> values = new LinkedHashMap<>();

        /**
         * 创建空结果分块。
         */
        LoopResultChunk() {
        }

        /**
         * 创建与源分块容器相互隔离的私有快照。
         *
         * <p>业务值在首次进入活动缓存时已经逐值序列化隔离，后续只复制 Map/List
         * 容器即可避免追加线程相互污染。</p>
         *
         * @param source 源分块
         */
        private LoopResultChunk(LoopResultChunk source) {
            this.version = source.version;
            source.values.forEach((name, items) ->
                    this.values.put(name, new ArrayList<>(items)));
        }

        /**
         * 复制当前分块，避免并发失锁执行者污染共享缓存。
         *
         * @return 独立分块快照
         */
        private LoopResultChunk copy() {
            return new LoopResultChunk(this);
        }

        /**
         * 获取当前分块输出。
         *
         * @return 输出名称到值列表的映射
         */
        Map<String, List<Object>> getValues() {
            return values;
        }
    }
}
