扣量

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

@@ -321,7 +321,7 @@ public class AppConfiguration {
                                                          HonorClient honorClient) {
         if (coldStore == null) return null;
         return new ConversionSyncService(baiduClient, hotStore, tencentClient, coldStore, tagEventResolver,
-                honorTagEventResolver, honorHotStore, honorColdStore, honorClient);
+                honorTagEventResolver, honorHotStore, honorColdStore, honorClient, props);
     }
 
     @Bean

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

@@ -64,6 +64,7 @@ public class AppProperties {
 
     // Tencent Act Mapping: baiduAct -> tencentActionType
     private Map<Integer, String> tencentActMap = new HashMap<>();
+    private int tencentNewLinkDeductionRate = 20;
 
     // Redis
     private String redisAddr = "127.0.0.1:6379";
@@ -432,6 +433,14 @@ public class AppProperties {
         this.tencentActMap = v;
     }
 
+    public int getTencentNewLinkDeductionRate() {
+        return tencentNewLinkDeductionRate;
+    }
+
+    public void setTencentNewLinkDeductionRate(int v) {
+        this.tencentNewLinkDeductionRate = v;
+    }
+
     public String getRedisAddr() {
         return redisAddr;
     }

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

@@ -4,6 +4,7 @@ import com.adx.tencent.baidu.ConversionClient;
 import com.adx.tencent.baidu.model.ConversionQuery;
 import com.adx.tencent.baidu.model.ConversionResponse;
 import com.adx.tencent.baidu.model.PaymentInfo;
+import com.adx.tencent.config.AppProperties;
 import com.adx.tencent.honor.client.HonorClient;
 import com.adx.tencent.honor.model.HonorBidRecord;
 import com.adx.tencent.honor.model.HonorMediaCallbackRecord;
@@ -21,6 +22,7 @@ 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;
@@ -32,6 +34,7 @@ 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;
 
 /**
  * 统一转化同步：
@@ -57,6 +60,7 @@ public class ConversionSyncService {
     private final HonorTagEventResolver honorTagEventResolver;
     private final ExecutorService realtimeExecutor;
     private final ExecutorService backfillExecutor;
+    private final int tencentNewLinkDeductionRate;
 
     public enum SyncMode {
         REALTIME,
@@ -71,7 +75,8 @@ public class ConversionSyncService {
                                  HonorTagEventResolver honorTagEventResolver,
                                  HonorHotStore honorHotStore,
                                  HonorColdStore honorColdStore,
-                                 HonorClient honorClient) {
+                                 HonorClient honorClient,
+                                 AppProperties props) {
         this.baiduClient = baiduClient;
         this.hotStore = hotStore;
         this.tencentClient = tencentClient;
@@ -81,6 +86,8 @@ public class ConversionSyncService {
         this.honorHotStore = honorHotStore;
         this.honorColdStore = honorColdStore;
         this.honorClient = honorClient;
+        this.tencentNewLinkDeductionRate = sanitizeDeductionRate(
+                props != null ? props.getTencentNewLinkDeductionRate() : 20);
         this.realtimeExecutor = Executors.newFixedThreadPool(
                 REALTIME_CONCURRENCY,
                 r -> {
@@ -420,6 +427,10 @@ public class ConversionSyncService {
 
     private CallbackDispatchResult sendCallback(CallbackTask task) {
         if (task.isTencent()) {
+            if (shouldDeductTencentCallback(task)) {
+                MediaCallbackRecord deductedRecord = buildDeductedCallbackRecord(task);
+                return CallbackDispatchResult.tencent(task, deductedRecord);
+            }
             String platform = resolvePlatform(task.tencentBid);
             MediaCallbackRecord callbackRecord = tencentClient.sendConversionEvent(
                     task.tencentBid, task.payment, platform, task.tencentActionType);
@@ -569,6 +580,89 @@ public class ConversionSyncService {
                 conversionId);
     }
 
+    private boolean shouldDeductTencentCallback(CallbackTask task) {
+        if (!task.isTencent() || task.tencentBid == null) {
+            return false;
+        }
+        if (task.payment.getAct() == 2001 || tencentNewLinkDeductionRate <= 0) {
+            return false;
+        }
+        if (!isNewTrackingLink(task.tencentBid)) {
+            return false;
+        }
+        return bucketOf(task) < deductionCountPerTen();
+    }
+
+    private static boolean isNewTrackingLink(BidRecord bid) {
+        if (bid == null || bid.getMediaParams() == null) {
+            return false;
+        }
+        String flag = bid.getMediaParams().get(BidRecord.NEW_LINK_FLAG_KEY);
+        if (flag != null && Boolean.parseBoolean(flag)) {
+            return true;
+        }
+        String version = bid.getMediaParams().get(BidRecord.TRACKING_VERSION_KEY);
+        return "v2".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();
+        record.setDedupeKey(task.callbackKey);
+        record.setMedia("tencent");
+        record.setQk(task.payment.getQk());
+        String callbackUrl = task.tencentBid.getMediaParams() != null
+                ? task.tencentBid.getMediaParams().get("callback")
+                : null;
+        record.setCallbackUrl(callbackUrl != null ? callbackUrl : "");
+        record.setEventType(task.payment.getAct());
+        record.setEventTimeMs(now.toEpochMilli());
+        record.setPurchase(task.payment.getPayment());
+        record.setStatus(0);
+        record.setOk(false);
+        record.setAttempt(1);
+        record.setErrorMessage("deducted_by_new_link_rate");
+        record.setDispatchStatus(MediaCallbackRecord.DISPATCH_STATUS_DEDUCTED);
+        record.setTrackingVersion(isNewTrackingLink(task.tencentBid) ? "v2" : "v1");
+        record.setCreatedAt(now);
+        log.info("[ConversionSync] 腾讯转化扣量 | qk={} | act={} | rate={} | tagId={} | bucket={} | callbackKey={}",
+                task.payment.getQk(), task.payment.getAct(), tencentNewLinkDeductionRate,
+                task.tencentBid.getTagId(), bucketOf(task), task.callbackKey);
+        return record;
+    }
+
+    private int deductionCountPerTen() {
+        int count = tencentNewLinkDeductionRate / 10;
+        if (count > 10) {
+            return 10;
+        }
+        return Math.max(count, 0);
+    }
+
+    private static String safe(String value) {
+        return value == null ? "" : value;
+    }
+
+    private static int sanitizeDeductionRate(int rate) {
+        if (rate < 0) {
+            return 0;
+        }
+        return Math.min(rate, 100);
+    }
+
     private static HonorMediaCallbackRecord toHonorCallbackRecord(CallbackTask task, MediaCallbackRecord record) {
         HonorMediaCallbackRecord callback = new HonorMediaCallbackRecord();
         callback.setDedupeKey(record.getDedupeKey());
@@ -603,6 +697,7 @@ public class ConversionSyncService {
         callbackRecord.setResponseBody(honorRecord.getResponseBody());
         callbackRecord.setErrorMessage(honorRecord.getErrorMessage());
         callbackRecord.setRequestBody(honorRecord.getRequestBody());
+        callbackRecord.setTrackingVersion("v1");
         callbackRecord.setCreatedAt(honorRecord.getCreatedAt());
         return callbackRecord;
     }

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

@@ -69,6 +69,8 @@ public class RetryService {
         copy.setAttempt(src.getAttempt());
         copy.setResponseBody(src.getResponseBody());
         copy.setErrorMessage(src.getErrorMessage());
+        copy.setDispatchStatus(src.getDispatchStatus());
+        copy.setTrackingVersion(src.getTrackingVersion());
         copy.setCreatedAt(src.getCreatedAt());
         return copy;
     }

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

@@ -69,14 +69,25 @@ public class TrackingController {
     public Object tencentImpression(HttpServletRequest request) throws Exception {
         impressionSemaphore.acquire();
         try {
-            return doImpression(request);
+            return doImpression(request, false);
         } finally {
             impressionSemaphore.release();
         }
     }
 
-    private Object doImpression(HttpServletRequest request) throws Exception {
+    @GetMapping("/tencent/v2/impression")
+    public Object tencentImpressionV2(HttpServletRequest request) throws Exception {
+        impressionSemaphore.acquire();
+        try {
+            return doImpression(request, true);
+        } finally {
+            impressionSemaphore.release();
+        }
+    }
+
+    private Object doImpression(HttpServletRequest request, boolean isNewLink) throws Exception {
         Map<String, String> params = flatQueryParams(request);
+        markTrackingVersion(params, isNewLink);
         String traceId = mediaTraceId(params);
         if (traceId == null || traceId.isBlank()) {
             log.warn("[腾讯][曝光] 缺少 traceId, params={}", params.keySet());
@@ -129,7 +140,18 @@ public class TrackingController {
             throws Exception {
         clickSemaphore.acquire();
         try {
-            return doClick(request, response);
+            return doClick(request, response, false);
+        } finally {
+            clickSemaphore.release();
+        }
+    }
+
+    @GetMapping("/tencent/v2/click")
+    public Object tencentClickV2(HttpServletRequest request, HttpServletResponse response)
+            throws Exception {
+        clickSemaphore.acquire();
+        try {
+            return doClick(request, response, true);
         } finally {
             clickSemaphore.release();
         }
@@ -137,11 +159,17 @@ public class TrackingController {
 
     private Object doClick(HttpServletRequest request, HttpServletResponse response)
             throws Exception {
+        return doClick(request, response, false);
+    }
+
+    private Object doClick(HttpServletRequest request, HttpServletResponse response, boolean isNewLink)
+            throws Exception {
         if (tencentPlacement == null) {
             throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "tencent placement not configured");
         }
 
         Map<String, String> params = flatQueryParams(request);
+        markTrackingVersion(params, isNewLink);
         String traceId = mediaTraceId(params);
         if (traceId == null || traceId.isBlank()) {
             log.warn("[腾讯][点击] 缺少 traceId, params={}", params.keySet());
@@ -197,6 +225,11 @@ public class TrackingController {
         return Map.of("qk", nvl(record.getQk()), "traceId", traceId);
     }
 
+    private static void markTrackingVersion(Map<String, String> params, boolean isNewLink) {
+        params.put(BidRecord.NEW_LINK_FLAG_KEY, Boolean.toString(isNewLink));
+        params.put(BidRecord.TRACKING_VERSION_KEY, isNewLink ? "v2" : "v1");
+    }
+
     // ─── 私有工具方法 ────────────────────────────────────────────────────────
 
     private BidRecord bidRecordFromMediaResponse(NormalizedBidResponse response,

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

@@ -117,6 +117,9 @@ public class TiDBColdStore {
                 attempt INT NOT NULL DEFAULT 1,
                 response_body TEXT,
                 error_message TEXT,
+                request_body TEXT,
+                dispatch_status VARCHAR(32) NOT NULL DEFAULT 'SENT',
+                tracking_version VARCHAR(16) NOT NULL DEFAULT 'v1',
                 created_at TIMESTAMP(3) NOT NULL,
                 KEY idx_tencent_media_callbacks_dedupe_ok_created_at (dedupe_key, ok, created_at),
                 KEY idx_tencent_media_callbacks_qk_created_at (qk, created_at),
@@ -158,6 +161,21 @@ public class TiDBColdStore {
                 stmt.execute("ALTER TABLE tencent_baidu_conversions ADD KEY idx_tencent_baidu_conversions_tag_id (tag_id)");
             } catch (Exception ignored) {}
             try {
+                stmt.execute("ALTER TABLE tencent_media_callbacks ADD COLUMN request_body TEXT AFTER error_message");
+            } catch (Exception ignored) {}
+            try {
+                stmt.execute("ALTER TABLE tencent_media_callbacks ADD COLUMN dispatch_status VARCHAR(32) NOT NULL DEFAULT 'SENT' AFTER request_body");
+            } catch (Exception ignored) {}
+            try {
+                stmt.execute("UPDATE tencent_media_callbacks SET dispatch_status = 'SENT' WHERE dispatch_status IS NULL OR dispatch_status = ''");
+            } catch (Exception ignored) {}
+            try {
+                stmt.execute("ALTER TABLE tencent_media_callbacks ADD COLUMN tracking_version VARCHAR(16) NOT NULL DEFAULT 'v1' AFTER dispatch_status");
+            } catch (Exception ignored) {}
+            try {
+                stmt.execute("UPDATE tencent_media_callbacks SET tracking_version = 'v1' WHERE tracking_version IS NULL OR tracking_version = ''");
+            } catch (Exception ignored) {}
+            try {
                 stmt.execute("ALTER TABLE tencent_tag_event ADD COLUMN baidu_act INT NULL COMMENT '百度转化行为' AFTER tag_id");
             } catch (Exception ignored) {}
             try {
@@ -275,6 +293,12 @@ public class TiDBColdStore {
     public void saveMediaCallback(MediaCallbackRecord record) {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         int attempt = record.getAttempt() <= 0 ? 1 : record.getAttempt();
+        if (record.getDispatchStatus() == null || record.getDispatchStatus().isBlank()) {
+            record.setDispatchStatus(MediaCallbackRecord.DISPATCH_STATUS_SENT);
+        }
+        if (record.getTrackingVersion() == null || record.getTrackingVersion().isBlank()) {
+            record.setTrackingVersion("v1");
+        }
         try (SqlSession session = sqlSessionFactory.openSession(true)) {
             MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
             mapper.insert(
@@ -282,7 +306,8 @@ public class TiDBColdStore {
                 record.getCallbackUrl(), record.getEventType(), record.getEventTimeMs(),
                 record.getPurchase(), record.getStatus(), record.isOk() ? 1 : 0,
                 attempt, record.getResponseBody(), record.getErrorMessage(),
-                record.getRequestBody(), Timestamp.from(record.getCreatedAt())
+                record.getRequestBody(), record.getDispatchStatus(), record.getTrackingVersion(),
+                Timestamp.from(record.getCreatedAt())
             );
         }
     }
@@ -292,6 +317,12 @@ public class TiDBColdStore {
         for (MediaCallbackRecord record : records) {
             if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
             if (record.getAttempt() <= 0) record.setAttempt(1);
+            if (record.getDispatchStatus() == null || record.getDispatchStatus().isBlank()) {
+                record.setDispatchStatus(MediaCallbackRecord.DISPATCH_STATUS_SENT);
+            }
+            if (record.getTrackingVersion() == null || record.getTrackingVersion().isBlank()) {
+                record.setTrackingVersion("v1");
+            }
         }
         try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH, false)) {
             MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);

src/main/java/com/adx/tencent/storage/mapper/MediaCallbackMapper.java

@@ -21,6 +21,8 @@ public interface MediaCallbackMapper {
                 @Param("responseBody") String responseBody,
                 @Param("errorMessage") String errorMessage,
                 @Param("requestBody") String requestBody,
+                @Param("dispatchStatus") String dispatchStatus,
+                @Param("trackingVersion") String trackingVersion,
                 @Param("createdAt") java.sql.Timestamp createdAt);
 
     int countSuccessfulByDedupeKey(@Param("dedupeKey") String dedupeKey);

src/main/java/com/adx/tencent/storage/model/BidRecord.java

@@ -9,6 +9,9 @@ import java.util.Map;
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public class BidRecord {
+    public static final String NEW_LINK_FLAG_KEY = "is_new_link";
+    public static final String TRACKING_VERSION_KEY = "tracking_version";
+
     @JsonProperty("qk")           private String qk;
     @JsonProperty("media")        private String media;
     @JsonProperty("mediaTraceId") private String mediaTraceId;
@@ -135,7 +138,8 @@ public class BidRecord {
                 "oaid", "oaid_md5",
                 "android_id", "android_id_md5",
                 "hash_oaid", "hash_android_id",
-                "callback", "click_id", "account_id"
+                "callback", "click_id", "account_id",
+                NEW_LINK_FLAG_KEY, TRACKING_VERSION_KEY
         );
         Map<String, String> slim = new LinkedHashMap<>();
         for (String key : deviceKeys) {

src/main/java/com/adx/tencent/storage/model/MediaCallbackRecord.java

@@ -4,6 +4,9 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 import java.time.Instant;
 
 public class MediaCallbackRecord {
+    public static final String DISPATCH_STATUS_SENT = "SENT";
+    public static final String DISPATCH_STATUS_DEDUCTED = "DEDUCTED";
+
     @JsonProperty("dedupeKey")    private String dedupeKey;
     @JsonProperty("media")        private String media;
     @JsonProperty("qk")           private String qk;
@@ -17,6 +20,8 @@ public class MediaCallbackRecord {
     @JsonProperty("responseBody") private String responseBody;
     @JsonProperty("errorMessage") private String errorMessage;
     @JsonProperty("requestBody")  private String requestBody;
+    @JsonProperty("dispatchStatus") private String dispatchStatus;
+    @JsonProperty("trackingVersion") private String trackingVersion;
     @JsonProperty("createdAt")    private Instant createdAt;
 
     public String getDedupeKey() { return dedupeKey; }
@@ -45,6 +50,10 @@ public class MediaCallbackRecord {
     public void setErrorMessage(String v) { this.errorMessage = v; }
     public String getRequestBody() { return requestBody; }
     public void setRequestBody(String v) { this.requestBody = v; }
+    public String getDispatchStatus() { return dispatchStatus; }
+    public void setDispatchStatus(String v) { this.dispatchStatus = v; }
+    public String getTrackingVersion() { return trackingVersion; }
+    public void setTrackingVersion(String v) { this.trackingVersion = v; }
     public Instant getCreatedAt() { return createdAt; }
     public void setCreatedAt(Instant v) { this.createdAt = v; }
     public java.sql.Timestamp getCreatedAtTs() { return createdAt == null ? null : java.sql.Timestamp.from(createdAt); }

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

@@ -81,6 +81,8 @@ public class TencentClient {
         record.setEventTimeMs(eventTimeMs);
         record.setPurchase(payment.getPayment());
         record.setAttempt(1);
+        record.setDispatchStatus(MediaCallbackRecord.DISPATCH_STATUS_SENT);
+        record.setTrackingVersion(resolveTrackingVersion(bid));
         record.setCreatedAt(Instant.now());
 
         // 取 callback URL
@@ -126,6 +128,8 @@ public class TencentClient {
             throws IOException, InterruptedException {
         MediaCallbackRecord retry = copyRecord(original);
         retry.setAttempt(original.getAttempt() + 1);
+        retry.setDispatchStatus(original.getDispatchStatus());
+        retry.setTrackingVersion(original.getTrackingVersion());
         retry.setCreatedAt(Instant.now());
 
         String callbackUrl = original.getCallbackUrl();
@@ -270,6 +274,14 @@ public class TencentClient {
         return action;
     }
 
+    private static String resolveTrackingVersion(BidRecord bid) {
+        if (bid == null || bid.getMediaParams() == null) {
+            return "v1";
+        }
+        String version = bid.getMediaParams().get(BidRecord.TRACKING_VERSION_KEY);
+        return (version == null || version.isBlank()) ? "v1" : version;
+    }
+
     private String buildOuterActionId(BidRecord bid, PaymentInfo payment, String actionType) {
         String source = String.join("|",
                 safe(payment.getQk()),

src/main/resources/application-dev.yml

@@ -30,6 +30,7 @@ adx:
   tencent-access-token-redis-key: "adq:access_token:78578478"
   tencent-android-user-action-set-id: 1222435591
   tencent-ios-user-action-set-id: 1222435768
+  tencent-new-link-deduction-rate: 20
   tencent-act-map:
     "1": ACTIVATE_APP      # 激活
     "2": PURCHASE          # 付费

src/main/resources/application-prod.yml

@@ -31,6 +31,7 @@ adx:
   tencent-api-base-url: "https://api.e.qq.com/v3.0"
   tencent-android-user-action-set-id: 1222435591
   tencent-ios-user-action-set-id: 1222435768
+  tencent-new-link-deduction-rate: 20
   tencent-act-map:
     "1": ACTIVATE_APP          # 激活
     "2": PURCHASE              # 付费

src/main/resources/application-test.yml

@@ -30,6 +30,7 @@ adx:
   tencent-access-token: "YOUR_TEST_ACCESS_TOKEN"
   tencent-android-user-action-set-id: 0
   tencent-ios-user-action-set-id: 0
+  tencent-new-link-deduction-rate: 20
   tencent-act-map:
     "1": ACTIVATE_APP
     "2": PURCHASE

src/main/resources/application.yml

@@ -25,6 +25,7 @@ adx:
   baidu-android-tag-ids: jlt52,jlt53,jlt54,jlt55,jlt56,jlt57,jlt58,jlt59,jlt60,jlt61,jlt62,jlt63,jlt64,jlt65,jlt66,jlt67,jlt68,jlt69,jlt70,jlt71,jlt72,jlt73,jlt74,jlt75,jlt76,jlt77,jlt78,jlt79,jlt80,jlt81,jlt82,jlt83,jlt84,jlt85,jlt86,jlt87,jlt88,jlt89,jlt90,jlt91,jlt92,jlt93,jlt94,jlt95,jlt96,jlt97,jlt98,jlt99,jlt100,jlt101
   baidu-ios-app-id: efff0562
   baidu-ios-tag-ids: jlt01,jlt03,jlt04,jlt05,jlt06,jlt07,jlt08,jlt09,jlt10,jlt11,jlt12,jlt13,jlt14,jlt15,jlt16,jlt17,jlt18,jlt19,jlt20,jlt21,jlt22,jlt23,jlt24,jlt25,jlt26,jlt27,jlt28,jlt29,jlt30,jlt31,jlt32,jlt33,jlt34,jlt35,jlt36,jlt37,jlt38,jlt39,jlt40,jlt41,jlt42,jlt43,jlt44,jlt45,jlt46,jlt47,jlt48,jlt49,jlt50,jlt51
+  tencent-new-link-deduction-rate: 20
 
 baidu:
   clue:

src/main/resources/mapper/MediaCallbackMapper.xml

@@ -16,28 +16,34 @@
         <result property="responseBody" column="response_body"/>
         <result property="errorMessage" column="error_message"/>
         <result property="requestBody" column="request_body"/>
+        <result property="dispatchStatus" column="dispatch_status"/>
+        <result property="trackingVersion" column="tracking_version"/>
         <result property="createdAt" column="created_at"/>
     </resultMap>
 
     <insert id="insert">
         INSERT INTO tencent_media_callbacks (
             dedupe_key, media, qk, callback_url, event_type, event_time_ms, purchase,
-            status, ok, attempt, response_body, error_message, request_body, created_at
+            status, ok, attempt, response_body, error_message, request_body, dispatch_status, tracking_version, created_at
         ) VALUES (
             #{dedupeKey}, #{media}, #{qk}, #{callbackUrl}, #{eventType}, #{eventTimeMs},
             #{purchase}, #{status}, #{ok}, #{attempt}, #{responseBody}, #{errorMessage},
-            #{requestBody}, #{createdAt}
+            #{requestBody}, #{dispatchStatus}, #{trackingVersion}, #{createdAt}
         )
     </insert>
 
     <select id="countSuccessfulByDedupeKey" resultType="int">
-        SELECT COUNT(1) FROM tencent_media_callbacks WHERE dedupe_key = #{dedupeKey} AND ok = 1 LIMIT 1
+        SELECT COUNT(1)
+        FROM tencent_media_callbacks
+        WHERE dedupe_key = #{dedupeKey}
+          AND (ok = 1 OR dispatch_status = 'DEDUCTED')
+        LIMIT 1
     </select>
 
     <select id="countSuccessfulForEvent" resultType="int">
         SELECT COUNT(1)
         FROM tencent_media_callbacks
-        WHERE ok = 1
+        WHERE (ok = 1 OR dispatch_status = 'DEDUCTED')
           AND (
             dedupe_key = #{dedupeKey}
             OR (dedupe_key = #{legacyDedupeKey} AND event_type = #{eventType})
@@ -48,7 +54,7 @@
     <select id="selectSuccessfulDedupeKeys" resultType="string">
         SELECT DISTINCT dedupe_key
         FROM tencent_media_callbacks
-        WHERE ok = 1
+        WHERE (ok = 1 OR dispatch_status = 'DEDUCTED')
           AND dedupe_key IN
           <foreach collection="dedupeKeys" item="dedupeKey" open="(" separator="," close=")">
               #{dedupeKey}
@@ -58,7 +64,7 @@
     <select id="selectSuccessfulKeysWithLegacy" resultType="string">
         SELECT DISTINCT dedupe_key
         FROM tencent_media_callbacks
-        WHERE ok = 1
+        WHERE (ok = 1 OR dispatch_status = 'DEDUCTED')
           AND dedupe_key IN
           <foreach collection="dedupeKeys" item="dedupeKey" open="(" separator="," close=")">
               #{dedupeKey}
@@ -66,7 +72,7 @@
         UNION
         SELECT DISTINCT dedupe_key
         FROM tencent_media_callbacks
-        WHERE ok = 1
+        WHERE (ok = 1 OR dispatch_status = 'DEDUCTED')
           AND event_type IN
           <foreach collection="eventTypes" item="eventType" open="(" separator="," close=")">
               #{eventType}
@@ -80,13 +86,13 @@
     <insert id="batchInsert">
         INSERT INTO tencent_media_callbacks (
             dedupe_key, media, qk, callback_url, event_type, event_time_ms, purchase,
-            status, ok, attempt, response_body, error_message, request_body, created_at
+            status, ok, attempt, response_body, error_message, request_body, dispatch_status, tracking_version, created_at
         ) VALUES
         <foreach collection="records" item="r" separator=",">
             (
                 #{r.dedupeKey}, #{r.media}, #{r.qk}, #{r.callbackUrl}, #{r.eventType}, #{r.eventTimeMs},
                 #{r.purchase}, #{r.status}, #{r.ok}, #{r.attempt}, #{r.responseBody}, #{r.errorMessage},
-                #{r.requestBody}, #{r.createdAtTs}
+                #{r.requestBody}, #{r.dispatchStatus}, #{r.trackingVersion}, #{r.createdAtTs}
             )
         </foreach>
     </insert>
@@ -94,7 +100,7 @@
     <select id="selectPendingCallbacks" resultMap="mediaCallbackResultMap">
         SELECT m.dedupe_key, m.media, m.qk, m.callback_url, m.event_type, m.event_time_ms,
                m.purchase, m.status, m.ok AS ok_val, m.attempt, m.response_body, m.error_message,
-               m.request_body, m.created_at
+               m.request_body, m.dispatch_status, m.tracking_version, m.created_at
         FROM tencent_media_callbacks m
         JOIN (
             SELECT dedupe_key, MAX(id) AS id
@@ -103,9 +109,11 @@
             GROUP BY dedupe_key
         ) latest ON latest.id = m.id
         WHERE m.media = #{media} AND m.ok = 0
+          AND COALESCE(m.dispatch_status, 'SENT') != 'DEDUCTED'
           AND NOT EXISTS (
             SELECT 1 FROM tencent_media_callbacks ok_cb
-            WHERE ok_cb.dedupe_key = m.dedupe_key AND ok_cb.ok = 1
+            WHERE ok_cb.dedupe_key = m.dedupe_key
+              AND (ok_cb.ok = 1 OR ok_cb.dispatch_status = 'DEDUCTED')
           )
         ORDER BY m.created_at ASC
         LIMIT #{limit}

src/main/resources/schema.sql

@@ -85,6 +85,9 @@ CREATE TABLE IF NOT EXISTS tencent_media_callbacks (
     attempt INT NOT NULL DEFAULT 1 COMMENT '尝试次数',
     response_body TEXT COMMENT '响应体(截断)',
     error_message TEXT COMMENT '错误信息',
+    request_body TEXT COMMENT '请求体',
+    dispatch_status VARCHAR(32) NOT NULL DEFAULT 'SENT' COMMENT '回传处理状态(SENT/DEDUCTED)',
+    tracking_version VARCHAR(16) NOT NULL DEFAULT 'v1' COMMENT '监测链接版本(v1老链接/v2新链接)',
     created_at TIMESTAMP(3) NOT NULL COMMENT '创建时间',
     KEY idx_tencent_media_callbacks_dedupe_ok_created_at (dedupe_key, ok, created_at),
     KEY idx_tencent_media_callbacks_qk_created_at (qk, created_at),
@@ -119,6 +122,11 @@ CREATE TABLE IF NOT EXISTS tencent_tag_event (
 -- ALTER TABLE tencent_baidu_conversions ADD COLUMN tag_id VARCHAR(128) AFTER media;
 -- ALTER TABLE tencent_baidu_conversions ADD KEY idx_tencent_baidu_conversions_tag_id (tag_id);
 
+-- 如果旧表没有 request_body / dispatch_status 字段
+-- ALTER TABLE tencent_media_callbacks ADD COLUMN request_body TEXT AFTER error_message;
+-- ALTER TABLE tencent_media_callbacks ADD COLUMN dispatch_status VARCHAR(32) NOT NULL DEFAULT 'SENT' AFTER request_body;
+-- ALTER TABLE tencent_media_callbacks ADD COLUMN tracking_version VARCHAR(16) NOT NULL DEFAULT 'v1' AFTER dispatch_status;
+
 -- 如果旧表没有 event_type 字段
 -- ALTER TABLE tencent_ad_bid_events ADD COLUMN event_type VARCHAR(16) COMMENT '事件类型(impression/click)' AFTER account_id;