荣耀配置

src/main/java/com/adx/tencent/AppConfiguration.java

@@ -141,7 +141,8 @@ public class AppConfiguration {
                 coldStore,
                 "adx:tencent:",
                 props.getRedisStream(),
-                props.getBidTtl());
+                props.getBidTtl(),
+                props.getRedisStreamMaxLen());
     }
 
     @Bean
@@ -263,7 +264,7 @@ public class AppConfiguration {
                                        @Nullable HonorColdStore honorColdStore) {
         if (honorColdStore == null) return null;
         return new HonorHotStore(redis, objectMapper, honorColdStore,
-                "adx:honor:", props.getHonorRedisStream(), props.getBidTtl());
+                "adx:honor:", props.getHonorRedisStream(), props.getBidTtl(), props.getRedisStreamMaxLen());
     }
 
     @Bean
@@ -337,7 +338,7 @@ public class AppConfiguration {
                                              @Nullable KuaishouColdStore kuaishouColdStore) {
         if (kuaishouColdStore == null) return null;
         return new KuaishouHotStore(redis, objectMapper, kuaishouColdStore,
-                "adx:kuaishou:", props.getKuaishouRedisStream(), props.getBidTtl());
+                "adx:kuaishou:", props.getKuaishouRedisStream(), props.getBidTtl(), props.getRedisStreamMaxLen());
     }
 
     @Bean
@@ -408,7 +409,7 @@ public class AppConfiguration {
                                      @Nullable VivoColdStore vivoColdStore) {
         if (vivoColdStore == null) return null;
         return new VivoHotStore(redis, objectMapper, vivoColdStore,
-                "adx:vivo:", props.getVivoRedisStream(), props.getBidTtl());
+                "adx:vivo:", props.getVivoRedisStream(), props.getBidTtl(), props.getRedisStreamMaxLen());
     }
 
     @Bean

src/main/java/com/adx/tencent/config/AppProperties.java

@@ -103,7 +103,7 @@ public class AppProperties {
     private int redisDb = 0;
     private String redisStream = "adx:tencent:events";
     private long redisStreamMaxLen = 50000;
-    private Duration bidTtl = Duration.ofDays(5);
+    private Duration bidTtl = Duration.ofDays(3);
     private Duration redisConnectTimeout = Duration.ofSeconds(5);
     private Duration redisCommandTimeout = Duration.ofSeconds(5);
     private Duration redisShutdownTimeout = Duration.ofMillis(100);

src/main/java/com/adx/tencent/honor/store/HonorHotStore.java

@@ -39,16 +39,19 @@ public class HonorHotStore {
     private final String prefix;
     private final String stream;
     private final Duration bidTtl;
+    private final long streamMaxLen;
     private final Set<String> initializedGroups = ConcurrentHashMap.newKeySet();
 
     public HonorHotStore(StringRedisTemplate redis, ObjectMapper objectMapper,
-                         HonorColdStore coldStore, String prefix, String stream, Duration bidTtl) {
+                         HonorColdStore coldStore, String prefix, String stream, Duration bidTtl,
+                         long streamMaxLen) {
         this.redis = redis;
         this.objectMapper = objectMapper;
         this.coldStore = coldStore;
         this.prefix = prefix;
         this.stream = stream;
         this.bidTtl = bidTtl;
+        this.streamMaxLen = streamMaxLen;
     }
 
     public void recordBid(HonorBidRecord record) {
@@ -57,6 +60,7 @@ public class HonorHotStore {
             String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
+                trimStreamIfNeeded(conn);
                 byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
                 long ttlMillis = bidTtl.toMillis();
                 conn.stringCommands().set(
@@ -87,6 +91,7 @@ public class HonorHotStore {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
             String payload = objectMapper.writeValueAsString(record);
+            trim(streamMaxLen);
             redis.opsForStream().add(MapRecord.create(stream, Map.of("type", "tracking", "payload", payload)).withStreamKey(stream));
         } catch (Exception e) {
             throw new RuntimeException("record honor tracking failed", e);
@@ -239,6 +244,13 @@ public class HonorHotStore {
         if (maxLen > 0) redis.opsForStream().trim(stream, maxLen, false);
     }
 
+    private void trimStreamIfNeeded(org.springframework.data.redis.connection.RedisConnection conn) {
+        if (streamMaxLen <= 0) {
+            return;
+        }
+        conn.streamCommands().xTrim(stream.getBytes(StandardCharsets.UTF_8), streamMaxLen);
+    }
+
     private String bidKey(String qk) { return prefix + "bid:" + qk; }
     private String mediaTraceKey(String traceId) { return prefix + "media:honor:" + traceId; }
 

src/main/java/com/adx/tencent/httpapi/AdminController.java

@@ -7,13 +7,19 @@ import com.adx.tencent.conversionsync.ConversionSyncRunner;
 import com.adx.tencent.conversionsync.ConversionSyncService;
 import com.adx.tencent.conversionsync.ManualTencentCallbackService;
 import com.adx.tencent.conversionsync.RetryService;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.http.HttpStatus;
 import org.springframework.lang.Nullable;
 import org.springframework.web.bind.annotation.*;
 import org.springframework.web.server.ResponseStatusException;
 
+import java.time.Duration;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 /**
  * 管理接口：手动触发转化同步和回调重试。
@@ -28,19 +34,22 @@ public class AdminController {
     private final ConversionSyncService conversionSyncer;
     private final RetryService retryService;
     private final ManualTencentCallbackService manualTencentCallbackService;
+    private final StringRedisTemplate redisTemplate;
 
     public AdminController(ConversionClient conversionClient,
                            @Nullable ConversionSyncRunner conversionSyncRunner,
                            @Nullable ConversionBackfillJobService backfillJobService,
                            @Nullable ConversionSyncService conversionSyncer,
                            @Nullable RetryService retryService,
-                           @Nullable ManualTencentCallbackService manualTencentCallbackService) {
+                           @Nullable ManualTencentCallbackService manualTencentCallbackService,
+                           @Nullable StringRedisTemplate redisTemplate) {
         this.conversionClient = conversionClient;
         this.conversionSyncRunner = conversionSyncRunner;
         this.backfillJobService = backfillJobService;
         this.conversionSyncer = conversionSyncer;
         this.retryService = retryService;
         this.manualTencentCallbackService = manualTencentCallbackService;
+        this.redisTemplate = redisTemplate;
     }
 
     @PostMapping("/conversions/query")
@@ -103,6 +112,57 @@ public class AdminController {
         return manualTencentCallbackService.replayDeductedCallback(id);
     }
 
+    @PostMapping("/redis/tighten-bid-media-ttl")
+    public Object tightenBidMediaTtl(@RequestBody(required = false) Map<String, Object> body) {
+        if (redisTemplate == null) {
+            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
+                    "redis template is not configured");
+        }
+        long ttlSeconds = toLong(body == null ? null : body.get("ttlSeconds"));
+        if (ttlSeconds <= 0) {
+            ttlSeconds = Duration.ofDays(3).getSeconds();
+        }
+        long scanCount = toLong(body == null ? null : body.get("scanCount"));
+        if (scanCount <= 0) {
+            scanCount = 2000;
+        }
+        long maxKeys = toLong(body == null ? null : body.get("maxKeys"));
+        if (maxKeys <= 0) {
+            maxKeys = 200000;
+        }
+        List<String> patterns = parseStringList(body == null ? null : body.get("patterns"));
+        if (patterns.isEmpty()) {
+            patterns = defaultBidMediaPatterns();
+        }
+
+        List<Map<String, Object>> results = new ArrayList<>();
+        long totalScanned = 0;
+        long totalUpdated = 0;
+        for (String pattern : patterns) {
+            TtlTightenResult result = tightenTtl(pattern, ttlSeconds, scanCount, Math.max(0, maxKeys - totalScanned));
+            results.add(Map.of(
+                    "pattern", pattern,
+                    "scanned", result.scanned(),
+                    "updated", result.updated(),
+                    "stoppedByLimit", result.stoppedByLimit()
+            ));
+            totalScanned += result.scanned();
+            totalUpdated += result.updated();
+            if (totalScanned >= maxKeys) {
+                break;
+            }
+        }
+        return Map.of(
+                "ttlSeconds", ttlSeconds,
+                "ttlHuman", Duration.ofSeconds(ttlSeconds).toString(),
+                "scanCount", scanCount,
+                "maxKeys", maxKeys,
+                "scanned", totalScanned,
+                "updated", totalUpdated,
+                "patterns", results
+        );
+    }
+
     private static int toInt(Object v) {
         if (v == null) return 0;
         if (v instanceof Number n) return n.intValue();
@@ -115,6 +175,58 @@ public class AdminController {
         try { return Long.parseLong(v.toString()); } catch (NumberFormatException e) { return 0L; }
     }
 
+    private TtlTightenResult tightenTtl(String pattern, long ttlSeconds, long scanCount, long maxKeys) {
+        if (maxKeys <= 0 || pattern == null || pattern.isBlank()) {
+            return new TtlTightenResult(0, 0, true);
+        }
+        long scanned = 0;
+        long updated = 0;
+        ScanOptions options = ScanOptions.scanOptions().match(pattern).count(scanCount).build();
+        try (Cursor<String> cursor = redisTemplate.scan(options)) {
+            while (cursor.hasNext()) {
+                String key = cursor.next();
+                scanned++;
+                Long currentTtl = redisTemplate.getExpire(key, TimeUnit.SECONDS);
+                if (currentTtl == null || currentTtl < 0 || currentTtl > ttlSeconds) {
+                    Boolean ok = redisTemplate.expire(key, Duration.ofSeconds(ttlSeconds));
+                    if (Boolean.TRUE.equals(ok)) {
+                        updated++;
+                    }
+                }
+                if (scanned >= maxKeys) {
+                    return new TtlTightenResult(scanned, updated, true);
+                }
+            }
+        }
+        return new TtlTightenResult(scanned, updated, false);
+    }
+
+    private static List<String> parseStringList(Object raw) {
+        if (!(raw instanceof Iterable<?> iterable)) {
+            return List.of();
+        }
+        List<String> values = new ArrayList<>();
+        for (Object item : iterable) {
+            if (item != null && !item.toString().isBlank()) {
+                values.add(item.toString());
+            }
+        }
+        return values;
+    }
+
+    private static List<String> defaultBidMediaPatterns() {
+        return List.of(
+                "adx:bid:*",
+                "adx:media:*",
+                "adx:honor:bid:*",
+                "adx:honor:media:*",
+                "adx:kuaishou:bid:*",
+                "adx:kuaishou:media:*",
+                "adx:vivo:bid:*",
+                "adx:vivo:media:*"
+        );
+    }
+
     private static List<Integer> parseOffsets(Map<String, Object> body) {
         if (body == null || body.isEmpty()) {
             return defaultBackfillOffsets();
@@ -133,4 +245,6 @@ public class AdminController {
     private static List<Integer> defaultBackfillOffsets() {
         return List.of(-1, -2, -3, -4, -5, -6, -7);
     }
+
+    private record TtlTightenResult(long scanned, long updated, boolean stoppedByLimit) {}
 }

src/main/java/com/adx/tencent/kuaishou/store/KuaishouHotStore.java

@@ -44,17 +44,20 @@ public class KuaishouHotStore {
     private final String prefix;
     private final String stream;
     private final Duration bidTtl;
+    private final long streamMaxLen;
     private final DefaultRedisScript<List> deductionDecisionScript;
     private final Set<String> initializedGroups = ConcurrentHashMap.newKeySet();
 
     public KuaishouHotStore(StringRedisTemplate redis, ObjectMapper objectMapper,
-                            KuaishouColdStore coldStore, String prefix, String stream, Duration bidTtl) {
+                            KuaishouColdStore coldStore, String prefix, String stream, Duration bidTtl,
+                            long streamMaxLen) {
         this.redis = redis;
         this.objectMapper = objectMapper;
         this.coldStore = coldStore;
         this.prefix = prefix;
         this.stream = stream;
         this.bidTtl = bidTtl;
+        this.streamMaxLen = streamMaxLen;
         this.deductionDecisionScript = buildDeductionDecisionScript();
     }
 
@@ -64,6 +67,7 @@ public class KuaishouHotStore {
             String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
+                trimStreamIfNeeded(conn);
                 byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
                 long ttlMillis = bidTtl.toMillis();
                 if (record.getQk() != null && !record.getQk().isBlank()) {
@@ -97,6 +101,7 @@ public class KuaishouHotStore {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
             String payload = objectMapper.writeValueAsString(record);
+            trim(streamMaxLen);
             redis.opsForStream().add(MapRecord.create(stream, Map.of("type", "tracking", "payload", payload)).withStreamKey(stream));
         } catch (Exception e) {
             throw new RuntimeException("record kuaishou tracking failed", e);
@@ -281,6 +286,13 @@ public class KuaishouHotStore {
         if (maxLen > 0) redis.opsForStream().trim(stream, maxLen, false);
     }
 
+    private void trimStreamIfNeeded(org.springframework.data.redis.connection.RedisConnection conn) {
+        if (streamMaxLen <= 0) {
+            return;
+        }
+        conn.streamCommands().xTrim(stream.getBytes(StandardCharsets.UTF_8), streamMaxLen);
+    }
+
     private String bidKey(String qk) { return prefix + "bid:" + qk; }
     private String mediaTraceKey(String traceId) { return prefix + "media:kuaishou:" + traceId; }
     private String conversionSeenKey(String dedupeKey) { return prefix + "conv:seen:kuaishou:" + dedupeKey; }

src/main/java/com/adx/tencent/storage/RedisHotStore.java

@@ -36,17 +36,20 @@ public class RedisHotStore {
     private final String prefix;
     private final String stream;
     private final Duration bidTtl;
+    private final long streamMaxLen;
     private final DefaultRedisScript<List> deductionDecisionScript;
     private final Set<String> initializedGroups = ConcurrentHashMap.newKeySet();
 
     public RedisHotStore(StringRedisTemplate redis, ObjectMapper objectMapper,
-                         TiDBColdStore coldStore, String prefix, String stream, Duration bidTtl) {
+                         TiDBColdStore coldStore, String prefix, String stream, Duration bidTtl,
+                         long streamMaxLen) {
         this.redis = redis;
         this.objectMapper = objectMapper;
         this.coldStore = coldStore;
         this.prefix = (prefix == null || prefix.isBlank()) ? DEFAULT_PREFIX : prefix;
         this.stream = (stream == null || stream.isBlank()) ? DEFAULT_STREAM : stream;
         this.bidTtl = (bidTtl == null || bidTtl.isZero()) ? Duration.ofHours(24) : bidTtl;
+        this.streamMaxLen = streamMaxLen;
         this.deductionDecisionScript = buildDeductionDecisionScript();
     }
 
@@ -67,6 +70,7 @@ public class RedisHotStore {
             String fullPayload = objectMapper.writeValueAsString(record);
 
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
+                trimStreamIfNeeded(conn);
                 byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
                 long ttlMillis = bidTtl.toMillis();
                 if (record.getQk() != null && !record.getQk().isEmpty()) {
@@ -108,6 +112,7 @@ public class RedisHotStore {
         }
         try {
             String payload = objectMapper.writeValueAsString(record);
+            trim(streamMaxLen);
             redis.opsForStream().add(
                     MapRecord.create(stream, Map.of("type", "tracking", "payload", payload))
                             .withStreamKey(stream)
@@ -380,6 +385,13 @@ public class RedisHotStore {
         redis.opsForStream().trim(stream, maxLen, false);
     }
 
+    private void trimStreamIfNeeded(org.springframework.data.redis.connection.RedisConnection conn) {
+        if (streamMaxLen <= 0) {
+            return;
+        }
+        conn.streamCommands().xTrim(stream.getBytes(StandardCharsets.UTF_8), streamMaxLen);
+    }
+
     // ─── 分布式锁支持（供 RedisLock 使用）────────────────────────────────────
 
     public StringRedisTemplate getRedisTemplate() {

src/main/java/com/adx/tencent/vivo/store/VivoHotStore.java

@@ -44,17 +44,20 @@ public class VivoHotStore {
     private final String prefix;
     private final String stream;
     private final Duration bidTtl;
+    private final long streamMaxLen;
     private final DefaultRedisScript<List> deductionDecisionScript;
     private final Set<String> initializedGroups = ConcurrentHashMap.newKeySet();
 
     public VivoHotStore(StringRedisTemplate redis, ObjectMapper objectMapper,
-                            VivoColdStore coldStore, String prefix, String stream, Duration bidTtl) {
+                            VivoColdStore coldStore, String prefix, String stream, Duration bidTtl,
+                            long streamMaxLen) {
         this.redis = redis;
         this.objectMapper = objectMapper;
         this.coldStore = coldStore;
         this.prefix = prefix;
         this.stream = stream;
         this.bidTtl = bidTtl;
+        this.streamMaxLen = streamMaxLen;
         this.deductionDecisionScript = buildDeductionDecisionScript();
     }
 
@@ -64,6 +67,7 @@ public class VivoHotStore {
             String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
+                trimStreamIfNeeded(conn);
                 byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
                 long ttlMillis = bidTtl.toMillis();
                 if (record.getQk() != null && !record.getQk().isBlank()) {
@@ -97,6 +101,7 @@ public class VivoHotStore {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
             String payload = objectMapper.writeValueAsString(record);
+            trim(streamMaxLen);
             redis.opsForStream().add(MapRecord.create(stream, Map.of("type", "tracking", "payload", payload)).withStreamKey(stream));
         } catch (Exception e) {
             throw new RuntimeException("record vivo tracking failed", e);
@@ -281,6 +286,13 @@ public class VivoHotStore {
         if (maxLen > 0) redis.opsForStream().trim(stream, maxLen, false);
     }
 
+    private void trimStreamIfNeeded(org.springframework.data.redis.connection.RedisConnection conn) {
+        if (streamMaxLen <= 0) {
+            return;
+        }
+        conn.streamCommands().xTrim(stream.getBytes(StandardCharsets.UTF_8), streamMaxLen);
+    }
+
     private String bidKey(String qk) { return prefix + "bid:" + qk; }
     private String mediaTraceKey(String traceId) { return prefix + "media:vivo:" + traceId; }
     private String conversionSeenKey(String dedupeKey) { return prefix + "conv:seen:vivo:" + dedupeKey; }

src/main/resources/application-prod.yml

@@ -49,6 +49,7 @@ adx:
   redis-username: "hcst"
   redis-password: "hcst@2026qwe.."
   redis-db: 5
+  bid-ttl: 3d
   redis-connect-timeout: 5s
   redis-command-timeout: 5s
   redis-shutdown-timeout: 100ms