70%

src/main/java/com/adx/tencent/conversionsync/ConversionSyncService.java

@@ -22,7 +22,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.time.Instant;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
@@ -34,7 +33,6 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.zip.CRC32;
 
 /**
  * 统一转化同步：
@@ -48,6 +46,7 @@ public class ConversionSyncService {
     private static final Logger log = LoggerFactory.getLogger(ConversionSyncService.class);
     private static final int REALTIME_CONCURRENCY = 10;
     private static final int BACKFILL_CONCURRENCY = 16;
+    private static final int DEDUCTION_WINDOW_SIZE = 100;
 
     private final ConversionClient baiduClient;
     private final RedisHotStore hotStore;
@@ -588,7 +587,23 @@ public class ConversionSyncService {
         if (!isNewTrackingLink(task.tencentBid)) {
             return false;
         }
-        return bucketOf(task) < deductionCountPerTen(deductionRate);
+        if (!hotStore.markConversionSeen(task.callbackKey)) {
+            log.info("[ConversionSync] 腾讯转化已见过，跳过扣量计数 | callbackKey={} | qk={} | accountId={}",
+                    task.callbackKey, task.payment.getQk(), safe(task.tencentBid.getAccountId()));
+            return false;
+        }
+        long counter = hotStore.incrementDeductionCounter(
+                task.tencentBid.getAccountId(),
+                task.payment.getAct(),
+                resolveTrackingVersion(task.tencentBid),
+                task.payment.getDate()
+        );
+        int slot = deductionSlot(counter);
+        boolean deduct = slot < deductionRate;
+        log.info("[ConversionSync] 腾讯转化扣量判定 | qk={} | act={} | accountId={} | version={} | rate={} | counter={} | slot={} | callbackKey={} | deduct={}",
+                task.payment.getQk(), task.payment.getAct(), safe(task.tencentBid.getAccountId()),
+                resolveTrackingVersion(task.tencentBid), deductionRate, counter, slot, task.callbackKey, deduct);
+        return deduct;
     }
 
     private static boolean isNewTrackingLink(BidRecord bid) {
@@ -605,19 +620,6 @@ public class ConversionSyncService {
                 || "v4".equalsIgnoreCase(version);
     }
 
-    private int bucketOf(CallbackTask task) {
-        CRC32 crc32 = new CRC32();
-        crc32.update(String.join("|",
-                safe(task.tencentBid.getMedia()),
-                safe(task.payment.getQk()),
-                safe(task.payment.getDate()),
-                safe(task.payment.getDeviceId()),
-                safe(task.tencentBid.getTagId()),
-                String.valueOf(task.payment.getAct()),
-                safe(task.tencentActionType)).getBytes(StandardCharsets.UTF_8));
-        return (int) (crc32.getValue() % 10);
-    }
-
     private MediaCallbackRecord buildDeductedCallbackRecord(CallbackTask task) {
         MediaCallbackRecord record = new MediaCallbackRecord();
         Instant now = Instant.now();
@@ -639,18 +641,15 @@ public class ConversionSyncService {
         record.setTrackingVersion(resolveTrackingVersion(task.tencentBid));
         record.setCreatedAt(now);
         int deductionRate = resolveDeductionRate(task.tencentBid);
-        log.info("[ConversionSync] 腾讯转化扣量 | qk={} | act={} | rate={} | tagId={} | bucket={} | callbackKey={}",
+        log.info("[ConversionSync] 腾讯转化扣量 | qk={} | act={} | rate={} | tagId={} | accountId={} | callbackKey={}",
                 task.payment.getQk(), task.payment.getAct(), deductionRate,
-                task.tencentBid.getTagId(), bucketOf(task), task.callbackKey);
+                task.tencentBid.getTagId(), safe(task.tencentBid.getAccountId()), task.callbackKey);
         return record;
     }
 
-    private int deductionCountPerTen(int deductionRate) {
-        int count = deductionRate / 10;
-        if (count > 10) {
-            return 10;
-        }
-        return Math.max(count, 0);
+    private int deductionSlot(long counter) {
+        long normalized = Math.max(counter - 1L, 0L);
+        return (int) (normalized % DEDUCTION_WINDOW_SIZE);
     }
 
     private static String resolveTrackingVersion(BidRecord bid) {

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

@@ -23,6 +23,8 @@ import java.util.concurrent.ConcurrentHashMap;
 public class RedisHotStore {
 
     public static final String BID_NOT_FOUND_MSG = "bid not found";
+    private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(45);
+    private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
     private static final String DEFAULT_PREFIX = "adx:";
     private static final String DEFAULT_STREAM  = "adx:events";
@@ -202,6 +204,37 @@ public class RedisHotStore {
         return parseBidRecord(payload);
     }
 
+    /**
+     * 首次见到某条转化时返回 true，并写入带 TTL 的判重标记。
+     * 使用 SETNX 保证多实例并发下只会有一个消费者把它视为“新转化”。
+     */
+    public boolean markConversionSeen(String dedupeKey) {
+        if (dedupeKey == null || dedupeKey.isBlank()) {
+            return false;
+        }
+        Boolean created = redis.opsForValue().setIfAbsent(
+                conversionSeenKey(dedupeKey),
+                "1",
+                DEFAULT_CONVERSION_SEEN_TTL
+        );
+        return Boolean.TRUE.equals(created);
+    }
+
+    /**
+     * 按账户/事件/版本/日期维度维护一个递增计数器，供窗口扣量使用。
+     */
+    public long incrementDeductionCounter(String accountId, int act, String trackingVersion, String date) {
+        String key = deductionCounterKey(accountId, act, trackingVersion, date);
+        Long count = redis.opsForValue().increment(key);
+        if (count == null) {
+            throw new RuntimeException("increment deduction counter failed");
+        }
+        if (count == 1L) {
+            redis.expire(key, DEFAULT_DEDUCTION_COUNTER_TTL);
+        }
+        return count;
+    }
+
     // ─── EventQueue ──────────────────────────────────────────────────────────
 
     /**
@@ -337,6 +370,26 @@ public class RedisHotStore {
         return prefix + "media:" + media + ":" + traceId;
     }
 
+    private String conversionSeenKey(String dedupeKey) {
+        return prefix + "conv:seen:" + dedupeKey;
+    }
+
+    private String deductionCounterKey(String accountId, int act, String trackingVersion, String date) {
+        return String.format("%sdeduct:counter:tencent:acct:%s:act:%d:ver:%s:date:%s",
+                prefix,
+                sanitizeCounterPart(accountId, "unknown"),
+                act,
+                sanitizeCounterPart(trackingVersion, "v1"),
+                sanitizeCounterPart(date, "unknown"));
+    }
+
+    private static String sanitizeCounterPart(String value, String fallback) {
+        if (value == null || value.isBlank()) {
+            return fallback;
+        }
+        return value.replace(':', '_').trim();
+    }
+
     private BidRecord parseBidRecord(String payload) {
         try {
             return objectMapper.readValue(payload, BidRecord.class);