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.storage.model.AccountTagEventRecord;
import com.adx.tencent.tagsync.AccountTagEventResolver;
import com.adx.tencent.tagsync.TagEventResolver;
import com.adx.tencent.tencent.TencentClient;
import com.adx.tencent.vivo.client.VivoClient;
import com.adx.tencent.vivo.model.VivoBidRecord;
import com.adx.tencent.vivo.model.VivoMediaCallbackRecord;
import com.adx.tencent.vivo.model.VivoTagEventRecord;
import com.adx.tencent.vivo.service.VivoTagEventResolver;
import com.adx.tencent.vivo.store.VivoColdStore;
import com.adx.tencent.vivo.store.VivoHotStore;
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 static final int KS_EVENT_ACTIVATION = 1;
    private static final int KS_EVENT_REGISTRATION = 2;
    private static final int KS_EVENT_PURCHASE = 3;
    private static final int KS_EVENT_NEXT_DAY_RETENTION = 7;
    private static final int KS_EVENT_SEVEN_DAY_RETENTION = 8;
    private static final int KS_EVENT_APP_WAKEUP = 84;
    private static final int KS_EVENT_KEY_ACTION = 143;
    private static final int KS_EVENT_THREE_DAY_RETENTION = 103;
    private static final String VIVO_EVENT_ACTIVATION = "ACTIVATION";
    private static final String VIVO_EVENT_REGISTER = "REGISTER";
    private static final String VIVO_EVENT_PAY = "PAY";
    private static final String VIVO_EVENT_RETENTION_1 = "RETENTION_1";
    private static final String VIVO_EVENT_RETENTION_3 = "RETENTION_3";
    private static final String VIVO_EVENT_RETENTION_7 = "RETENTION_7";
    private static final String VIVO_EVENT_REACTIVATION = "REACTIVATION";
    private static final String VIVO_EVENT_OTHER = "OTHER";

    private final ConversionClient baiduClient;
    private final RedisHotStore hotStore;
    private final HonorHotStore honorHotStore;
    private final KuaishouHotStore kuaishouHotStore;
    private final VivoHotStore vivoHotStore;
    private final TencentClient tencentClient;
    private final TiDBColdStore coldStore;
    private final HonorColdStore honorColdStore;
    private final KuaishouColdStore kuaishouColdStore;
    private final VivoColdStore vivoColdStore;
    private final HonorClient honorClient;
    private final KuaishouClient kuaishouClient;
    private final VivoClient vivoClient;
    private final TagEventResolver tagEventResolver;
    private final AccountTagEventResolver accountTagEventResolver;
    private final HonorTagEventResolver honorTagEventResolver;
    private final KuaishouTagEventResolver kuaishouTagEventResolver;
    private final VivoTagEventResolver vivoTagEventResolver;
    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,
                                 AccountTagEventResolver accountTagEventResolver,
                                 HonorTagEventResolver honorTagEventResolver,
                                 KuaishouTagEventResolver kuaishouTagEventResolver,
                                 VivoTagEventResolver vivoTagEventResolver,
                                 HonorHotStore honorHotStore,
                                 HonorColdStore honorColdStore,
                                 KuaishouHotStore kuaishouHotStore,
                                 KuaishouColdStore kuaishouColdStore,
                                 VivoHotStore vivoHotStore,
                                 VivoColdStore vivoColdStore,
                                 HonorClient honorClient,
                                 KuaishouClient kuaishouClient,
                                 VivoClient vivoClient,
                                 AppProperties props) {
        this.baiduClient = baiduClient;
        this.hotStore = hotStore;
        this.tencentClient = tencentClient;
        this.coldStore = coldStore;
        this.tagEventResolver = tagEventResolver;
        this.accountTagEventResolver = accountTagEventResolver;
        this.honorTagEventResolver = honorTagEventResolver;
        this.kuaishouTagEventResolver = kuaishouTagEventResolver;
        this.vivoTagEventResolver = vivoTagEventResolver;
        this.honorHotStore = honorHotStore;
        this.honorColdStore = honorColdStore;
        this.kuaishouHotStore = kuaishouHotStore;
        this.kuaishouColdStore = kuaishouColdStore;
        this.vivoHotStore = vivoHotStore;
        this.vivoColdStore = vivoColdStore;
        this.honorClient = honorClient;
        this.kuaishouClient = kuaishouClient;
        this.vivoClient = vivoClient;
        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();
        Map<String, VivoBidRecord> vivoBids = vivoHotStore != null
                ? vivoHotStore.findBidsByQks(qkList)
                : Map.of();
        return new BidLookup(tencentBids, honorBids, kuaishouBids, vivoBids);
    }

    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());
            VivoBidRecord vivoBid = lookup.vivoBids.get(payment.getQk());

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

            boolean hasTencentBid = tencentBid != null;
            boolean hasHonorBid = honorBid != null;
            boolean hasKuaishouBid = kuaishouBid != null;
            boolean hasVivoBid = vivoBid != null;
            boolean validTencent = hasTencentBid && "tencent".equals(tencentBid.getMedia());
            boolean validHonor = hasHonorBid && "honor".equals(honorBid.getMedia());
            boolean validKuaishou = hasKuaishouBid && "kuaishou".equals(kuaishouBid.getMedia());
            boolean validVivo = hasVivoBid && "vivo".equals(vivoBid.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());
            }

            if (hasVivoBid && !validVivo) {
                log.warn("[ConversionSync] vivo bid media 非法 | qk={} | media={}", payment.getQk(), vivoBid.getMedia());
            }

            int matchedMediaCount = (validTencent ? 1 : 0) + (validHonor ? 1 : 0) + (validKuaishou ? 1 : 0) + (validVivo ? 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) {
                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);
                }
                continue;
            }

            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());

                List<CallbackTask> kuaishouTasks = buildKuaishouCallbackTasks(payment, kuaishouBid);
                if (!kuaishouTasks.isEmpty()) {
                    callbackTasks.addAll(kuaishouTasks);
                }
                continue;
            }

            if (validVivo) {
                conversionRecords.add(toVivoConversionRecord(payment, vivoBid));
                result.matched.incrementAndGet();
                log.info("[ConversionSync] 匹配到 vivo 转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
                        payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());

                List<CallbackTask> vivoTasks = buildVivoCallbackTasks(payment, vivoBid);
                if (!vivoTasks.isEmpty()) {
                    callbackTasks.addAll(vivoTasks);
                }
                continue;
            }

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

    private CallbackTask buildTencentCallbackTask(PaymentInfo payment, BidRecord bid) {
        String actionType = null;
        String customAction = null;
        if (accountTagEventResolver != null) {
            AccountTagEventRecord accountMapping = accountTagEventResolver.resolveAction(
                    bid.getTagId(), bid.getAccountId(), payment.getAct());
            if (accountMapping != null) {
                actionType = accountMapping.getTencentActionType();
                customAction = accountMapping.getTencentCustomAction();
                if ("CUSTOM".equals(actionType) && (customAction == null || customAction.isBlank())) {
                    log.warn("[ConversionSync] 账户级 CUSTOM 回传缺少 custom_action, 忽略账户级配置并回退广告位配置 | qk={} | tagId={} | accountId={} | act={}",
                            payment.getQk(), bid.getTagId(), bid.getAccountId(), payment.getAct());
                    actionType = null;
                    customAction = null;
                }
            }
        }
        if (tagEventResolver != null) {
            if (actionType == null || actionType.isBlank()) {
                actionType = tagEventResolver.resolveActionType(bid.getTagId(), payment.getAct());
                customAction = null;
            }
        }
        if (actionType == null || actionType.isBlank()) {
            try {
                actionType = tencentClient.conversionActionType(payment.getAct());
                log.warn("[ConversionSync] 账户级/广告位回传方式未配置, 回退act-map | qk={} | tagId={} | accountId={} | act={} | actionType={}",
                        payment.getQk(), bid.getTagId(), bid.getAccountId(), payment.getAct(), actionType);
            } catch (IllegalArgumentException e) {
                // act 既没配广告位回传方式，也没配 act-map：只落库不回传，避免整个任务中断
                log.warn("[ConversionSync] 腾讯回传方式未配置, 跳过回传 | qk={} | tagId={} | accountId={} | act={} | date={} | error={}",
                        payment.getQk(), bid.getTagId(), bid.getAccountId(), payment.getAct(), payment.getDate(), e.getMessage());
                return null;
            }
        }

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

    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) {
            conversionId = fallbackHonorConversionId(payment.getAct());
            if (conversionId == null) {
                log.warn("[ConversionSync] 荣耀 conversionId 未配置且无兜底映射 | qk={} | tagId={} | act={}",
                        payment.getQk(), bid.getTagId(), payment.getAct());
                return null;
            }
            log.warn("[ConversionSync] 荣耀 conversionId 未配置, 启用兜底映射 | qk={} | tagId={} | act={} | conversionId={}",
                    payment.getQk(), bid.getTagId(), payment.getAct(), conversionId);
        }

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

    private static Integer fallbackHonorConversionId(int baiduAct) {
        return switch (baiduAct) {
            case 1 -> 10001;      // 激活
            case 2, 5 -> 10004;   // 付费
            case 3 -> 10002;      // 注册
            case 4 -> 10003;      // 次留
            case 6, 2001 -> 10022;// 拉活
            case 8 -> 10014;      // 7日留存
            case 9 -> 10013;      // 3日留存
            default -> null;
        };
    }

    private List<CallbackTask> buildKuaishouCallbackTasks(PaymentInfo payment, KuaishouBidRecord bid) {
        if (kuaishouColdStore == null && kuaishouTagEventResolver == null) {
            return List.of();
        }

        List<KuaishouTagEventRecord> tagEvents = resolveKuaishouTagEvents(bid.getTagId(), payment.getAct());
        if (tagEvents.isEmpty()) {
            log.warn("[ConversionSync] 快手 eventType 未配置且无兜底映射 | qk={} | tagId={} | act={}",
                    payment.getQk(), bid.getTagId(), payment.getAct());
            return List.of();
        }

        List<CallbackTask> tasks = new ArrayList<>(tagEvents.size());
        for (KuaishouTagEventRecord tagEvent : tagEvents) {
            tasks.add(CallbackTask.kuaishou(
                    payment,
                    bid,
                    kuaishouCallbackDedupeKey(payment, bid, tagEvent.getEventType()),
                    tagEvent
            ));
        }
        return tasks;
    }

    private List<CallbackTask> buildVivoCallbackTasks(PaymentInfo payment, VivoBidRecord bid) {
        if (vivoColdStore == null && vivoTagEventResolver == null) {
            return List.of();
        }

        List<VivoTagEventRecord> tagEvents = resolveVivoTagEvents(bid.getTagId(), payment.getAct());
        if (tagEvents.isEmpty()) {
            log.warn("[ConversionSync] vivo eventType 未配置且无兜底映射 | qk={} | tagId={} | act={}",
                    payment.getQk(), bid.getTagId(), payment.getAct());
            return List.of();
        }

        List<CallbackTask> tasks = new ArrayList<>(tagEvents.size());
        for (VivoTagEventRecord tagEvent : tagEvents) {
            tasks.add(CallbackTask.vivo(
                    payment,
                    bid,
                    vivoCallbackDedupeKey(payment, bid, tagEvent.getEventType()),
                    tagEvent
            ));
        }
        return tasks;
    }

    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 List<KuaishouTagEventRecord> resolveKuaishouTagEvents(String tagId, int baiduAct) {
        List<KuaishouTagEventRecord> configured = resolveConfiguredKuaishouTagEvents(tagId, baiduAct);
        if (!configured.isEmpty()) {
            return configured;
        }

        List<KuaishouTagEventRecord> fallback = buildKuaishouFallbackTagEvents(tagId, baiduAct);
        if (!fallback.isEmpty()) {
            log.warn("[ConversionSync] 快手 eventType 未配置, 启用兜底映射 | tagId={} | act={} | fallbackEventTypes={}",
                    tagId, baiduAct, fallback.stream().map(KuaishouTagEventRecord::getEventType).toList());
        }
        return fallback;
    }

    private List<VivoTagEventRecord> resolveVivoTagEvents(String tagId, int baiduAct) {
        List<VivoTagEventRecord> configured = resolveConfiguredVivoTagEvents(tagId, baiduAct);
        if (!configured.isEmpty()) {
            return configured;
        }

        List<VivoTagEventRecord> fallback = buildVivoFallbackTagEvents(tagId, baiduAct);
        if (!fallback.isEmpty()) {
            log.warn("[ConversionSync] vivo eventType 未配置, 启用兜底映射 | tagId={} | act={} | fallbackEventTypes={}",
                    tagId, baiduAct, fallback.stream().map(VivoTagEventRecord::getEventType).toList());
        }
        return fallback;
    }

    private List<KuaishouTagEventRecord> resolveConfiguredKuaishouTagEvents(String tagId, int baiduAct) {
        if (tagId == null || tagId.isBlank()) {
            return List.of();
        }
        try {
            if (kuaishouTagEventResolver != null) {
                List<Integer> eventTypes = kuaishouTagEventResolver.resolveEventTypes(tagId, baiduAct);
                if (eventTypes != null && !eventTypes.isEmpty()) {
                    List<KuaishouTagEventRecord> records = new ArrayList<>(eventTypes.size());
                    for (Integer eventType : eventTypes) {
                        records.add(buildKuaishouTagEvent(tagId, baiduAct, eventType));
                    }
                    return records;
                }
                return List.of();
            }
            if (kuaishouColdStore == null) {
                return List.of();
            }
            List<KuaishouTagEventRecord> records = kuaishouColdStore.getTagEvents(tagId, baiduAct);
            if (records == null || records.isEmpty()) {
                records = kuaishouColdStore.getTagEvents(tagId, 0);
            }
            return records == null ? List.of() : records;
        } catch (Exception e) {
            log.warn("[ConversionSync] 查询快手 tag event 失败 | tagId={} | act={} | error={}",
                    tagId, baiduAct, e.getMessage(), e);
            return List.of();
        }
    }

    private static KuaishouTagEventRecord buildKuaishouTagEvent(String tagId, int baiduAct, int eventType) {
        KuaishouTagEventRecord record = new KuaishouTagEventRecord();
        record.setTagId(tagId);
        record.setBaiduAct(baiduAct);
        record.setEventType(eventType);
        return record;
    }

    private static List<KuaishouTagEventRecord> buildKuaishouFallbackTagEvents(String tagId, int baiduAct) {
        List<Integer> eventTypes = switch (baiduAct) {
            case 1 -> List.of(KS_EVENT_ACTIVATION);
            case 2, 5 -> List.of(KS_EVENT_PURCHASE);
            case 3 -> List.of(KS_EVENT_REGISTRATION, KS_EVENT_ACTIVATION);
            case 4 -> List.of(KS_EVENT_NEXT_DAY_RETENTION);
            case 6 -> List.of(KS_EVENT_APP_WAKEUP);
            case 7 -> List.of(KS_EVENT_KEY_ACTION);
            case 8 -> List.of(KS_EVENT_SEVEN_DAY_RETENTION);
            case 9 -> List.of(KS_EVENT_THREE_DAY_RETENTION);
            case 2001 -> List.of(KS_EVENT_APP_WAKEUP);
            default -> List.of();
        };
        if (eventTypes.isEmpty()) {
            return List.of();
        }
        List<KuaishouTagEventRecord> records = new ArrayList<>(eventTypes.size());
        for (Integer eventType : eventTypes) {
            records.add(buildKuaishouTagEvent(tagId, baiduAct, eventType));
        }
        return records;
    }

    private List<VivoTagEventRecord> resolveConfiguredVivoTagEvents(String tagId, int baiduAct) {
        if (tagId == null || tagId.isBlank()) {
            return List.of();
        }
        try {
            if (vivoTagEventResolver != null) {
                List<String> eventTypes = vivoTagEventResolver.resolveEventTypes(tagId, baiduAct);
                if (eventTypes != null && !eventTypes.isEmpty()) {
                    List<VivoTagEventRecord> records = new ArrayList<>(eventTypes.size());
                    for (String eventType : eventTypes) {
                        records.add(buildVivoTagEvent(tagId, baiduAct, eventType));
                    }
                    return records;
                }
                return List.of();
            }
            if (vivoColdStore == null) {
                return List.of();
            }
            List<VivoTagEventRecord> records = vivoColdStore.getTagEvents(tagId, baiduAct);
            if (records == null || records.isEmpty()) {
                records = vivoColdStore.getTagEvents(tagId, 0);
            }
            return records == null ? List.of() : records;
        } catch (Exception e) {
            log.warn("[ConversionSync] 查询 vivo tag event 失败 | tagId={} | act={} | error={}",
                    tagId, baiduAct, e.getMessage(), e);
            return List.of();
        }
    }

    private static VivoTagEventRecord buildVivoTagEvent(String tagId, int baiduAct, String eventType) {
        VivoTagEventRecord record = new VivoTagEventRecord();
        record.setTagId(tagId);
        record.setBaiduAct(baiduAct);
        record.setEventType(eventType);
        return record;
    }

    private static List<VivoTagEventRecord> buildVivoFallbackTagEvents(String tagId, int baiduAct) {
        List<String> eventTypes = switch (baiduAct) {
            case 1 -> List.of(VIVO_EVENT_ACTIVATION);
            case 2, 5 -> List.of(VIVO_EVENT_PAY);
            case 3 -> List.of(VIVO_EVENT_REGISTER, VIVO_EVENT_ACTIVATION);
            case 4 -> List.of(VIVO_EVENT_RETENTION_1);
            case 6, 2001 -> List.of(VIVO_EVENT_REACTIVATION);
            case 7 -> List.of(VIVO_EVENT_OTHER);
            case 8 -> List.of(VIVO_EVENT_RETENTION_7);
            case 9 -> List.of(VIVO_EVENT_RETENTION_3);
            default -> List.of();
        };
        if (eventTypes.isEmpty()) {
            return List.of();
        }
        List<VivoTagEventRecord> records = new ArrayList<>(eventTypes.size());
        for (String eventType : eventTypes) {
            records.add(buildVivoTagEvent(tagId, baiduAct, eventType));
        }
        return records;
    }

    // ─── 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));
        existing.addAll(collectVivoSuccessfulCallbackKeys(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();
        }
    }

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

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

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

        try {
            return vivoColdStore.successfulCallbackKeys(keys);
        } catch (Exception e) {
            log.error("[ConversionSync] vivo 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, task.tencentCustomAction);
            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));
        }

        if (task.isVivo()) {
            int deductionRate = resolveVivoDeductionRate(task.vivoBid);
            if (handleVivoDeduction(task)) {
                return CallbackDispatchResult.vivo(task, toCallbackRecord(task, buildDeductedVivoCallbackRecord(task, deductionRate)));
            }
            VivoMediaCallbackRecord vivoRecord = vivoClient.sendConversion(
                    task.vivoBid, task.payment, task.vivoTagEvent, true);
            vivoRecord.setDeductionRate(deductionRate);
            return CallbackDispatchResult.vivo(task, toCallbackRecord(task, vivoRecord));
        }

        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<>();
        List<VivoMediaCallbackRecord> vivoRecords = 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 if (result.isVivo()) {
                vivoRecords.add(toVivoCallbackRecord(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);
            }
        }

        if (vivoColdStore != null && !vivoRecords.isEmpty()) {
            try {
                vivoColdStore.saveMediaCallbacks(vivoRecords);
            } catch (Exception e) {
                log.error("[ConversionSync] batch save vivo callbacks failed: size={}, {}", vivoRecords.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;
    }

    private static ConversionRecord toVivoConversionRecord(PaymentInfo payment, VivoBidRecord bid) {
        ConversionRecord record = new ConversionRecord();
        record.setDedupeKey(conversionDedupeKey(payment));
        record.setQk(payment.getQk());
        record.setMedia("vivo");
        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);
    }

    static String vivoCallbackDedupeKey(PaymentInfo payment, VivoBidRecord bid, String eventType) {
        return String.format("vivo:%s:%s:%s:%s:%d:%s",
                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, task.payment.getAct());
        if (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(String.valueOf(resolveDeductionRate(task.tencentBid, task.payment.getAct())));
        record.setCreatedAt(now);
        int deductionRate = resolveDeductionRate(task.tencentBid, task.payment.getAct());
        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 (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 boolean handleVivoDeduction(CallbackTask task) {
        if (!task.isVivo() || task.vivoBid == null || vivoHotStore == null) {
            return false;
        }
        int deductionRate = resolveVivoDeductionRate(task.vivoBid);
        if (deductionRate <= 0) {
            return false;
        }
        if (!vivoHotStore.markConversionSeen(task.callbackKey)) {
            log.info("[ConversionSync] vivo 转化已见过，跳过重复扣量判断 | callbackKey={} | qk={} | accountId={}",
                    task.callbackKey, task.payment.getQk(), safe(task.vivoBid.getAccountId()));
            return false;
        }
        VivoHotStore.DeductionDecision decision = vivoHotStore.evaluateDeduction(
                task.vivoBid.getAccountId(),
                task.payment.getAct(),
                task.payment.getDate(),
                deductionRate
        );
        boolean deduct = decision.isShouldDeduct();
        log.info("[ConversionSync] vivo 转化扣量判定 | qk={} | act={} | accountId={} | tagId={} | rate={} | totalSeen={} | totalDeducted={} | callbackKey={} | deduct={}",
                task.payment.getQk(), task.payment.getAct(), safe(task.vivoBid.getAccountId()),
                safe(task.vivoBid.getTagId()), deductionRate,
                decision.getTotalSeen(), decision.getTotalDeducted(), task.callbackKey, deduct);
        return deduct;
    }

    private VivoMediaCallbackRecord buildDeductedVivoCallbackRecord(CallbackTask task, int deductionRate) {
        VivoMediaCallbackRecord record = new VivoMediaCallbackRecord();
        Instant now = Instant.now();
        record.setDedupeKey(task.callbackKey);
        record.setQk(task.payment.getQk());
        record.setTagId(task.vivoBid.getTagId());
        record.setCallbackUrl("");
        record.setEventType(task.vivoTagEvent.getEventType());
        record.setEventTimeMs(now.toEpochMilli());
        record.setPurchase(Math.max(task.payment.getPayment(), 0D));
        record.setDeductionRate(deductionRate);
        record.setDirectMatch(false);
        record.setDispatchStatus(VivoMediaCallbackRecord.DISPATCH_STATUS_DEDUCTED);
        record.setStatus(0);
        record.setOk(false);
        record.setAttempt(1);
        record.setErrorMessage("deducted_by_rate");
        record.setCreatedAt(now);
        log.info("[ConversionSync] vivo 转化扣量 | qk={} | act={} | rate={} | tagId={} | accountId={} | callbackKey={}",
                task.payment.getQk(), task.payment.getAct(), deductionRate,
                task.vivoBid.getTagId(), safe(task.vivoBid.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, int act) {
        if (bid == null || bid.getMediaParams() == null) {
            return 0;
        }
        if (act == 2001) {
            String rawActRate = firstParam(bid.getMediaParams(),
                    BidRecord.ACT_2001_DEDUCTION_RATE_KEY,
                    "act2001DeductionRate",
                    "deductionRate2001",
                    "deduction_rate_2001");
            Integer actRate = parseDeductionRate(rawActRate);
            if (actRate != null) {
                return actRate;
            }
        }
        String rawRate = bid.getMediaParams().get(BidRecord.DEDUCTION_RATE_KEY);
        Integer rate = parseDeductionRate(rawRate);
        if (rate != null) {
            return rate;
        }
        return 0;
    }

    private static Integer parseDeductionRate(String rawRate) {
        if (rawRate == null || rawRate.isBlank()) {
            return null;
        }
        try {
            return sanitizeDeductionRate(Integer.parseInt(rawRate.trim()));
        } catch (NumberFormatException ignored) {
            return null;
        }
    }

    private static String firstParam(Map<String, String> params, String... keys) {
        for (String key : keys) {
            String value = params.get(key);
            if (value != null && !value.isBlank()) {
                return value.trim();
            }
        }
        return null;
    }

    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 int resolveVivoDeductionRate(VivoBidRecord bid) {
        if (bid == null || bid.getMediaParams() == null) {
            return 0;
        }
        String rawRate = bid.getMediaParams().get(VivoBidRecord.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 VivoMediaCallbackRecord toVivoCallbackRecord(CallbackTask task, MediaCallbackRecord record) {
        VivoMediaCallbackRecord callback = new VivoMediaCallbackRecord();
        callback.setDedupeKey(record.getDedupeKey());
        callback.setQk(record.getQk());
        callback.setTagId(task.tagId);
        callback.setCallbackUrl(record.getCallbackUrl());
        callback.setEventType(record.getEventType() == 0 ? null : String.valueOf(record.getEventType()));
        if (task.vivoTagEvent != null) {
            callback.setEventType(task.vivoTagEvent.getEventType());
        }
        callback.setEventTimeMs(record.getEventTimeMs());
        callback.setPurchase(record.getPurchase());
        callback.setDeductionRate(parseVivoDeductionRate(task.vivoBid));
        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(!VivoMediaCallbackRecord.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 MediaCallbackRecord toCallbackRecord(CallbackTask task, VivoMediaCallbackRecord vivoRecord) {
        MediaCallbackRecord callbackRecord = new MediaCallbackRecord();
        callbackRecord.setDedupeKey(task.callbackKey);
        callbackRecord.setMedia("vivo");
        callbackRecord.setQk(vivoRecord.getQk());
        callbackRecord.setCallbackUrl(vivoRecord.getCallbackUrl());
        callbackRecord.setEventType(0);
        callbackRecord.setEventTimeMs(vivoRecord.getEventTimeMs());
        callbackRecord.setPurchase(vivoRecord.getPurchase());
        callbackRecord.setStatus(vivoRecord.getStatus());
        callbackRecord.setOk(vivoRecord.isOk());
        callbackRecord.setAttempt(vivoRecord.getAttempt());
        callbackRecord.setResponseBody(vivoRecord.getResponseBody());
        callbackRecord.setErrorMessage(vivoRecord.getErrorMessage());
        callbackRecord.setRequestBody(vivoRecord.getRequestBody());
        callbackRecord.setDispatchStatus(vivoRecord.getDispatchStatus());
        callbackRecord.setTrackingVersion(vivoRecord.isDirectMatch() ? "direct" : "deducted");
        callbackRecord.setCreatedAt(vivoRecord.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 int parseVivoDeductionRate(VivoBidRecord bid) {
        if (bid == null || bid.getMediaParams() == null) {
            return 0;
        }
        String rawRate = bid.getMediaParams().get(VivoBidRecord.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 final Map<String, VivoBidRecord> vivoBids;

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

    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 String tencentCustomAction;
        private final HonorBidRecord honorBid;
        private final Integer honorConversionId;
        private final KuaishouBidRecord kuaishouBid;
        private final KuaishouTagEventRecord kuaishouTagEvent;
        private final VivoBidRecord vivoBid;
        private final VivoTagEventRecord vivoTagEvent;

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

        private static CallbackTask tencent(PaymentInfo payment,
                                            BidRecord bid,
                                            String callbackKey,
                                            String actionType,
                                            String customAction) {
            return new CallbackTask("tencent", payment, bid.getTagId(), callbackKey, bid, actionType, customAction, null, null, 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, null, bid, conversionId, null, null, 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, null, bid, tagEvent, null, null);
        }

        private static CallbackTask vivo(PaymentInfo payment,
                                         VivoBidRecord bid,
                                         String callbackKey,
                                         VivoTagEventRecord tagEvent) {
            return new CallbackTask("vivo", payment, bid.getTagId(), callbackKey, null, null, null, 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 boolean isVivo() {
            return "vivo".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 static CallbackDispatchResult vivo(CallbackTask task, MediaCallbackRecord record) {
            return new CallbackDispatchResult("vivo", task, record);
        }

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

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

        private boolean isVivo() {
            return "vivo".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();
    }
}