feat(mq-starter-redis-stream): 去重 Redis 默认实现 RedisMqIdempotentStore

- 实现 MqIdempotentStore：setIfAbsent 原子占用 + processing/success/fail 状态机 + TTL
- 修正 legacy 读写 key 不一致缺陷（读写统一前缀键）
- autoconfig 在 redis-stream 模式装配为默认 store（@ConditionalOnMissingBean）
- 集成测试连真实 redis 验证状态机；@Bean 声明契约护栏

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java

@@ -0,0 +1,60 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.mq.spi.MqIdempotentStore;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.time.Duration;
+
+/**
+ * 基于 Redis 的去重默认实现。状态机 processing / success / fail，各带 TTL。
+ *
+ * <p>对齐 legacy filter+consumeSuccess+saveException 的语义，但修正了 legacy「读裸 key、写带前缀 key」
+ * 导致去重失效的缺陷：本实现读写统一使用 {@code prefix + key}。
+ */
+public class RedisMqIdempotentStore implements MqIdempotentStore {
+
+    private static final String PROCESSING = "processing";
+    private static final String SUCCESS = "success";
+    private static final String FAIL = "fail";
+
+    private final StringRedisTemplate redis;
+    private final String prefix;
+    private final long processingTtl;
+    private final long successTtl;
+    private final long failTtl;
+
+    public RedisMqIdempotentStore(StringRedisTemplate redis, String prefix,
+                                  long processingTtlSeconds, long successTtlSeconds, long failTtlSeconds) {
+        this.redis = redis;
+        this.prefix = prefix;
+        this.processingTtl = processingTtlSeconds;
+        this.successTtl = successTtlSeconds;
+        this.failTtl = failTtlSeconds;
+    }
+
+    @Override
+    public boolean tryAcquire(String key) {
+        String k = prefix + key;
+        // setIfAbsent 原子占用：key 不存在时写 processing 并放行。
+        Boolean acquired = redis.opsForValue().setIfAbsent(k, PROCESSING, Duration.ofSeconds(processingTtl));
+        if (Boolean.TRUE.equals(acquired)) {
+            return true;
+        }
+        // key 已存在：fail（上次失败）允许重新消费；processing / success 一律挡回。
+        if (FAIL.equals(redis.opsForValue().get(k))) {
+            redis.opsForValue().set(k, PROCESSING, Duration.ofSeconds(processingTtl));
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public void markSuccess(String key) {
+        redis.opsForValue().set(prefix + key, SUCCESS, Duration.ofSeconds(successTtl));
+    }
+
+    @Override
+    public void markFail(String key) {
+        redis.opsForValue().set(prefix + key, FAIL, Duration.ofSeconds(failTtl));
+    }
+}

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java

@@ -3,6 +3,7 @@ import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
 import com.njcn.middle.stream.support.StreamKeyBuilder;
 import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
 import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqIdempotentStore;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -21,4 +22,11 @@ public class RedisStreamMqDriverAutoConfiguration {
     public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder, RedisStreamProperties props) {
         return new RedisStreamMqDriver(redis, streamKeyBuilder, props);
     }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MqIdempotentStore mqIdempotentStore(StringRedisTemplate redis) {
+        // 默认 TTL：processing 60s / success 300s / fail 300s，对齐 legacy 量级；统一前缀（修正 key 一致性）。
+        return new RedisMqIdempotentStore(redis, "mq:idem:", 60, 300, 300);
+    }
 }

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java

@@ -0,0 +1,38 @@
+package com.njcn.mq.driver.redis;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+class RedisMqIdempotentStoreIT {
+
+    @Test
+    void stateMachine_firstAcquire_thenBlocked_failReopens_successBlocks() {
+        StringRedisTemplate redis = newRedisOrSkip();
+        RedisMqIdempotentStore store = new RedisMqIdempotentStore(redis, "mq:idem:it:", 60, 300, 300);
+        String key = "k_" + System.nanoTime();
+
+        assertTrue(store.tryAcquire(key), "首次应放行并占用 processing");
+        assertFalse(store.tryAcquire(key), "processing 中应挡回");
+
+        store.markFail(key);
+        assertTrue(store.tryAcquire(key), "上次 fail 后应允许重新消费");
+
+        store.markSuccess(key);
+        assertFalse(store.tryAcquire(key), "success 后应挡回，避免重复消费");
+    }
+
+    private static StringRedisTemplate newRedisOrSkip() {
+        try {
+            LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379);
+            cf.afterPropertiesSet();
+            StringRedisTemplate t = new StringRedisTemplate(cf);
+            t.afterPropertiesSet();
+            t.hasKey("ping");
+            return t;
+        } catch (Exception e) { assumeTrue(false, "no local redis, skip"); return null; }
+    }
+}

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java

@@ -1,8 +1,13 @@
 package com.njcn.mq.driver.redis;
 
 import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
+import com.njcn.mq.spi.MqIdempotentStore;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.context.annotation.Bean;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.lang.reflect.Method;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -30,4 +35,16 @@ class RedisStreamMqDriverWiringTest {
                 .as("@AutoConfigureBefore 必须指向 MqCoreAutoConfiguration")
                 .contains(MqCoreAutoConfiguration.class);
     }
+
+    @Test
+    void declaresIdempotentStoreBean() throws Exception {
+        Method m = RedisStreamMqDriverAutoConfiguration.class
+                .getDeclaredMethod("mqIdempotentStore", StringRedisTemplate.class);
+        assertThat(m.getAnnotation(Bean.class))
+                .as("redis-stream 模式应提供默认去重实现")
+                .isNotNull();
+        assertThat(MqIdempotentStore.class)
+                .as("默认去重实现须为 MqIdempotentStore 类型")
+                .isAssignableFrom(m.getReturnType());
+    }
 }