import io.lettuce.core.ReadFrom;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.RedisPassword;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;

import java.time.Duration;

/**
 * Redis 從讀（Replica Read）配置
 *
 * 副本連接工廠 ReadFrom = SLAVE_PREFERRED（優(yōu)先從庫，從庫不可用回退主庫）。
 * 復用主庫的集群拓撲/密碼/連接池/超時，僅 ReadFrom 不同。
 * 注意：副本模板僅用于讀命令；集群模式下寫命令始終由主節(jié)點處理。
 */
@Configuration
public class ReplicaRedisConfig {

    @Bean
    public LettuceConnectionFactory replicaConnectionFactory(RedisProperties redisProperties) {
        if (redisProperties.getCluster() == null
                || redisProperties.getCluster().getNodes() == null
                || redisProperties.getCluster().getNodes().isEmpty()) {
            throw new IllegalStateException(
                    "ReplicaRedisConfig 需要集群配置 spring.redis.cluster.nodes，當前未配置");
        }

        RedisClusterConfiguration clusterConfig = new RedisClusterConfiguration();
        for (String node : redisProperties.getCluster().getNodes()) {
            String[] parts = node.split(":");
            clusterConfig.clusterNode(parts[0].trim(), Integer.parseInt(parts[1].trim()));
        }
        clusterConfig.setMaxRedirects(redisProperties.getCluster().getMaxRedirects());
        if (redisProperties.getPassword() != null && !redisProperties.getPassword().isEmpty()) {
            clusterConfig.setPassword(RedisPassword.of(redisProperties.getPassword()));
        }

        GenericObjectPoolConfig<?> poolConfig = new GenericObjectPoolConfig<>();
        RedisProperties.Pool pool = redisProperties.getLettuce().getPool();
        if (pool != null) {
            poolConfig.setMaxTotal(pool.getMaxActive());
            poolConfig.setMaxIdle(pool.getMaxIdle());
            poolConfig.setMinIdle(pool.getMinIdle());
            if (pool.getMaxWait() != null) {
                poolConfig.setMaxWaitMillis(pool.getMaxWait().toMillis());
            }
        }

        LettucePoolingClientConfiguration.LettucePoolingClientConfigurationBuilder builder =
                LettucePoolingClientConfiguration.builder()
                        .readFrom(ReadFrom.SLAVE_PREFERRED)   // 關鍵：優(yōu)先從庫
                        .poolConfig(poolConfig);

        Duration timeout = redisProperties.getTimeout();
        if (timeout != null) {
            builder.commandTimeout(timeout);
        }

        LettuceConnectionFactory factory = new LettuceConnectionFactory(clusterConfig, builder.build());
        factory.afterPropertiesSet();
        return factory;
    }

    /** 副本純字符串序列化模板 */
    @Bean
    public RedisTemplate<String, Object> customStringRedisTemplateReplica(
            LettuceConnectionFactory replicaConnectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(replicaConnectionFactory);
        template.setKeySerializer(RedisSerializer.string());
        template.setHashKeySerializer(RedisSerializer.string());
        template.setValueSerializer(RedisSerializer.string());
        template.setHashValueSerializer(RedisSerializer.string());
        template.afterPropertiesSet();
        return template;
    }

    // 其他副本模板（gzip / Protobuf+Zstd）序列化器與主庫保持一致，略
}

@Autowired
@Qualifier("customStringRedisTemplateReplica")
private RedisTemplate<String, Object> customStringRedisTemplateReplica;

@Autowired
@Qualifier("gzipRedisTemplateReplica")
private RedisTemplate<String, Object> gzipRedisTemplateReplica;

@Autowired
@Qualifier("menuRedisTemplateReplica")
private RedisTemplate<String, Map<String, Menu>> menuRedisTemplateReplica;

// ==================== 從讀（Replica Read）方法 ====================
// 走副本連接工廠（ReadFrom = SLAVE_PREFERRED），僅用于可容忍主從復制延遲的讀場景。
// 現(xiàn)有同名（不帶 FromReplica）方法仍走主庫，語義不變。

public Object getStrFromReplica(String key) {
    return key == null ? null : customStringRedisTemplateReplica.opsForValue().get(key);
}

public Object hgetStringFromReplica(String key, String field) throws IOException {
    final Object object = customStringRedisTemplateReplica.opsForHash().get(key, field);
    if (object == null) {
        log.debug("data is null");
        return null;
    }
    return object;
}

public boolean existsFromReplica(String key) {
    return customStringRedisTemplateReplica.hasKey(key);
}

public Map<String, Menu> loadCompressedMenuByProtobufZstdFromReplica(String key) {
    try {
        Map<String, Menu> value = menuRedisTemplateReplica.opsForValue().get(key);
        return value == null ? null : value;
    } catch (Exception e) {
        throw new RuntimeException("Failed to load compressed menu from replica", e);
    }
}

// 改造前：走主庫
Object object = redisUtils.getStr(redisKey);

// 改造后：走從庫優(yōu)先
Object object = redisUtils.getStrFromReplica(redisKey);