1

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

@@ -56,7 +56,7 @@ public class AppProperties {
     private int redisDb = 0;
     private String redisStream = "adx:tencent:events";
     private long redisStreamMaxLen = 50000;
-    private Duration bidTtl = Duration.ofHours(24);
+    private Duration bidTtl = Duration.ofDays(5);
 
     // TiDB
     private String tidbUrl;

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

@@ -14,7 +14,11 @@ import org.slf4j.LoggerFactory;
 
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -94,122 +98,129 @@ public class ConversionSyncService {
     }
 
     private void processPayments(List<PaymentInfo> payments, SyncResult result, SyncMode mode) throws Exception {
+        if (payments == null || payments.isEmpty()) {
+            return;
+        }
+
         ExecutorService executor = mode == SyncMode.BACKFILL ? backfillExecutor : realtimeExecutor;
-        List<CompletableFuture<Void>> futures = new ArrayList<>(payments.size());
+        Set<String> qks = new LinkedHashSet<>();
         for (PaymentInfo payment : payments) {
-            futures.add(CompletableFuture.runAsync(() -> processSinglePayment(payment, result), executor));
+            result.fetched.incrementAndGet();
+            if (payment.getQk() != null && !payment.getQk().isBlank()) {
+                qks.add(payment.getQk());
+            }
         }
-        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
-    }
 
-    private void processSinglePayment(PaymentInfo payment, SyncResult result) {
-        result.fetched.incrementAndGet();
+        Map<String, BidRecord> bidMap = hotStore.findBidsByQks(new ArrayList<>(qks));
+        List<ConversionRecord> conversionRecords = new ArrayList<>();
+        List<CallbackWork> callbackWorks = new ArrayList<>();
 
-        // 用 qk 查找 bid
-        BidRecord bid;
-        try {
-            bid = hotStore.findBidByQk(payment.getQk());
-        } catch (Exception e) {
-            log.warn("[ConversionSync] hot bid lookup failed, fallback to cold store: qk={}, {}", payment.getQk(), e.getMessage());
-            bid = coldStore != null ? coldStore.getBidByQk(payment.getQk()) : null;
-        }
+        for (PaymentInfo payment : payments) {
+            BidRecord bid = bidMap.get(payment.getQk());
+            if (bid == null) {
+                conversionRecords.add(conversionRecord(payment, null, ""));
+                result.skipped.incrementAndGet();
+                continue;
+            }
 
-        if (bid == null) {
-            if (coldStore != null) {
-                try {
-                    coldStore.saveConversion(conversionRecord(payment, null, ""));
-                } catch (Exception e) {
-                    log.error("[ConversionSync] save conversion failed: qk={}, act={}, error={}",
-                            payment.getQk(), payment.getAct(), e.getMessage(), e);
-                }
+            if (!"tencent".equals(bid.getMedia())) {
+                conversionRecords.add(conversionRecord(payment, bid, bid.getMedia()));
+                result.skipped.incrementAndGet();
+                continue;
             }
-            result.skipped.incrementAndGet();
-            return;
-        }
 
+            result.matched.incrementAndGet();
+            log.info("[ConversionSync] 匹配到腾讯转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
+                    payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());
+            conversionRecords.add(conversionRecord(payment, bid, bid.getMedia()));
 
-        // bid 不是腾讯，或没有设备标识
-        if (!"tencent".equals(bid.getMedia())) {
-            if (coldStore != null) {
-                try {
-                    coldStore.saveConversion(conversionRecord(payment, bid, bid.getMedia()));
-                } catch (Exception e) {
-                    log.error("[ConversionSync] save conversion failed: qk={}, act={}, error={}",
-                            payment.getQk(), payment.getAct(), e.getMessage(), e);
-                }
+            String actionType = null;
+            if (tagEventResolver != null) {
+                actionType = tagEventResolver.resolveActionType(bid.getTagId(), payment.getAct());
             }
-            result.skipped.incrementAndGet();
-            return;
+            if (actionType == null || actionType.isBlank()) {
+                actionType = tencentClient.conversionActionType(payment.getAct());
+                log.warn("[ConversionSync] 广告位回传方式未配置, 回退act-map | qk={} | tagId={} | act={} | actionType={}",
+                        payment.getQk(), bid.getTagId(), payment.getAct(), actionType);
+            }
+
+            callbackWorks.add(new CallbackWork(payment, bid, actionType, callbackDedupeKey(payment, bid, actionType)));
         }
 
-        result.matched.incrementAndGet();
-        log.info("[ConversionSync] 匹配到腾讯转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
-                payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());
-        if (coldStore != null) {
+        if (coldStore != null && !conversionRecords.isEmpty()) {
             try {
-                coldStore.saveConversion(conversionRecord(payment, bid, bid.getMedia()));
+                coldStore.saveConversions(conversionRecords);
             } catch (Exception e) {
-                log.error("[ConversionSync] save conversion failed: qk={}, act={}, error={}",
-                        payment.getQk(), payment.getAct(), e.getMessage(), e);
+                log.error("[ConversionSync] batch save conversions failed: size={}, {}", conversionRecords.size(), e.getMessage(), e);
             }
         }
 
-        String actionType = null;
-        if (tagEventResolver != null) {
-            actionType = tagEventResolver.resolveActionType(bid.getTagId(), payment.getAct());
-        }
-        if (actionType == null || actionType.isBlank()) {
-            actionType = tencentClient.conversionActionType(payment.getAct());
-            log.warn("[ConversionSync] 广告位回传方式未配置, 回退act-map | qk={} | tagId={} | act={} | actionType={}",
-                    payment.getQk(), bid.getTagId(), payment.getAct(), actionType);
+        if (callbackWorks.isEmpty()) {
+            return;
         }
 
-        // 幂等检查
-        String callbackKey = callbackDedupeKey(payment, bid, actionType);
+        Set<String> existing = new HashSet<>();
         if (coldStore != null) {
             try {
-                String legacyCallbackKey = legacyCallbackDedupeKey(payment, bid.getMedia());
-                boolean exists = coldStore.successfulCallbackExistsForEvent(
-                        callbackKey, legacyCallbackKey, payment.getAct());
-                if (exists) {
-                    result.alreadySent.incrementAndGet();
-                    return;
+                List<String> callbackKeys = new ArrayList<>(callbackWorks.size());
+                List<String> legacyKeys = new ArrayList<>(callbackWorks.size());
+                List<Integer> eventTypes = new ArrayList<>(callbackWorks.size());
+                for (CallbackWork work : callbackWorks) {
+                    callbackKeys.add(work.callbackKey);
+                    legacyKeys.add(legacyCallbackDedupeKey(work.payment, work.bid.getMedia()));
+                    eventTypes.add(work.payment.getAct());
                 }
+                existing.addAll(coldStore.successfulCallbackKeysWithLegacy(callbackKeys, legacyKeys, eventTypes));
             } catch (Exception e) {
-                log.error("[ConversionSync] dedup check error: key={}, {}", callbackKey, e.getMessage());
+                log.error("[ConversionSync] batch dedup check error: size={}, {}", callbackWorks.size(), e.getMessage(), e);
             }
         }
 
-        String platform = resolvePlatform(bid);
-        MediaCallbackRecord callbackRecord = tencentClient.sendConversionEvent(bid, payment, platform, actionType);
-        callbackRecord.setDedupeKey(callbackKey);
-
-        if (callbackRecord.getStatus() == 0 && callbackRecord.getErrorMessage() != null) {
-            // 网络级错误
-            result.failed.incrementAndGet();
-            if (coldStore != null) {
-                try {
-                    coldStore.saveMediaCallback(callbackRecord);
-                } catch (Exception e) {
-                    log.error("[ConversionSync] 保存callback记录失败: dedupeKey={}, error={}",
-                            callbackKey, e.getMessage());
-                }
+        List<CallbackWork> toSend = new ArrayList<>(callbackWorks.size());
+        for (CallbackWork work : callbackWorks) {
+            if (existing.contains(work.callbackKey)) {
+                result.alreadySent.incrementAndGet();
+            } else {
+                toSend.add(work);
             }
-            return;
         }
 
-        // HTTP 请求完成
-        result.sent.incrementAndGet();
-        if (coldStore != null) {
+        List<CompletableFuture<MediaCallbackRecord>> futures = new ArrayList<>(toSend.size());
+        for (CallbackWork work : toSend) {
+            futures.add(CompletableFuture.supplyAsync(() -> sendConversionEvent(work), executor));
+        }
+
+        List<MediaCallbackRecord> callbackRecords = new ArrayList<>(toSend.size());
+        for (CompletableFuture<MediaCallbackRecord> future : futures) {
+            MediaCallbackRecord callbackRecord = future.join();
+            if (callbackRecord == null) {
+                continue;
+            }
+            if (callbackRecord.getStatus() == 0 && callbackRecord.getErrorMessage() != null) {
+                result.failed.incrementAndGet();
+            } else {
+                result.sent.incrementAndGet();
+            }
+            callbackRecords.add(callbackRecord);
+        }
+
+        if (coldStore != null && !callbackRecords.isEmpty()) {
             try {
-                coldStore.saveMediaCallback(callbackRecord);
+                coldStore.saveMediaCallbacks(callbackRecords);
             } catch (Exception e) {
-                log.error("[ConversionSync] 保存callback记录失败: dedupeKey={}, error={}",
-                        callbackKey, e.getMessage());
+                log.error("[ConversionSync] batch save callbacks failed: size={}, {}", callbackRecords.size(), e.getMessage(), e);
             }
         }
     }
 
+    private MediaCallbackRecord sendConversionEvent(CallbackWork work) {
+        String platform = resolvePlatform(work.bid);
+        MediaCallbackRecord callbackRecord = tencentClient.sendConversionEvent(
+                work.bid, work.payment, platform, work.actionType);
+        callbackRecord.setDedupeKey(work.callbackKey);
+        return callbackRecord;
+    }
+
     // ─── 辅助方法 ─────────────────────────────────────────────────────────────
 
     /**
@@ -270,6 +281,20 @@ public class ConversionSyncService {
                 media, p.getQk(), p.getDate(), p.getDeviceId(), p.getAct(), p.getAct());
     }
 
+    private static class CallbackWork {
+        private final PaymentInfo payment;
+        private final BidRecord bid;
+        private final String actionType;
+        private final String callbackKey;
+
+        private CallbackWork(PaymentInfo payment, BidRecord bid, String actionType, String callbackKey) {
+            this.payment = payment;
+            this.bid = bid;
+            this.actionType = actionType;
+            this.callbackKey = callbackKey;
+        }
+    }
+
     // ─── Result DTO ─────────────────────────────────────────────────────────
 
     public static class SyncResult {

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

@@ -162,6 +162,38 @@ public class RedisHotStore {
         return null;
     }
 
+    public Map<String, BidRecord> findBidsByQks(List<String> qks) {
+        if (qks == null || qks.isEmpty()) return Collections.emptyMap();
+        Map<String, BidRecord> result = new HashMap<>();
+        List<String> keys = new ArrayList<>(qks.size());
+        Map<String, String> keyToQk = new HashMap<>();
+        for (String qk : qks) {
+            if (qk == null || qk.isBlank() || keyToQk.containsValue(qk)) continue;
+            String key = bidKey(qk);
+            keys.add(key);
+            keyToQk.put(key, qk);
+        }
+        List<String> payloads = redis.opsForValue().multiGet(keys);
+        Set<String> missing = new LinkedHashSet<>();
+        for (int i = 0; i < keys.size(); i++) {
+            String payload = payloads != null ? payloads.get(i) : null;
+            String qk = keyToQk.get(keys.get(i));
+            if (payload != null) {
+                result.put(qk, parseBidRecord(payload));
+            } else if (qk != null) {
+                missing.add(qk);
+            }
+        }
+        if (!missing.isEmpty() && coldStore != null) {
+            for (BidRecord bid : coldStore.getBidsByQks(new ArrayList<>(missing))) {
+                if (bid != null && bid.getQk() != null) {
+                    result.put(bid.getQk(), bid);
+                }
+            }
+        }
+        return result;
+    }
+
     public BidRecord findBidByMediaTrace(String media, String traceId) {
         String payload = redis.opsForValue().get(mediaTraceKey(media, traceId));
         if (payload == null) return null;

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

@@ -3,6 +3,7 @@ package com.adx.tencent.storage;
 import com.adx.tencent.storage.mapper.*;
 import com.adx.tencent.storage.model.*;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.ibatis.session.ExecutorType;
 import org.apache.ibatis.session.SqlSession;
 import org.apache.ibatis.session.SqlSessionFactory;
 
@@ -210,6 +211,14 @@ public class TiDBColdStore {
         }
     }
 
+    public List<BidRecord> getBidsByQks(List<String> qks) {
+        if (qks == null || qks.isEmpty()) return List.of();
+        try (SqlSession session = sqlSessionFactory.openSession(true)) {
+            BidEventMapper mapper = session.getMapper(BidEventMapper.class);
+            return mapper.selectByQks(qks);
+        }
+    }
+
     // ─── SaveTracking ─────────────────────────────────────────────────────────
 
     public void saveTracking(TrackingRecord record) {
@@ -240,6 +249,17 @@ public class TiDBColdStore {
         }
     }
 
+    public void saveConversions(List<ConversionRecord> records) {
+        if (records == null || records.isEmpty()) return;
+        for (ConversionRecord record : records) {
+            if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
+        }
+        try (SqlSession session = sqlSessionFactory.openSession(true)) {
+            ConversionMapper mapper = session.getMapper(ConversionMapper.class);
+            mapper.batchInsertOrUpdate(records);
+        }
+    }
+
     // ─── SaveMediaCallback ───────────────────────────────────────────────────
 
     public void saveMediaCallback(MediaCallbackRecord record) {
@@ -257,6 +277,19 @@ public class TiDBColdStore {
         }
     }
 
+    public void saveMediaCallbacks(List<MediaCallbackRecord> records) {
+        if (records == null || records.isEmpty()) return;
+        for (MediaCallbackRecord record : records) {
+            if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
+            if (record.getAttempt() <= 0) record.setAttempt(1);
+        }
+        try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH, false)) {
+            MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
+            mapper.batchInsert(records);
+            session.commit();
+        }
+    }
+
     // ─── 查询：成功回调是否存在 ────────────────────────────────────────────────
 
     public boolean successfulCallbackExists(String dedupeKey) {
@@ -277,6 +310,24 @@ public class TiDBColdStore {
         }
     }
 
+    public List<String> successfulCallbackKeys(List<String> dedupeKeys) {
+        if (dedupeKeys == null || dedupeKeys.isEmpty()) return List.of();
+        try (SqlSession session = sqlSessionFactory.openSession(true)) {
+            MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
+            return mapper.selectSuccessfulDedupeKeys(dedupeKeys);
+        }
+    }
+
+    public List<String> successfulCallbackKeysWithLegacy(List<String> dedupeKeys,
+                                                         List<String> legacyDedupeKeys,
+                                                         List<Integer> eventTypes) {
+        if (dedupeKeys == null || dedupeKeys.isEmpty()) return List.of();
+        try (SqlSession session = sqlSessionFactory.openSession(true)) {
+            MediaCallbackMapper mapper = session.getMapper(MediaCallbackMapper.class);
+            return mapper.selectSuccessfulKeysWithLegacy(dedupeKeys, legacyDedupeKeys, eventTypes);
+        }
+    }
+
     // ─── 查询：待重试的回调 ────────────────────────────────────────────────────
 
     public List<MediaCallbackRecord> pendingMediaCallbacks(String media, int limit) {

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

@@ -1,11 +1,16 @@
 package com.adx.tencent.storage.mapper;
 
+import com.adx.tencent.storage.model.BidRecord;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
 
+import java.util.List;
+
 @Mapper
 public interface BidEventMapper {
-    com.adx.tencent.storage.model.BidRecord selectByQk(@Param("qk") String qk);
+    BidRecord selectByQk(@Param("qk") String qk);
+
+    List<BidRecord> selectByQks(@Param("qks") List<String> qks);
 
     void insertOrUpdate(@Param("qk") String qk,
                         @Param("media") String media,

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

@@ -1,8 +1,11 @@
 package com.adx.tencent.storage.mapper;
 
+import com.adx.tencent.storage.model.ConversionRecord;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
 
+import java.util.List;
+
 @Mapper
 public interface ConversionMapper {
     void insertOrUpdate(@Param("dedupeKey") String dedupeKey,
@@ -20,4 +23,6 @@ public interface ConversionMapper {
                         @Param("tu") String tu,
                         @Param("clkTime") String clkTime,
                         @Param("createdAt") java.sql.Timestamp createdAt);
+
+    void batchInsertOrUpdate(@Param("records") List<ConversionRecord> records);
 }

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

@@ -29,6 +29,14 @@ public interface MediaCallbackMapper {
                                 @Param("legacyDedupeKey") String legacyDedupeKey,
                                 @Param("eventType") int eventType);
 
+    List<String> selectSuccessfulDedupeKeys(@Param("dedupeKeys") List<String> dedupeKeys);
+
+    List<String> selectSuccessfulKeysWithLegacy(@Param("dedupeKeys") List<String> dedupeKeys,
+                                                @Param("legacyDedupeKeys") List<String> legacyDedupeKeys,
+                                                @Param("eventTypes") List<Integer> eventTypes);
+
+    void batchInsert(@Param("records") List<MediaCallbackRecord> records);
+
     List<MediaCallbackRecord> selectPendingCallbacks(@Param("media") String media,
                                                      @Param("limit") int limit);
 }

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

@@ -50,4 +50,5 @@ public class ConversionRecord {
     public void setClkTime(String v) { this.clkTime = 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/storage/model/MediaCallbackRecord.java

@@ -47,4 +47,5 @@ public class MediaCallbackRecord {
     public void setRequestBody(String v) { this.requestBody = 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/resources/mapper/BidEventMapper.xml

@@ -34,6 +34,17 @@
         LIMIT 1
     </select>
 
+    <select id="selectByQks" resultMap="bidRecordResultMap">
+        SELECT qk, media, media_trace_id, platform, ad_id, account_id, event_type,
+               req_id, bid_id, creative_id, imp_id, tag_id, price, show_urls,
+               click_urls, landing_page, app_store_link, package_name, media_params, created_at
+        FROM tencent_ad_bid_events
+        WHERE qk IN
+        <foreach collection="qks" item="qk" open="(" separator="," close=")">
+            #{qk}
+        </foreach>
+    </select>
+
     <insert id="insertOrUpdate">
         INSERT INTO tencent_ad_bid_events (
             qk, media, media_trace_id, platform, ad_id, account_id, event_type,

src/main/resources/mapper/ConversionMapper.xml

@@ -24,4 +24,31 @@
             gmv = VALUES(gmv),
             clk_time = VALUES(clk_time)
     </insert>
+
+    <insert id="batchInsertOrUpdate">
+        INSERT INTO tencent_baidu_conversions (
+            dedupe_key, qk, media, tag_id, date_value, appsid, customer_name, device_id, conv,
+            payment, gmv, act, tu, clk_time, created_at
+        ) VALUES
+        <foreach collection="records" item="r" separator=",">
+            (
+                #{r.dedupeKey}, #{r.qk}, #{r.media}, #{r.tagId}, #{r.date}, #{r.appSid}, #{r.customerName},
+                #{r.deviceId}, #{r.conv}, #{r.payment}, #{r.gmv}, #{r.act}, #{r.tu}, #{r.clkTime},
+                #{r.createdAtTs}
+            )
+        </foreach>
+        ON DUPLICATE KEY UPDATE
+            media = CASE
+                WHEN VALUES(media) IS NOT NULL AND VALUES(media) != '' THEN VALUES(media)
+                ELSE media
+            END,
+            tag_id = CASE
+                WHEN VALUES(tag_id) IS NOT NULL AND VALUES(tag_id) != '' THEN VALUES(tag_id)
+                ELSE tag_id
+            END,
+            conv = VALUES(conv),
+            payment = VALUES(payment),
+            gmv = VALUES(gmv),
+            clk_time = VALUES(clk_time)
+    </insert>
 </mapper>

src/main/resources/mapper/MediaCallbackMapper.xml

@@ -45,6 +45,52 @@
         LIMIT 1
     </select>
 
+    <select id="selectSuccessfulDedupeKeys" resultType="string">
+        SELECT DISTINCT dedupe_key
+        FROM tencent_media_callbacks
+        WHERE ok = 1
+          AND dedupe_key IN
+          <foreach collection="dedupeKeys" item="dedupeKey" open="(" separator="," close=")">
+              #{dedupeKey}
+          </foreach>
+    </select>
+
+    <select id="selectSuccessfulKeysWithLegacy" resultType="string">
+        SELECT DISTINCT dedupe_key
+        FROM tencent_media_callbacks
+        WHERE ok = 1
+          AND dedupe_key IN
+          <foreach collection="dedupeKeys" item="dedupeKey" open="(" separator="," close=")">
+              #{dedupeKey}
+          </foreach>
+        UNION
+        SELECT DISTINCT dedupe_key
+        FROM tencent_media_callbacks
+        WHERE ok = 1
+          AND event_type IN
+          <foreach collection="eventTypes" item="eventType" open="(" separator="," close=")">
+              #{eventType}
+          </foreach>
+          AND dedupe_key IN
+          <foreach collection="legacyDedupeKeys" item="legacyDedupeKey" open="(" separator="," close=")">
+              #{legacyDedupeKey}
+          </foreach>
+    </select>
+
+    <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
+        ) 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}
+            )
+        </foreach>
+    </insert>
+
     <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,