package tech.easyflow.common.cache;

import com.alicp.jetcache.support.JavaValueEncoder;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * 基于 Redis Hash 与 Lua 的原子版本对象存储。
 *
 * <p>对象继续使用项目既有的 Java 序列化协议，避免状态内多态值在迁移后改变类型。
 * payload 与 version 保存在同一 Redis Hash 中，创建、版本比较、写入和 TTL 刷新均由
 * 单次 Lua 脚本原子完成。</p>
 */
@Component
public class RedisVersionedObjectStore implements VersionedObjectStore {

    private static final byte[] PAYLOAD_FIELD = bytes("payload");
    private static final byte[] CREATE_SCRIPT = bytes(
            "if redis.call('exists', KEYS[1]) == 1 then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[1], 'payload', ARGV[2]); "
                    + "redis.call('pexpire', KEYS[1], ARGV[3]); return 1");
    private static final byte[] GUARDED_CREATE_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "if not guard or guard ~= ARGV[2] then return -1 end; "
                    + "if redis.call('exists', KEYS[1]) == 1 then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[1], 'payload', ARGV[3]); "
                    + "redis.call('pexpire', KEYS[1], ARGV[4]); return 1");
    private static final byte[] DOUBLE_GUARDED_CREATE_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "local secondary = redis.call('hget', KEYS[3], 'version'); "
                    + "if not guard or guard ~= ARGV[2] "
                    + "or not secondary or secondary ~= ARGV[3] then return -1 end; "
                    + "if redis.call('exists', KEYS[1]) == 1 then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[1], 'payload', ARGV[4]); "
                    + "redis.call('pexpire', KEYS[1], ARGV[5]); return 1");
    private static final byte[] CAS_SCRIPT = bytes(
            "local current = redis.call('hget', KEYS[1], 'version'); "
                    + "if not current then return -1 end; "
                    + "if current ~= ARGV[1] then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[2], 'payload', ARGV[3]); "
                    + "redis.call('pexpire', KEYS[1], ARGV[4]); return 1");
    private static final byte[] GUARDED_CAS_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "if not guard or guard ~= ARGV[3] then return -2 end; "
                    + "local current = redis.call('hget', KEYS[1], 'version'); "
                    + "if not current then return -1 end; "
                    + "if current ~= ARGV[1] then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[2], 'payload', ARGV[4]); "
                    + "redis.call('pexpire', KEYS[1], ARGV[5]); return 1");
    private static final byte[] DOUBLE_GUARDED_CAS_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "local secondary = redis.call('hget', KEYS[3], 'version'); "
                    + "if not guard or guard ~= ARGV[3] "
                    + "or not secondary or secondary ~= ARGV[4] then return -2 end; "
                    + "local current = redis.call('hget', KEYS[1], 'version'); "
                    + "if not current then return -1 end; "
                    + "if current ~= ARGV[1] then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[2], 'payload', ARGV[5]); "
                    + "redis.call('pexpire', KEYS[1], ARGV[6]); return 1");
    private static final byte[] DOUBLE_GUARDED_DELETE_ALL_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[#KEYS - 1], 'version'); "
                    + "local secondary = redis.call('hget', KEYS[#KEYS], 'version'); "
                    + "if not guard or guard ~= ARGV[1] "
                    + "or not secondary or secondary ~= ARGV[2] then return -1 end; "
                    + "for i = 1, #KEYS - 2 do redis.call('del', KEYS[i]); end; "
                    + "return 1");
    private static final byte[] CREATE_FIELDS_SCRIPT = bytes(
            "if redis.call('exists', KEYS[1]) == 1 then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[1]); "
                    + "for i = 2, #ARGV - 1, 2 do redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV]); return 1");
    private static final byte[] GUARDED_CREATE_FIELDS_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "if not guard or guard ~= ARGV[2] then return -1 end; "
                    + "if redis.call('exists', KEYS[1]) == 1 then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[1]); "
                    + "for i = 3, #ARGV - 1, 2 do redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV]); return 1");
    private static final byte[] DOUBLE_GUARDED_CREATE_FIELDS_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "local secondary = redis.call('hget', KEYS[3], 'version'); "
                    + "if not guard or guard ~= ARGV[2] "
                    + "or not secondary or secondary ~= ARGV[3] then return -1 end; "
                    + "if redis.call('exists', KEYS[1]) == 1 then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[1]); "
                    + "for i = 4, #ARGV - 1, 2 do "
                    + "redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV]); return 1");
    private static final byte[] TRIPLE_GUARDED_CREATE_FIELDS_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "local secondary = redis.call('hget', KEYS[3], 'version'); "
                    + "local tertiary = redis.call('hget', KEYS[4], 'version'); "
                    + "if not guard or guard ~= ARGV[2] "
                    + "or not secondary or secondary ~= ARGV[3] "
                    + "or not tertiary or tertiary ~= ARGV[4] then return -1 end; "
                    + "if redis.call('exists', KEYS[1]) == 1 then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[1]); "
                    + "for i = 5, #ARGV - 1, 2 do "
                    + "redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV]); return 1");
    private static final byte[] CAS_FIELDS_SCRIPT = bytes(
            "local current = redis.call('hget', KEYS[1], 'version'); "
                    + "if not current then return -1 end; "
                    + "if current ~= ARGV[1] then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[2]); "
                    + "for i = 3, #ARGV - 1, 2 do redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV]); return 1");
    private static final byte[] GUARDED_CAS_FIELDS_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "if not guard or guard ~= ARGV[3] then return -2 end; "
                    + "local current = redis.call('hget', KEYS[1], 'version'); "
                    + "if not current then return -1 end; "
                    + "if current ~= ARGV[1] then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[2]); "
                    + "for i = 4, #ARGV - 1, 2 do redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV]); return 1");
    private static final byte[] DOUBLE_GUARDED_CAS_FIELDS_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "local secondary = redis.call('hget', KEYS[3], 'version'); "
                    + "if not guard or guard ~= ARGV[3] "
                    + "or not secondary or secondary ~= ARGV[4] then return -2 end; "
                    + "local current = redis.call('hget', KEYS[1], 'version'); "
                    + "if not current then return -1 end; "
                    + "if current ~= ARGV[1] then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[2]); "
                    + "for i = 5, #ARGV - 1, 2 do "
                    + "redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV]); return 1");
    private static final byte[] TRIPLE_GUARDED_CAS_FIELDS_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "local secondary = redis.call('hget', KEYS[3], 'version'); "
                    + "local tertiary = redis.call('hget', KEYS[4], 'version'); "
                    + "if not guard or guard ~= ARGV[3] "
                    + "or not secondary or secondary ~= ARGV[4] "
                    + "or not tertiary or tertiary ~= ARGV[5] then return -2 end; "
                    + "local current = redis.call('hget', KEYS[1], 'version'); "
                    + "if not current then return -1 end; "
                    + "if current ~= ARGV[1] then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[2]); "
                    + "for i = 6, #ARGV - 1, 2 do "
                    + "redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV]); return 1");
    private static final byte[] DOUBLE_GUARDED_CAS_FIELDS_REFRESH_SCRIPT = bytes(
            "local guard = redis.call('hget', KEYS[2], 'version'); "
                    + "local secondary = redis.call('hget', KEYS[3], 'version'); "
                    + "if not guard or guard ~= ARGV[3] "
                    + "or not secondary or secondary ~= ARGV[4] then return -2 end; "
                    + "local current = redis.call('hget', KEYS[1], 'version'); "
                    + "if not current then return -1 end; "
                    + "if current ~= ARGV[1] then return 0 end; "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[2]); "
                    + "for i = 5, #ARGV - 2, 2 do "
                    + "redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV - 1]); "
                    + "if redis.call('exists', KEYS[4]) == 0 then "
                    + "redis.call('hset', KEYS[4], 'version', '0'); end; "
                    + "redis.call('pexpire', KEYS[4], ARGV[#ARGV]); return 1");
    private static final byte[] REWRITE_FIELDS_SCRIPT = bytes(
            "local current = redis.call('hget', KEYS[1], 'version'); "
                    + "if not current or current ~= ARGV[1] then return 0 end; "
                    + "redis.call('del', KEYS[1]); "
                    + "redis.call('hset', KEYS[1], 'version', ARGV[1]); "
                    + "for i = 2, #ARGV - 1, 2 do redis.call('hset', KEYS[1], ARGV[i], ARGV[i + 1]); end; "
                    + "redis.call('pexpire', KEYS[1], ARGV[#ARGV]); return 1");

    private final RedisConnectionFactory connectionFactory;
    private final ApplicationClassLoaderJavaValueDecoder valueDecoder =
            new ApplicationClassLoaderJavaValueDecoder();

    /**
     * 创建 Redis 版本对象存储。
     *
     * @param connectionFactory Redis 连接工厂
     */
    public RedisVersionedObjectStore(RedisConnectionFactory connectionFactory) {
        this.connectionFactory = Objects.requireNonNull(
                connectionFactory, "connectionFactory must not be null");
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> T load(String key, Class<T> type) {
        requireKey(key);
        Objects.requireNonNull(type, "type must not be null");
        try (RedisConnection connection = connectionFactory.getConnection()) {
            byte[] payload = connection.hashCommands().hGet(bytes(key), PAYLOAD_FIELD);
            if (payload == null) {
                return null;
            }
            return type.cast(valueDecoder.apply(payload));
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> List<T> loadAll(List<String> keys, Class<T> type) {
        Objects.requireNonNull(keys, "keys must not be null");
        Objects.requireNonNull(type, "type must not be null");
        if (keys.isEmpty()) {
            return List.of();
        }
        try (RedisConnection connection = connectionFactory.getConnection()) {
            connection.openPipeline();
            for (String key : keys) {
                connection.hashCommands().hGet(bytes(requireKey(key)), PAYLOAD_FIELD);
            }
            List<Object> encodedValues = connection.closePipeline();
            List<T> values = new ArrayList<>(keys.size());
            for (int index = 0; index < keys.size(); index++) {
                Object encoded = encodedValues != null && index < encodedValues.size()
                        ? encodedValues.get(index)
                        : null;
                values.add(encoded instanceof byte[]
                        ? type.cast(valueDecoder.apply((byte[]) encoded))
                        : null);
            }
            return values;
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteAll(List<String> keys) {
        Objects.requireNonNull(keys, "keys must not be null");
        if (keys.isEmpty()) {
            return;
        }
        byte[][] encodedKeys = new byte[keys.size()][];
        for (int index = 0; index < keys.size(); index++) {
            encodedKeys[index] = bytes(requireKey(keys.get(index)));
        }
        try (RedisConnection connection = connectionFactory.getConnection()) {
            connection.keyCommands().del(encodedKeys);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean deleteAll(
            List<String> keys,
            String guardKey,
            long guardVersion,
            String secondaryGuardKey,
            long secondaryGuardVersion) {
        Objects.requireNonNull(keys, "keys must not be null");
        if (keys.isEmpty()) {
            return true;
        }
        List<byte[]> arguments = new ArrayList<>(keys.size() + 4);
        for (String key : keys) {
            arguments.add(bytes(requireKey(key)));
        }
        arguments.add(bytes(requireKey(guardKey)));
        arguments.add(bytes(requireKey(secondaryGuardKey)));
        arguments.add(bytes(guardVersion));
        arguments.add(bytes(secondaryGuardVersion));
        Long result = eval(
                DOUBLE_GUARDED_DELETE_ALL_SCRIPT,
                keys.size() + 2,
                arguments.toArray(new byte[0][]));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void refreshExpirations(List<String> keys, Duration ttl) {
        Objects.requireNonNull(keys, "keys must not be null");
        if (keys.isEmpty()) {
            return;
        }
        long ttlMillis = ttlMillis(ttl);
        try (RedisConnection connection = connectionFactory.getConnection()) {
            connection.openPipeline();
            for (String key : keys) {
                connection.keyCommands().pExpire(
                        bytes(requireKey(key)), ttlMillis);
            }
            connection.closePipeline();
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean createIfAbsent(String key,
                                  Serializable value,
                                  long version,
                                  Duration ttl) {
        Long result = eval(
                CREATE_SCRIPT,
                1,
                bytes(requireKey(key)),
                bytes(version),
                encode(value),
                bytes(ttlMillis(ttl)));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean createIfAbsent(
            String key,
            Serializable value,
            long version,
            String guardKey,
            long guardVersion,
            String secondaryGuardKey,
            long secondaryGuardVersion,
            Duration ttl) {
        Long result = eval(
                DOUBLE_GUARDED_CREATE_SCRIPT,
                3,
                bytes(requireKey(key)),
                bytes(requireKey(guardKey)),
                bytes(requireKey(secondaryGuardKey)),
                bytes(version),
                bytes(guardVersion),
                bytes(secondaryGuardVersion),
                encode(value),
                bytes(ttlMillis(ttl)));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean createIfAbsent(String key,
                                  Serializable value,
                                  long version,
                                  String guardKey,
                                  long guardVersion,
                                  Duration ttl) {
        Long result = eval(
                GUARDED_CREATE_SCRIPT,
                2,
                bytes(requireKey(key)),
                bytes(requireKey(guardKey)),
                bytes(version),
                bytes(guardVersion),
                encode(value),
                bytes(ttlMillis(ttl)));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean compareAndSet(String key,
                                 long expectedVersion,
                                 Serializable value,
                                 long newVersion,
                                 Duration ttl) {
        Long result = eval(
                CAS_SCRIPT,
                1,
                bytes(requireKey(key)),
                bytes(expectedVersion),
                bytes(newVersion),
                encode(value),
                bytes(ttlMillis(ttl)));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public VersionedFields loadFields(String key) {
        requireKey(key);
        try (RedisConnection connection = connectionFactory.getConnection()) {
            Map<byte[], byte[]> encoded = connection.hashCommands().hGetAll(bytes(key));
            if (encoded == null || encoded.isEmpty()) {
                return null;
            }
            Long version = null;
            Map<String, Object> fields = new LinkedHashMap<>();
            for (Map.Entry<byte[], byte[]> entry : encoded.entrySet()) {
                String fieldName = new String(entry.getKey(), StandardCharsets.UTF_8);
                if ("version".equals(fieldName)) {
                    version = Long.parseLong(new String(entry.getValue(), StandardCharsets.UTF_8));
                    continue;
                }
                Object value = valueDecoder.apply(entry.getValue());
                fields.put(fieldName, value == NullFieldValue.INSTANCE ? null : value);
            }
            return version == null ? null : new VersionedFields(version, fields);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Long loadVersion(String key) {
        requireKey(key);
        try (RedisConnection connection = connectionFactory.getConnection()) {
            byte[] encoded = connection.hashCommands().hGet(bytes(key), bytes("version"));
            return encoded == null
                    ? null
                    : Long.parseLong(new String(encoded, StandardCharsets.UTF_8));
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean createFieldsIfAbsent(String key,
                                        Map<String, ? extends Serializable> fields,
                                        long version,
                                        Duration ttl) {
        Long result = eval(
                CREATE_FIELDS_SCRIPT,
                1,
                fieldArguments(
                        List.of(bytes(requireKey(key)), bytes(version)),
                        fields,
                        ttl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean createFieldsIfAbsent(
            String key,
            Map<String, ? extends Serializable> fields,
            long version,
            String guardKey,
            long guardVersion,
            String secondaryGuardKey,
            long secondaryGuardVersion,
            Duration ttl) {
        Long result = eval(
                DOUBLE_GUARDED_CREATE_FIELDS_SCRIPT,
                3,
                fieldArguments(
                        List.of(
                                bytes(requireKey(key)),
                                bytes(requireKey(guardKey)),
                                bytes(requireKey(secondaryGuardKey)),
                                bytes(version),
                                bytes(guardVersion),
                                bytes(secondaryGuardVersion)),
                        fields,
                        ttl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean createFieldsIfAbsent(
            String key,
            Map<String, ? extends Serializable> fields,
            long version,
            String guardKey,
            long guardVersion,
            String secondaryGuardKey,
            long secondaryGuardVersion,
            String tertiaryGuardKey,
            long tertiaryGuardVersion,
            Duration ttl) {
        Long result = eval(
                TRIPLE_GUARDED_CREATE_FIELDS_SCRIPT,
                4,
                fieldArguments(
                        List.of(
                                bytes(requireKey(key)),
                                bytes(requireKey(guardKey)),
                                bytes(requireKey(secondaryGuardKey)),
                                bytes(requireKey(tertiaryGuardKey)),
                                bytes(version),
                                bytes(guardVersion),
                                bytes(secondaryGuardVersion),
                                bytes(tertiaryGuardVersion)),
                        fields,
                        ttl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean createFieldsIfAbsent(String key,
                                        Map<String, ? extends Serializable> fields,
                                        long version,
                                        String guardKey,
                                        long guardVersion,
                                        Duration ttl) {
        Long result = eval(
                GUARDED_CREATE_FIELDS_SCRIPT,
                2,
                fieldArguments(
                        List.of(
                                bytes(requireKey(key)),
                                bytes(requireKey(guardKey)),
                                bytes(version),
                                bytes(guardVersion)),
                        fields,
                        ttl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean compareAndSetFields(String key,
                                       long expectedVersion,
                                       Map<String, ? extends Serializable> fields,
                                       long newVersion,
                                       Duration ttl) {
        Long result = eval(
                CAS_FIELDS_SCRIPT,
                1,
                fieldArguments(
                        List.of(
                                bytes(requireKey(key)),
                                bytes(expectedVersion),
                                bytes(newVersion)),
                        fields,
                        ttl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean compareAndSetFields(
            String key,
            long expectedVersion,
            Map<String, ? extends Serializable> fields,
            long newVersion,
            String guardKey,
            long guardVersion,
            String secondaryGuardKey,
            long secondaryGuardVersion,
            Duration ttl) {
        Long result = eval(
                DOUBLE_GUARDED_CAS_FIELDS_SCRIPT,
                3,
                fieldArguments(
                        List.of(
                                bytes(requireKey(key)),
                                bytes(requireKey(guardKey)),
                                bytes(requireKey(secondaryGuardKey)),
                                bytes(expectedVersion),
                                bytes(newVersion),
                                bytes(guardVersion),
                                bytes(secondaryGuardVersion)),
                        fields,
                        ttl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean compareAndSetFields(
            String key,
            long expectedVersion,
            Map<String, ? extends Serializable> fields,
            long newVersion,
            String guardKey,
            long guardVersion,
            String secondaryGuardKey,
            long secondaryGuardVersion,
            String tertiaryGuardKey,
            long tertiaryGuardVersion,
            Duration ttl) {
        Long result = eval(
                TRIPLE_GUARDED_CAS_FIELDS_SCRIPT,
                4,
                fieldArguments(
                        List.of(
                                bytes(requireKey(key)),
                                bytes(requireKey(guardKey)),
                                bytes(requireKey(secondaryGuardKey)),
                                bytes(requireKey(tertiaryGuardKey)),
                                bytes(expectedVersion),
                                bytes(newVersion),
                                bytes(guardVersion),
                                bytes(secondaryGuardVersion),
                                bytes(tertiaryGuardVersion)),
                        fields,
                        ttl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean compareAndSetFieldsAndRefresh(
            String key,
            long expectedVersion,
            Map<String, ? extends Serializable> fields,
            long newVersion,
            String guardKey,
            long guardVersion,
            String secondaryGuardKey,
            long secondaryGuardVersion,
            Duration ttl,
            String refreshKey,
            Duration refreshTtl) {
        Long result = eval(
                DOUBLE_GUARDED_CAS_FIELDS_REFRESH_SCRIPT,
                4,
                fieldArgumentsWithRefreshTtl(
                        List.of(
                                bytes(requireKey(key)),
                                bytes(requireKey(guardKey)),
                                bytes(requireKey(secondaryGuardKey)),
                                bytes(requireKey(refreshKey)),
                                bytes(expectedVersion),
                                bytes(newVersion),
                                bytes(guardVersion),
                                bytes(secondaryGuardVersion)),
                        fields,
                        ttl,
                        refreshTtl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean compareAndSetFields(String key,
                                       long expectedVersion,
                                       Map<String, ? extends Serializable> fields,
                                       long newVersion,
                                       String guardKey,
                                       long guardVersion,
                                       Duration ttl) {
        Long result = eval(
                GUARDED_CAS_FIELDS_SCRIPT,
                2,
                fieldArguments(
                        List.of(
                                bytes(requireKey(key)),
                                bytes(requireKey(guardKey)),
                                bytes(expectedVersion),
                                bytes(newVersion),
                                bytes(guardVersion)),
                        fields,
                        ttl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean rewriteAsFields(String key,
                                   long expectedVersion,
                                   Map<String, ? extends Serializable> fields,
                                   Duration ttl) {
        Long result = eval(
                REWRITE_FIELDS_SCRIPT,
                1,
                fieldArguments(
                        List.of(bytes(requireKey(key)), bytes(expectedVersion)),
                        fields,
                        ttl));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean compareAndSet(String key,
                                 long expectedVersion,
                                 Serializable value,
                                 long newVersion,
                                 String guardKey,
                                 long guardVersion,
                                 Duration ttl) {
        Long result = eval(
                GUARDED_CAS_SCRIPT,
                2,
                bytes(requireKey(key)),
                bytes(requireKey(guardKey)),
                bytes(expectedVersion),
                bytes(newVersion),
                bytes(guardVersion),
                encode(value),
                bytes(ttlMillis(ttl)));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean compareAndSet(
            String key,
            long expectedVersion,
            Serializable value,
            long newVersion,
            String guardKey,
            long guardVersion,
            String secondaryGuardKey,
            long secondaryGuardVersion,
            Duration ttl) {
        Long result = eval(
                DOUBLE_GUARDED_CAS_SCRIPT,
                3,
                bytes(requireKey(key)),
                bytes(requireKey(guardKey)),
                bytes(requireKey(secondaryGuardKey)),
                bytes(expectedVersion),
                bytes(newVersion),
                bytes(guardVersion),
                bytes(secondaryGuardVersion),
                encode(value),
                bytes(ttlMillis(ttl)));
        return Long.valueOf(1L).equals(result);
    }

    /**
     * 执行返回整数的 Redis Lua 脚本。
     *
     * @param script Lua 脚本
     * @param keyCount 参数中的 Redis key 数量
     * @param keysAndArgs key 与脚本参数
     * @return Redis 整数结果
     */
    private Long eval(byte[] script, int keyCount, byte[]... keysAndArgs) {
        try (RedisConnection connection = connectionFactory.getConnection()) {
            return connection.scriptingCommands().eval(
                    script, ReturnType.INTEGER, keyCount, keysAndArgs);
        }
    }

    /**
     * 使用项目既有 Java 协议编码对象。
     *
     * @param value 待编码对象
     * @return 编码后的二进制 payload
     */
    private byte[] encode(Serializable value) {
        return JavaValueEncoder.INSTANCE.apply(
                Objects.requireNonNull(value, "value must not be null"));
    }

    /**
     * 组装动态字段脚本参数，每个字段只编码一次。
     *
     * @param prefix Redis key 与固定参数
     * @param fields 待写字段
     * @param ttl 有效期
     * @return EVAL 的完整二进制参数
     */
    private byte[][] fieldArguments(List<byte[]> prefix,
                                    Map<String, ? extends Serializable> fields,
                                    Duration ttl) {
        Objects.requireNonNull(fields, "fields must not be null");
        List<byte[]> arguments = new ArrayList<>(prefix.size() + fields.size() * 2 + 1);
        arguments.addAll(prefix);
        for (Map.Entry<String, ? extends Serializable> entry : fields.entrySet()) {
            String fieldName = requireKey(entry.getKey());
            if ("version".equals(fieldName)) {
                throw new IllegalArgumentException("version is a reserved field");
            }
            Serializable value = entry.getValue() == null ? NullFieldValue.INSTANCE : entry.getValue();
            arguments.add(bytes(fieldName));
            arguments.add(encode(value));
        }
        arguments.add(bytes(ttlMillis(ttl)));
        return arguments.toArray(new byte[0][]);
    }

    /**
     * 组装需要同时刷新关联键 TTL 的字段脚本参数。
     *
     * @param prefix Redis key 与固定参数
     * @param fields 待写字段
     * @param ttl 主对象有效期
     * @param refreshTtl 关联键有效期
     * @return EVAL 的完整二进制参数
     */
    private byte[][] fieldArgumentsWithRefreshTtl(
            List<byte[]> prefix,
            Map<String, ? extends Serializable> fields,
            Duration ttl,
            Duration refreshTtl) {
        byte[][] base = fieldArguments(prefix, fields, ttl);
        byte[][] arguments = java.util.Arrays.copyOf(base, base.length + 1);
        arguments[base.length] = bytes(ttlMillis(refreshTtl));
        return arguments;
    }

    /**
     * 校验并换算有效期。
     *
     * @param ttl 有效期
     * @return 至少一毫秒的有效期
     */
    private long ttlMillis(Duration ttl) {
        Objects.requireNonNull(ttl, "ttl must not be null");
        if (ttl.isNegative() || ttl.isZero()) {
            throw new IllegalArgumentException("ttl must be positive");
        }
        return Math.max(1L, ttl.toMillis());
    }

    /**
     * 校验 Redis key。
     *
     * @param key Redis key
     * @return 原 key
     */
    private String requireKey(String key) {
        if (key == null || key.isBlank()) {
            throw new IllegalArgumentException("key must not be blank");
        }
        return key;
    }

    /**
     * 将文本转为 Redis 二进制参数。
     *
     * @param value 文本
     * @return UTF-8 字节
     */
    private static byte[] bytes(Object value) {
        return String.valueOf(value).getBytes(StandardCharsets.UTF_8);
    }

    /**
     * Redis Hash 中显式保存的空字段占位。
     */
    private enum NullFieldValue {
        INSTANCE
    }
}
