tencent-baidu-tracking/src/main/java/com/adx/tencent/conversionsync/ConversionSyncService.java

package com.adx.tencent.conversionsync;

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;
import com.adx.tencent.honor.service.HonorTagEventResolver;
import com.adx.tencent.honor.store.HonorColdStore;
import com.adx.tencent.honor.store.HonorHotStore;
import com.adx.tencent.kuaishou.client.KuaishouClient;
import com.adx.tencent.kuaishou.model.KuaishouBidRecord;
import com.adx.tencent.kuaishou.model.KuaishouMediaCallbackRecord;
import com.adx.tencent.kuaishou.model.KuaishouTagEventRecord;
import com.adx.tencent.kuaishou.service.KuaishouTagEventResolver;
import com.adx.tencent.kuaishou.store.KuaishouColdStore;
import com.adx.tencent.kuaishou.store.KuaishouHotStore;
import com.adx.tencent.storage.RedisHotStore;
import com.adx.tencent.storage.TiDBColdStore;
import com.adx.tencent.storage.model.BidRecord;
import com.adx.tencent.storage.model.ConversionRecord;
import com.adx.tencent.storage.model.MediaCallbackRecord;
import com.adx.tencent.tagsync.TagEventResolver;
import com.adx.tencent.tencent.TencentClient;
import org.slf4j.Logger;
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.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 统一转化同步：
 * 1. 只从百度拉一次转化
 * 2. 按 qk 匹配腾讯/荣耀/快手 bid
 * 3. 百度转化明细统一落 tencent_baidu_conversions
 * 4. 媒体回传结果分别落各自表
 */
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 final ConversionClient baiduClient;
    private final RedisHotStore hotStore;
    private final HonorHotStore honorHotStore;
    private final KuaishouHotStore kuaishouHotStore;
    private final TencentClient tencentClient;
    private final TiDBColdStore coldStore;
    private final HonorColdStore honorColdStore;
    private final KuaishouColdStore kuaishouColdStore;
    private final HonorClient honorClient;
    private final KuaishouClient kuaishouClient;
    private final TagEventResolver tagEventResolver;
    private final HonorTagEventResolver honorTagEventResolver;
    private final KuaishouTagEventResolver kuaishouTagEventResolver;
    private final ExecutorService realtimeExecutor;
    private final ExecutorService backfillExecutor;

    public enum SyncMode {
        REALTIME,
        BACKFILL
    }

    public ConversionSyncService(ConversionClient baiduClient,
                                 RedisHotStore hotStore,
                                 TencentClient tencentClient,
                                 TiDBColdStore coldStore,
                                 TagEventResolver tagEventResolver,
                                 HonorTagEventResolver honorTagEventResolver,
                                 KuaishouTagEventResolver kuaishouTagEventResolver,
                                 HonorHotStore honorHotStore,
                                 HonorColdStore honorColdStore,
                                 KuaishouHotStore kuaishouHotStore,
                                 KuaishouColdStore kuaishouColdStore,
                                 HonorClient honorClient,
                                 KuaishouClient kuaishouClient,
                                 AppProperties props) {
        this.baiduClient = baiduClient;
        this.hotStore = hotStore;
        this.tencentClient = tencentClient;
        this.coldStore = coldStore;
        this.tagEventResolver = tagEventResolver;
        this.honorTagEventResolver = honorTagEventResolver;
        this.kuaishouTagEventResolver = kuaishouTagEventResolver;
        this.honorHotStore = honorHotStore;
        this.honorColdStore = honorColdStore;
        this.kuaishouHotStore = kuaishouHotStore;
        this.kuaishouColdStore = kuaishouColdStore;
        this.honorClient = honorClient;
        this.kuaishouClient = kuaishouClient;
        this.realtimeExecutor = Executors.newFixedThreadPool(
                REALTIME_CONCURRENCY,
                r -> {
                    Thread t = new Thread(r, "conv-sync-worker");
                    t.setDaemon(true);
                    return t;
                });
        this.backfillExecutor = Executors.newFixedThreadPool(
                BACKFILL_CONCURRENCY,
                r -> {
                    Thread t = new Thread(r, "conv-backfill-worker");
                    t.setDaemon(true);
                    return t;
                });
    }

    public SyncResult syncTencentConversions(ConversionQuery query) throws Exception {
        return syncTencentConversions(query, SyncMode.REALTIME);
    }

    public SyncResult syncTencentConversions(ConversionQuery query, SyncMode mode) throws Exception {
        if (query.getPageSize() <= 0) {
            query.setPageSize(1);
        }

        SyncResult result = new SyncResult();
        int maxPages = 10000;
        for (int page = query.getPageSize(); ; page++) {
            query.setPageSize(page);
            long pageStartNs = System.nanoTime();
            log.info("[ConversionSync] fetching date={}, page={}, mode={}", query.getDate(), page, mode);

            ConversionResponse response = baiduClient.queryPayments(query);
            List<PaymentInfo> data = response.getData();
            int dataSize = data != null ? data.size() : 0;
            log.info("[ConversionSync] date={}, page={} returned {} records, responsePageSize={}, mode={}",
                    query.getDate(), page, dataSize, response.getPageSize(), mode);

            if (data != null) {
                processPayments(data, result, mode);
            }

            long pageCostMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - pageStartNs);
            log.info("[ConversionSync] date={}, page={} processed, cost={}ms, totalFetched={}, matched={}, sent={}, failed={}, skipped={}, alreadySent={}, mode={}",
                    query.getDate(), page, pageCostMs,
                    result.fetched.get(), result.matched.get(), result.sent.get(), result.failed.get(),
                    result.skipped.get(), result.alreadySent.get(), mode);

            if (response.getPageSize() <= page) {
                break;
            }
            if (page >= maxPages) {
                log.warn("[ConversionSync] reached max page limit ({}), stopping", maxPages);
                break;
            }
        }
        return result;
    }

    private void processPayments(List<PaymentInfo> payments, SyncResult result, SyncMode mode) {
        if (payments == null || payments.isEmpty()) {
            return;
        }

        ExecutorService executor = mode == SyncMode.BACKFILL ? backfillExecutor : realtimeExecutor;

        // 1. 批量查这一页转化对应的腾讯/荣耀/快手 bid；Redis miss 时会自动回落到冷库。
        BidLookup lookup = loadBidLookup(payments, result);

        // 2. 统一整理要落库的百度转化明细，以及要回传给媒体的任务。
        List<ConversionRecord> conversionRecords = new ArrayList<>();
        List<CallbackTask> callbackTasks = new ArrayList<>();
        buildRecordsAndTasks(payments, lookup, conversionRecords, callbackTasks, result);

        // 3. 百度转化明细统一落 tencent_baidu_conversions。
        saveConversions(conversionRecords);
        if (callbackTasks.isEmpty()) {
            return;
        }

        // 4. 回传前先做去重，只发送未成功回传过的任务。
        List<CallbackTask> pendingTasks = filterPendingCallbacks(callbackTasks, result);
        if (pendingTasks.isEmpty()) {
            return;
        }

        // 5. 并发回传媒体，并把回传结果分别落到腾讯/荣耀回调表。
        List<CallbackDispatchResult> dispatchResults = dispatchCallbacks(pendingTasks, executor, result);
        saveCallbackResults(dispatchResults);
    }

    // ─── Step 1: 加载 bid ───────────────────────────────────────────────────

    private BidLookup loadBidLookup(List<PaymentInfo> payments, SyncResult result) {
        Set<String> qks = new LinkedHashSet<>();
        for (PaymentInfo payment : payments) {
            result.fetched.incrementAndGet();
            if (payment.getQk() != null && !payment.getQk().isBlank()) {
                qks.add(payment.getQk());
            }
        }

        List<String> qkList = new ArrayList<>(qks);
        Map<String, BidRecord> tencentBids = hotStore.findBidsByQks(qkList);
        backfillTencentAccountIds(qkList, tencentBids);
        Map<String, HonorBidRecord> honorBids = honorHotStore != null
                ? honorHotStore.findBidsByQks(qkList)
                : Map.of();
        Map<String, KuaishouBidRecord> kuaishouBids = kuaishouHotStore != null
                ? kuaishouHotStore.findBidsByQks(qkList)
                : Map.of();
        return new BidLookup(tencentBids, honorBids, kuaishouBids);
    }

    private void backfillTencentAccountIds(List<String> qkList, Map<String, BidRecord> tencentBids) {
        if (coldStore == null || qkList == null || qkList.isEmpty() || tencentBids == null || tencentBids.isEmpty()) {
            return;
        }

        List<String> missingAccountQks = new ArrayList<>();
        for (String qk : qkList) {
            BidRecord bid = tencentBids.get(qk);
            if (bid != null
                    && "tencent".equals(bid.getMedia())
                    && (bid.getAccountId() == null || bid.getAccountId().isBlank())) {
                missingAccountQks.add(qk);
            }
        }

        if (missingAccountQks.isEmpty()) {
            return;
        }

        try {
            for (BidRecord coldBid : coldStore.getBidsByQks(missingAccountQks)) {
                if (coldBid == null || coldBid.getQk() == null) {
                    continue;
                }
                BidRecord hotBid = tencentBids.get(coldBid.getQk());
                if (hotBid == null || hotBid.getAccountId() != null && !hotBid.getAccountId().isBlank()) {
                    continue;
                }
                if (coldBid.getAccountId() == null || coldBid.getAccountId().isBlank()) {
                    continue;
                }
                hotBid.setAccountId(coldBid.getAccountId());
                log.info("[ConversionSync] 腾讯 bid accountId 回落冷库补齐 | qk={} | accountId={}",
                        coldBid.getQk(), coldBid.getAccountId());
            }
        } catch (Exception e) {
            log.warn("[ConversionSync] 腾讯 bid accountId 回落冷库失败 | size={} | error={}",
                    missingAccountQks.size(), e.getMessage(), e);
        }
    }

    // ─── Step 2: 组装转化记录与回传任务 ───────────────────────────────────────

    private void buildRecordsAndTasks(List<PaymentInfo> payments,
                                      BidLookup lookup,
                                      List<ConversionRecord> conversionRecords,
                                      List<CallbackTask> callbackTasks,
                                      SyncResult result) {
        for (PaymentInfo payment : payments) {
            BidRecord tencentBid = lookup.tencentBids.get(payment.getQk());
            HonorBidRecord honorBid = lookup.honorBids.get(payment.getQk());
            KuaishouBidRecord kuaishouBid = lookup.kuaishouBids.get(payment.getQk());

            if (tencentBid == null && honorBid == null && kuaishouBid == null) {
                conversionRecords.add(toConversionRecord(payment, null, ""));
                result.skipped.incrementAndGet();
                continue;
            }

            boolean hasTencentBid = tencentBid != null;
            boolean hasHonorBid = honorBid != null;
            boolean hasKuaishouBid = kuaishouBid != null;
            boolean validTencent = hasTencentBid && "tencent".equals(tencentBid.getMedia());
            boolean validHonor = hasHonorBid && "honor".equals(honorBid.getMedia());
            boolean validKuaishou = hasKuaishouBid && "kuaishou".equals(kuaishouBid.getMedia());

            if (hasTencentBid && !validTencent) {
                log.warn("[ConversionSync] 腾讯 bid media 非法 | qk={} | media={}", payment.getQk(), tencentBid.getMedia());
            }

            if (hasHonorBid && !validHonor) {
                log.warn("[ConversionSync] 荣耀 bid media 非法 | qk={} | media={}", payment.getQk(), honorBid.getMedia());
            }

            if (hasKuaishouBid && !validKuaishou) {
                log.warn("[ConversionSync] 快手 bid media 非法 | qk={} | media={}", payment.getQk(), kuaishouBid.getMedia());
            }

            int matchedMediaCount = (validTencent ? 1 : 0) + (validHonor ? 1 : 0) + (validKuaishou ? 1 : 0);
            if (matchedMediaCount > 1) {
                log.warn("[ConversionSync] qk 同时命中多媒体，跳过分发 | qk={} | act={} | date={}",
                        payment.getQk(), payment.getAct(), payment.getDate());
                result.skipped.incrementAndGet();
                continue;
            }

            if (validTencent) {
                conversionRecords.add(toConversionRecord(payment, tencentBid, "tencent"));
                result.matched.incrementAndGet();
                log.info("[ConversionSync] 匹配到腾讯转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
                        payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());

                CallbackTask tencentTask = buildTencentCallbackTask(payment, tencentBid);
                if (tencentTask != null) {
                    callbackTasks.add(tencentTask);
                }
                continue;
            }

            if (!validHonor) {
                if (validKuaishou) {
                    conversionRecords.add(toKuaishouConversionRecord(payment, kuaishouBid));
                    result.matched.incrementAndGet();
                    log.info("[ConversionSync] 匹配到快手转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
                            payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());

                    CallbackTask kuaishouTask = buildKuaishouCallbackTask(payment, kuaishouBid);
                    if (kuaishouTask != null) {
                        callbackTasks.add(kuaishouTask);
                    }
                    continue;
                }

                conversionRecords.add(toConversionRecord(payment, hasTencentBid ? tencentBid : null, hasTencentBid ? tencentBid.getMedia() : ""));
                result.skipped.incrementAndGet();
                continue;
            }

            conversionRecords.add(toHonorConversionRecord(payment, honorBid));
            result.matched.incrementAndGet();
            log.info("[ConversionSync] 匹配到荣耀转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
                    payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());

            CallbackTask honorTask = buildHonorCallbackTask(payment, honorBid);
            if (honorTask != null) {
                callbackTasks.add(honorTask);
            }
        }
    }

    private CallbackTask buildTencentCallbackTask(PaymentInfo payment, BidRecord bid) {
        String actionType = null;
        if (tagEventResolver != null) {
            actionType = tagEventResolver.resolveActionType(bid.getTagId(), payment.getAct());
        }
        if (actionType == null || actionType.isBlank()) {
            try {
                actionType = tencentClient.conversionActionType(payment.getAct());
                log.warn("[ConversionSync] 广告位回传方式未配置, 回退act-map | qk={} | tagId={} | act={} | actionType={}",
                        payment.getQk(), bid.getTagId(), payment.getAct(), actionType);
            } catch (IllegalArgumentException e) {
                // act 既没配广告位回传方式，也没配 act-map：只落库不回传，避免整个任务中断
                log.warn("[ConversionSync] 腾讯回传方式未配置, 跳过回传 | qk={} | tagId={} | act={} | date={} | error={}",
                        payment.getQk(), bid.getTagId(), payment.getAct(), payment.getDate(), e.getMessage());
                return null;
            }
        }

        return CallbackTask.tencent(
                payment,
                bid,
                callbackDedupeKey(payment, bid, actionType),
                actionType
        );
    }

    private CallbackTask buildHonorCallbackTask(PaymentInfo payment, HonorBidRecord bid) {
        if (honorColdStore == null && honorTagEventResolver == null) {
            return null;
        }

        Integer conversionId = honorTagEventResolver != null
                ? honorTagEventResolver.resolveConversionId(bid.getTagId(), payment.getAct())
                : resolveHonorConversionIdFromDb(bid.getTagId(), payment.getAct());
        if (conversionId == null) {
            log.warn("[ConversionSync] 荣耀 conversionId 未配置 | qk={} | tagId={} | act={}",
                    payment.getQk(), bid.getTagId(), payment.getAct());
            return null;
        }

        return CallbackTask.honor(
                payment,
                bid,
                honorCallbackDedupeKey(payment, bid, conversionId),
                conversionId
        );
    }

    private CallbackTask buildKuaishouCallbackTask(PaymentInfo payment, KuaishouBidRecord bid) {
        if (kuaishouColdStore == null && kuaishouTagEventResolver == null) {
            return null;
        }

        KuaishouTagEventRecord tagEvent = resolveKuaishouTagEvent(bid.getTagId(), payment.getAct());
        if (tagEvent == null) {
            log.warn("[ConversionSync] 快手 eventType 未配置 | qk={} | tagId={} | act={}",
                    payment.getQk(), bid.getTagId(), payment.getAct());
            return null;
        }

        return CallbackTask.kuaishou(
                payment,
                bid,
                kuaishouCallbackDedupeKey(payment, bid, tagEvent.getEventType()),
                tagEvent
        );
    }

    private Integer resolveHonorConversionIdFromDb(String tagId, int baiduAct) {
        if (honorColdStore == null) {
            return null;
        }
        com.adx.tencent.honor.model.HonorTagEventRecord exact = honorColdStore.getTagEvent(tagId, baiduAct);
        if (exact != null) {
            return exact.getConversionId();
        }
        com.adx.tencent.honor.model.HonorTagEventRecord fallback = honorColdStore.getTagEvent(tagId, 0);
        return fallback != null ? fallback.getConversionId() : null;
    }

    private KuaishouTagEventRecord resolveKuaishouTagEvent(String tagId, int baiduAct) {
        if (tagId == null || tagId.isBlank() || kuaishouColdStore == null) {
            return null;
        }
        try {
            KuaishouTagEventRecord record = kuaishouColdStore.getTagEvent(tagId, baiduAct);
            if (record == null) {
                record = kuaishouColdStore.getTagEvent(tagId, 0);
            }
            return record;
        } catch (Exception e) {
            log.warn("[ConversionSync] 查询快手 tag event 失败 | tagId={} | act={} | error={}",
                    tagId, baiduAct, e.getMessage(), e);
            return null;
        }
    }

    // ─── Step 3: 统一落百度转化明细 ───────────────────────────────────────────

    private void saveConversions(List<ConversionRecord> records) {
        if (coldStore == null || records.isEmpty()) {
            return;
        }
        try {
            coldStore.saveConversions(records);
        } catch (Exception e) {
            log.error("[ConversionSync] batch save conversions failed: size={}, {}", records.size(), e.getMessage(), e);
        }
    }

    // ─── Step 4: 回传前去重 ─────────────────────────────────────────────────

    private List<CallbackTask> filterPendingCallbacks(List<CallbackTask> callbackTasks, SyncResult result) {
        Set<String> existing = new HashSet<>();
        existing.addAll(collectTencentSuccessfulCallbackKeys(callbackTasks));
        existing.addAll(collectHonorSuccessfulCallbackKeys(callbackTasks));
        existing.addAll(collectKuaishouSuccessfulCallbackKeys(callbackTasks));

        List<CallbackTask> pending = new ArrayList<>(callbackTasks.size());
        for (CallbackTask task : callbackTasks) {
            if (existing.contains(task.callbackKey)) {
                result.alreadySent.incrementAndGet();
            } else {
                pending.add(task);
            }
        }
        return pending;
    }

    private Set<String> collectTencentSuccessfulCallbackKeys(List<CallbackTask> callbackTasks) {
        if (coldStore == null) {
            return Set.of();
        }

        List<String> callbackKeys = new ArrayList<>();
        List<String> legacyKeys = new ArrayList<>();
        List<Integer> eventTypes = new ArrayList<>();
        for (CallbackTask task : callbackTasks) {
            if (!task.isTencent()) {
                continue;
            }
            callbackKeys.add(task.callbackKey);
            legacyKeys.add(legacyCallbackDedupeKey(task.payment, "tencent"));
            eventTypes.add(task.payment.getAct());
        }

        if (callbackKeys.isEmpty()) {
            return Set.of();
        }

        try {
            return new HashSet<>(coldStore.successfulCallbackKeysWithLegacy(callbackKeys, legacyKeys, eventTypes));
        } catch (Exception e) {
            log.error("[ConversionSync] batch dedup check error: size={}, {}", callbackKeys.size(), e.getMessage(), e);
            return Set.of();
        }
    }

    private Set<String> collectHonorSuccessfulCallbackKeys(List<CallbackTask> callbackTasks) {
        if (honorColdStore == null) {
            return Set.of();
        }

        Set<String> keys = new HashSet<>();
        for (CallbackTask task : callbackTasks) {
            if (task.isHonor()) {
                keys.add(task.callbackKey);
            }
        }

        if (keys.isEmpty()) {
            return Set.of();
        }

        try {
            return honorColdStore.successfulCallbackKeys(keys);
        } catch (Exception e) {
            log.error("[ConversionSync] honor dedup check error: size={}, {}", keys.size(), e.getMessage(), e);
            return Set.of();
        }
    }

    private Set<String> collectKuaishouSuccessfulCallbackKeys(List<CallbackTask> callbackTasks) {
        if (kuaishouColdStore == null) {
            return Set.of();
        }

        Set<String> keys = new HashSet<>();
        for (CallbackTask task : callbackTasks) {
            if (task.isKuaishou()) {
                keys.add(task.callbackKey);
            }
        }

        if (keys.isEmpty()) {
            return Set.of();
        }

        try {
            return kuaishouColdStore.successfulCallbackKeys(keys);
        } catch (Exception e) {
            log.error("[ConversionSync] kuaishou dedup check error: size={}, {}", keys.size(), e.getMessage(), e);
            return Set.of();
        }
    }

    // ─── Step 5: 并发回传媒体 ───────────────────────────────────────────────

    private List<CallbackDispatchResult> dispatchCallbacks(List<CallbackTask> tasks,
                                                           ExecutorService executor,
                                                           SyncResult result) {
        List<CompletableFuture<CallbackDispatchResult>> futures = new ArrayList<>(tasks.size());
        for (CallbackTask task : tasks) {
            futures.add(CompletableFuture.supplyAsync(() -> sendCallback(task), executor));
        }

        List<CallbackDispatchResult> results = new ArrayList<>(tasks.size());
        for (CompletableFuture<CallbackDispatchResult> future : futures) {
            CallbackDispatchResult dispatchResult = future.join();
            if (dispatchResult == null || dispatchResult.record == null) {
                continue;
            }
            if (dispatchResult.record.isOk()) {
                result.sent.incrementAndGet();
            } else {
                result.failed.incrementAndGet();
            }
            results.add(dispatchResult);
        }
        return results;
    }

    private CallbackDispatchResult sendCallback(CallbackTask task) {
        if (task.isTencent()) {
            DeductionHandling handling = handleTencentDeduction(task);
            if (handling == DeductionHandling.SKIP_DUPLICATE) {
                return null;
            }
            if (handling == DeductionHandling.DEDUCT) {
                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);
            callbackRecord.setDedupeKey(task.callbackKey);
            return CallbackDispatchResult.tencent(task, callbackRecord);
        }

        if (task.isHonor()) {
            HonorMediaCallbackRecord honorRecord = honorClient.sendConversion(
                    task.honorBid, task.payment, task.honorConversionId);
            return CallbackDispatchResult.honor(task, toCallbackRecord(task, honorRecord));
        }

        if (task.isKuaishou()) {
            int deductionRate = resolveKuaishouDeductionRate(task.kuaishouBid);
            if (handleKuaishouDeduction(task)) {
                return CallbackDispatchResult.kuaishou(task, toCallbackRecord(task, buildDeductedKuaishouCallbackRecord(task, deductionRate)));
            }
            KuaishouMediaCallbackRecord kuaishouRecord = kuaishouClient.sendConversion(
                    task.kuaishouBid, task.payment, task.kuaishouTagEvent, true);
            kuaishouRecord.setDeductionRate(deductionRate);
            return CallbackDispatchResult.kuaishou(task, toCallbackRecord(task, kuaishouRecord));
        }

        return null;
    }

    // ─── Step 6: 落媒体回传结果 ─────────────────────────────────────────────

    private void saveCallbackResults(List<CallbackDispatchResult> results) {
        if (results.isEmpty()) {
            return;
        }

        List<MediaCallbackRecord> tencentRecords = new ArrayList<>();
        List<HonorMediaCallbackRecord> honorRecords = new ArrayList<>();
        List<KuaishouMediaCallbackRecord> kuaishouRecords = new ArrayList<>();
        for (CallbackDispatchResult result : results) {
            if (result.isHonor()) {
                honorRecords.add(toHonorCallbackRecord(result.task, result.record));
            } else if (result.isKuaishou()) {
                kuaishouRecords.add(toKuaishouCallbackRecord(result.task, result.record));
            } else {
                tencentRecords.add(result.record);
            }
        }

        if (coldStore != null && !tencentRecords.isEmpty()) {
            try {
                coldStore.saveMediaCallbacks(tencentRecords);
            } catch (Exception e) {
                log.error("[ConversionSync] batch save callbacks failed: size={}, {}", tencentRecords.size(), e.getMessage(), e);
            }
        }

        if (honorColdStore != null && !honorRecords.isEmpty()) {
            try {
                honorColdStore.saveMediaCallbacks(honorRecords);
            } catch (Exception e) {
                log.error("[ConversionSync] batch save honor callbacks failed: size={}, {}", honorRecords.size(), e.getMessage(), e);
            }
        }

        if (kuaishouColdStore != null && !kuaishouRecords.isEmpty()) {
            try {
                kuaishouColdStore.saveMediaCallbacks(kuaishouRecords);
            } catch (Exception e) {
                log.error("[ConversionSync] batch save kuaishou callbacks failed: size={}, {}", kuaishouRecords.size(), e.getMessage(), e);
            }
        }
    }

    // ─── 工具方法 ────────────────────────────────────────────────────────────

    private static String resolvePlatform(BidRecord bid) {
        if (bid.getPlatform() != null && !bid.getPlatform().isBlank()) {
            return bid.getPlatform().toLowerCase();
        }
        if (bid.getMediaParams() == null) {
            return "android";
        }
        String platform = bid.getMediaParams().get("baidu_platform");
        if (platform != null && !platform.isBlank()) {
            return platform.toLowerCase();
        }
        String idfa = bid.getMediaParams().get("idfa");
        if (idfa != null && !idfa.isBlank()) {
            return "ios";
        }
        return "android";
    }

    private static ConversionRecord toConversionRecord(PaymentInfo payment, BidRecord bid, String media) {
        ConversionRecord record = new ConversionRecord();
        record.setDedupeKey(conversionDedupeKey(payment));
        record.setQk(payment.getQk());
        record.setMedia(media);
        record.setTagId(bid != null ? bid.getTagId() : null);
        record.setDate(payment.getDate());
        record.setAppSid(payment.getAppSid());
        record.setCustomerName(payment.getCustomerName());
        record.setDeviceId(payment.getDeviceId());
        record.setConv(payment.getConv());
        record.setPayment(payment.getPayment());
        record.setGmv(payment.getGmv());
        record.setAct(payment.getAct());
        record.setTu(payment.getTu());
        record.setClkTime(payment.getClkTime());
        record.setCreatedAt(Instant.now());
        return record;
    }

    private static ConversionRecord toHonorConversionRecord(PaymentInfo payment, HonorBidRecord bid) {
        ConversionRecord record = new ConversionRecord();
        record.setDedupeKey(conversionDedupeKey(payment));
        record.setQk(payment.getQk());
        record.setMedia("honor");
        record.setTagId(bid != null ? bid.getTagId() : null);
        record.setDate(payment.getDate());
        record.setAppSid(payment.getAppSid());
        record.setCustomerName(payment.getCustomerName());
        record.setDeviceId(payment.getDeviceId());
        record.setConv(payment.getConv());
        record.setPayment(payment.getPayment());
        record.setGmv(payment.getGmv());
        record.setAct(payment.getAct());
        record.setTu(payment.getTu());
        record.setClkTime(payment.getClkTime());
        record.setCreatedAt(Instant.now());
        return record;
    }

    private static ConversionRecord toKuaishouConversionRecord(PaymentInfo payment, KuaishouBidRecord bid) {
        ConversionRecord record = new ConversionRecord();
        record.setDedupeKey(conversionDedupeKey(payment));
        record.setQk(payment.getQk());
        record.setMedia("kuaishou");
        record.setTagId(bid != null ? bid.getTagId() : null);
        record.setDate(payment.getDate());
        record.setAppSid(payment.getAppSid());
        record.setCustomerName(payment.getCustomerName());
        record.setDeviceId(payment.getDeviceId());
        record.setConv(payment.getConv());
        record.setPayment(payment.getPayment());
        record.setGmv(payment.getGmv());
        record.setAct(payment.getAct());
        record.setTu(payment.getTu());
        record.setClkTime(payment.getClkTime());
        record.setCreatedAt(Instant.now());
        return record;
    }

    static String conversionDedupeKey(PaymentInfo payment) {
        return String.format("%s:%s:%s:%d",
                payment.getQk(),
                payment.getDate(),
                payment.getDeviceId(),
                payment.getAct());
    }

    static String callbackDedupeKey(PaymentInfo payment, BidRecord bid, String actionType) {
        return String.format("%s:%s:%s:%s:%s:%d:%s",
                bid.getMedia(),
                payment.getQk(),
                payment.getDate(),
                payment.getDeviceId(),
                bid.getTagId(),
                payment.getAct(),
                actionType);
    }

    static String legacyCallbackDedupeKey(PaymentInfo payment, String media) {
        return String.format("%s:%s:%s:%s:%d:%d",
                media,
                payment.getQk(),
                payment.getDate(),
                payment.getDeviceId(),
                payment.getAct(),
                payment.getAct());
    }

    static String honorCallbackDedupeKey(PaymentInfo payment, HonorBidRecord bid, int conversionId) {
        return String.format("honor:%s:%s:%s:%s:%d:%d",
                payment.getQk(),
                payment.getDate(),
                payment.getDeviceId(),
                bid.getTagId(),
                payment.getAct(),
                conversionId);
    }

    static String kuaishouCallbackDedupeKey(PaymentInfo payment, KuaishouBidRecord bid, int eventType) {
        return String.format("kuaishou:%s:%s:%s:%s:%d:%d",
                payment.getQk(),
                payment.getDate(),
                payment.getDeviceId(),
                bid.getTagId(),
                payment.getAct(),
                eventType);
    }

    private DeductionHandling handleTencentDeduction(CallbackTask task) {
        if (!task.isTencent() || task.tencentBid == null) {
            return DeductionHandling.SEND;
        }
        int deductionRate = resolveDeductionRate(task.tencentBid);
        if (task.payment.getAct() == 2001 || deductionRate <= 0) {
            return DeductionHandling.SEND;
        }
        if (!isNewTrackingLink(task.tencentBid)) {
            return DeductionHandling.SEND;
        }
        if (!hotStore.markConversionSeen(task.callbackKey)) {
            log.info("[ConversionSync] 腾讯转化已见过，跳过后续发送 | callbackKey={} | qk={} | accountId={}",
                    task.callbackKey, task.payment.getQk(), safe(task.tencentBid.getAccountId()));
            return DeductionHandling.SKIP_DUPLICATE;
        }
        RedisHotStore.DeductionDecision decision = hotStore.evaluateDeduction(
                task.tencentBid.getAccountId(),
                task.payment.getAct(),
                resolveTrackingVersion(task.tencentBid),
                task.payment.getDate(),
                deductionRate
        );
        boolean deduct = decision.isShouldDeduct();
        log.info("[ConversionSync] 腾讯转化扣量判定 | qk={} | act={} | accountId={} | version={} | rate={} | totalSeen={} | totalDeducted={} | callbackKey={} | deduct={}",
                task.payment.getQk(), task.payment.getAct(), safe(task.tencentBid.getAccountId()),
                resolveTrackingVersion(task.tencentBid), deductionRate,
                decision.getTotalSeen(), decision.getTotalDeducted(), task.callbackKey, deduct);
        return deduct ? DeductionHandling.DEDUCT : DeductionHandling.SEND;
    }

    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)
                || "v3".equalsIgnoreCase(version)
                || "v4".equalsIgnoreCase(version);
    }

    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(resolveTrackingVersion(task.tencentBid));
        record.setCreatedAt(now);
        int deductionRate = resolveDeductionRate(task.tencentBid);
        log.info("[ConversionSync] 腾讯转化扣量 | qk={} | act={} | rate={} | tagId={} | accountId={} | callbackKey={}",
                task.payment.getQk(), task.payment.getAct(), deductionRate,
                task.tencentBid.getTagId(), safe(task.tencentBid.getAccountId()), task.callbackKey);
        return record;
    }

    private boolean handleKuaishouDeduction(CallbackTask task) {
        if (!task.isKuaishou() || task.kuaishouBid == null || kuaishouHotStore == null) {
            return false;
        }
        int deductionRate = resolveKuaishouDeductionRate(task.kuaishouBid);
        if (task.payment.getAct() == 2001 || deductionRate <= 0) {
            return false;
        }
        if (!kuaishouHotStore.markConversionSeen(task.callbackKey)) {
            log.info("[ConversionSync] 快手转化已见过，跳过重复扣量判断 | callbackKey={} | qk={} | accountId={}",
                    task.callbackKey, task.payment.getQk(), safe(task.kuaishouBid.getAccountId()));
            return false;
        }
        KuaishouHotStore.DeductionDecision decision = kuaishouHotStore.evaluateDeduction(
                task.kuaishouBid.getAccountId(),
                task.payment.getAct(),
                task.payment.getDate(),
                deductionRate
        );
        boolean deduct = decision.isShouldDeduct();
        log.info("[ConversionSync] 快手转化扣量判定 | qk={} | act={} | accountId={} | tagId={} | rate={} | totalSeen={} | totalDeducted={} | callbackKey={} | deduct={}",
                task.payment.getQk(), task.payment.getAct(), safe(task.kuaishouBid.getAccountId()),
                safe(task.kuaishouBid.getTagId()), deductionRate,
                decision.getTotalSeen(), decision.getTotalDeducted(), task.callbackKey, deduct);
        return deduct;
    }

    private KuaishouMediaCallbackRecord buildDeductedKuaishouCallbackRecord(CallbackTask task, int deductionRate) {
        KuaishouMediaCallbackRecord record = new KuaishouMediaCallbackRecord();
        Instant now = Instant.now();
        record.setDedupeKey(task.callbackKey);
        record.setQk(task.payment.getQk());
        record.setTagId(task.kuaishouBid.getTagId());
        String callbackUrl = task.kuaishouBid.getMediaParams() != null
                ? task.kuaishouBid.getMediaParams().get(KuaishouBidRecord.CALLBACK_KEY)
                : null;
        record.setCallbackUrl(callbackUrl != null ? callbackUrl : "");
        record.setEventType(task.kuaishouTagEvent.getEventType());
        record.setEventTimeMs(now.toEpochMilli());
        record.setPurchase(Math.max(task.payment.getPayment(), 0D));
        record.setDeductionRate(deductionRate);
        record.setDirectMatch(false);
        record.setDispatchStatus(KuaishouMediaCallbackRecord.DISPATCH_STATUS_DEDUCTED);
        record.setStatus(0);
        record.setOk(false);
        record.setAttempt(1);
        record.setErrorMessage("deducted_by_rate");
        record.setCreatedAt(now);
        log.info("[ConversionSync] 快手转化扣量 | qk={} | act={} | rate={} | tagId={} | accountId={} | callbackKey={}",
                task.payment.getQk(), task.payment.getAct(), deductionRate,
                task.kuaishouBid.getTagId(), safe(task.kuaishouBid.getAccountId()), task.callbackKey);
        return record;
    }

    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 int resolveDeductionRate(BidRecord bid) {
        if (bid == null || bid.getMediaParams() == null) {
            return 0;
        }
        String rawRate = bid.getMediaParams().get(BidRecord.DEDUCTION_RATE_KEY);
        if (rawRate != null && !rawRate.isBlank()) {
            try {
                return sanitizeDeductionRate(Integer.parseInt(rawRate.trim()));
            } catch (NumberFormatException ignored) {
            }
        }
        return 0;
    }

    private int resolveKuaishouDeductionRate(KuaishouBidRecord bid) {
        if (bid == null || bid.getMediaParams() == null) {
            return 0;
        }
        String rawRate = bid.getMediaParams().get(KuaishouBidRecord.DEDUCTION_RATE_KEY);
        if (rawRate != null && !rawRate.isBlank()) {
            try {
                return sanitizeDeductionRate(Integer.parseInt(rawRate.trim()));
            } catch (NumberFormatException ignored) {
            }
        }
        return 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());
        callback.setQk(record.getQk());
        callback.setTagId(task.tagId);
        callback.setCallbackUrl(record.getCallbackUrl());
        callback.setEventType(record.getEventType());
        callback.setEventTimeMs(record.getEventTimeMs());
        callback.setPurchase(record.getPurchase());
        callback.setStatus(record.getStatus());
        callback.setOk(record.isOk());
        callback.setAttempt(record.getAttempt());
        callback.setResponseBody(record.getResponseBody());
        callback.setErrorMessage(record.getErrorMessage());
        callback.setRequestBody(record.getRequestBody());
        callback.setCreatedAt(record.getCreatedAt());
        return callback;
    }

    private static KuaishouMediaCallbackRecord toKuaishouCallbackRecord(CallbackTask task, MediaCallbackRecord record) {
        KuaishouMediaCallbackRecord callback = new KuaishouMediaCallbackRecord();
        callback.setDedupeKey(record.getDedupeKey());
        callback.setQk(record.getQk());
        callback.setTagId(task.tagId);
        callback.setCallbackUrl(record.getCallbackUrl());
        callback.setEventType(record.getEventType());
        callback.setEventTimeMs(record.getEventTimeMs());
        callback.setPurchase(record.getPurchase());
        callback.setDeductionRate(parseKuaishouDeductionRate(task.kuaishouBid));
        callback.setStatus(record.getStatus());
        callback.setOk(record.isOk());
        callback.setAttempt(record.getAttempt());
        callback.setResponseBody(record.getResponseBody());
        callback.setErrorMessage(record.getErrorMessage());
        callback.setRequestBody(record.getRequestBody());
        callback.setDispatchStatus(record.getDispatchStatus());
        callback.setDirectMatch(!KuaishouMediaCallbackRecord.DISPATCH_STATUS_DEDUCTED.equals(record.getDispatchStatus()));
        callback.setCreatedAt(record.getCreatedAt());
        return callback;
    }

    private static MediaCallbackRecord toCallbackRecord(CallbackTask task, HonorMediaCallbackRecord honorRecord) {
        MediaCallbackRecord callbackRecord = new MediaCallbackRecord();
        callbackRecord.setDedupeKey(task.callbackKey);
        callbackRecord.setMedia("honor");
        callbackRecord.setQk(honorRecord.getQk());
        callbackRecord.setCallbackUrl(honorRecord.getCallbackUrl());
        callbackRecord.setEventType(honorRecord.getEventType());
        callbackRecord.setEventTimeMs(honorRecord.getEventTimeMs());
        callbackRecord.setPurchase(honorRecord.getPurchase());
        callbackRecord.setStatus(honorRecord.getStatus());
        callbackRecord.setOk(honorRecord.isOk());
        callbackRecord.setAttempt(honorRecord.getAttempt());
        callbackRecord.setResponseBody(honorRecord.getResponseBody());
        callbackRecord.setErrorMessage(honorRecord.getErrorMessage());
        callbackRecord.setRequestBody(honorRecord.getRequestBody());
        callbackRecord.setTrackingVersion("v1");
        callbackRecord.setCreatedAt(honorRecord.getCreatedAt());
        return callbackRecord;
    }

    private static MediaCallbackRecord toCallbackRecord(CallbackTask task, KuaishouMediaCallbackRecord kuaishouRecord) {
        MediaCallbackRecord callbackRecord = new MediaCallbackRecord();
        callbackRecord.setDedupeKey(task.callbackKey);
        callbackRecord.setMedia("kuaishou");
        callbackRecord.setQk(kuaishouRecord.getQk());
        callbackRecord.setCallbackUrl(kuaishouRecord.getCallbackUrl());
        callbackRecord.setEventType(kuaishouRecord.getEventType());
        callbackRecord.setEventTimeMs(kuaishouRecord.getEventTimeMs());
        callbackRecord.setPurchase(kuaishouRecord.getPurchase());
        callbackRecord.setStatus(kuaishouRecord.getStatus());
        callbackRecord.setOk(kuaishouRecord.isOk());
        callbackRecord.setAttempt(kuaishouRecord.getAttempt());
        callbackRecord.setResponseBody(kuaishouRecord.getResponseBody());
        callbackRecord.setErrorMessage(kuaishouRecord.getErrorMessage());
        callbackRecord.setRequestBody(kuaishouRecord.getRequestBody());
        callbackRecord.setDispatchStatus(kuaishouRecord.getDispatchStatus());
        callbackRecord.setTrackingVersion(kuaishouRecord.isDirectMatch() ? "direct" : "deducted");
        callbackRecord.setCreatedAt(kuaishouRecord.getCreatedAt());
        return callbackRecord;
    }

    private static int parseKuaishouDeductionRate(KuaishouBidRecord bid) {
        if (bid == null || bid.getMediaParams() == null) {
            return 0;
        }
        String rawRate = bid.getMediaParams().get(KuaishouBidRecord.DEDUCTION_RATE_KEY);
        if (rawRate == null || rawRate.isBlank()) {
            return 0;
        }
        try {
            int rate = Integer.parseInt(rawRate.trim());
            if (rate < 0) {
                return 0;
            }
            return Math.min(rate, 100);
        } catch (NumberFormatException ignored) {
            return 0;
        }
    }

    // ─── 内部数据结构 ────────────────────────────────────────────────────────

    private static class BidLookup {
        private final Map<String, BidRecord> tencentBids;
        private final Map<String, HonorBidRecord> honorBids;
        private final Map<String, KuaishouBidRecord> kuaishouBids;

        private BidLookup(Map<String, BidRecord> tencentBids,
                          Map<String, HonorBidRecord> honorBids,
                          Map<String, KuaishouBidRecord> kuaishouBids) {
            this.tencentBids = tencentBids;
            this.honorBids = honorBids;
            this.kuaishouBids = kuaishouBids;
        }
    }

    private static class CallbackTask {
        private final String media;
        private final PaymentInfo payment;
        private final String tagId;
        private final String callbackKey;
        private final BidRecord tencentBid;
        private final String tencentActionType;
        private final HonorBidRecord honorBid;
        private final Integer honorConversionId;
        private final KuaishouBidRecord kuaishouBid;
        private final KuaishouTagEventRecord kuaishouTagEvent;

        private CallbackTask(String media,
                             PaymentInfo payment,
                             String tagId,
                             String callbackKey,
                             BidRecord tencentBid,
                             String tencentActionType,
                             HonorBidRecord honorBid,
                             Integer honorConversionId,
                             KuaishouBidRecord kuaishouBid,
                             KuaishouTagEventRecord kuaishouTagEvent) {
            this.media = media;
            this.payment = payment;
            this.tagId = tagId;
            this.callbackKey = callbackKey;
            this.tencentBid = tencentBid;
            this.tencentActionType = tencentActionType;
            this.honorBid = honorBid;
            this.honorConversionId = honorConversionId;
            this.kuaishouBid = kuaishouBid;
            this.kuaishouTagEvent = kuaishouTagEvent;
        }

        private static CallbackTask tencent(PaymentInfo payment,
                                            BidRecord bid,
                                            String callbackKey,
                                            String actionType) {
            return new CallbackTask("tencent", payment, bid.getTagId(), callbackKey, bid, actionType, null, null, null, null);
        }

        private static CallbackTask honor(PaymentInfo payment,
                                          HonorBidRecord bid,
                                          String callbackKey,
                                          Integer conversionId) {
            return new CallbackTask("honor", payment, bid.getTagId(), callbackKey, null, null, bid, conversionId, null, null);
        }

        private static CallbackTask kuaishou(PaymentInfo payment,
                                             KuaishouBidRecord bid,
                                             String callbackKey,
                                             KuaishouTagEventRecord tagEvent) {
            return new CallbackTask("kuaishou", payment, bid.getTagId(), callbackKey, null, null, null, null, bid, tagEvent);
        }

        private boolean isTencent() {
            return "tencent".equals(media);
        }

        private boolean isHonor() {
            return "honor".equals(media);
        }

        private boolean isKuaishou() {
            return "kuaishou".equals(media);
        }
    }

    private static class CallbackDispatchResult {
        private final String media;
        private final CallbackTask task;
        private final MediaCallbackRecord record;

        private CallbackDispatchResult(String media, CallbackTask task, MediaCallbackRecord record) {
            this.media = media;
            this.task = task;
            this.record = record;
        }

        private static CallbackDispatchResult tencent(CallbackTask task, MediaCallbackRecord record) {
            return new CallbackDispatchResult("tencent", task, record);
        }

        private static CallbackDispatchResult honor(CallbackTask task, MediaCallbackRecord record) {
            return new CallbackDispatchResult("honor", task, record);
        }

        private static CallbackDispatchResult kuaishou(CallbackTask task, MediaCallbackRecord record) {
            return new CallbackDispatchResult("kuaishou", task, record);
        }

        private boolean isHonor() {
            return "honor".equals(media);
        }

        private boolean isKuaishou() {
            return "kuaishou".equals(media);
        }
    }

    private enum DeductionHandling {
        SEND,
        DEDUCT,
        SKIP_DUPLICATE
    }

    public static class SyncResult {
        public final AtomicInteger fetched = new AtomicInteger();
        public final AtomicInteger matched = new AtomicInteger();
        public final AtomicInteger sent = new AtomicInteger();
        public final AtomicInteger skipped = new AtomicInteger();
        public final AtomicInteger failed = new AtomicInteger();
        public final AtomicInteger alreadySent = new AtomicInteger();
    }
}