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.tencent.TencentClient;
import com.adx.tencent.storage.RedisHotStore;
import com.adx.tencent.storage.TiDBColdStore;
import com.adx.tencent.storage.model.*;
import com.adx.tencent.tagsync.TagEventResolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 转化同步核心逻辑。
 * 定时从百度拉取转化数据，匹配腾讯 bid，通过 callback URL 回传腾讯。
 */
public class ConversionSyncService {

    private static final Logger log = LoggerFactory.getLogger(ConversionSyncService.class);
    private static final int CONCURRENCY = 10;

    private final ConversionClient baiduClient;
    private final RedisHotStore hotStore;
    private final TencentClient tencentClient;
    private final TiDBColdStore coldStore;
    private final TagEventResolver tagEventResolver;
    private final ExecutorService executor;

    public ConversionSyncService(ConversionClient baiduClient, RedisHotStore hotStore,
                                  TencentClient tencentClient, TiDBColdStore coldStore,
                                  TagEventResolver tagEventResolver) {
        this.baiduClient = baiduClient;
        this.hotStore = hotStore;
        this.tencentClient = tencentClient;
        this.coldStore = coldStore;
        this.tagEventResolver = tagEventResolver;
        this.executor = Executors.newFixedThreadPool(CONCURRENCY,
                r -> { Thread t = new Thread(r, "conv-sync-worker"); t.setDaemon(true); return t; });
    }

    /**
     * 分页拉取百度转化数据并回传腾讯。
     */
    public SyncResult syncTencentConversions(ConversionQuery query) throws Exception {
        if (query.getPageSize() <= 0) query.setPageSize(1);

        SyncResult result = new SyncResult();
        int maxPages = 100;
        for (int page = query.getPageSize(); ; page++) {
            query.setPageSize(page);
            log.info("[ConversionSync] fetching page={}", page);
            ConversionResponse response = baiduClient.queryPayments(query);
            List<PaymentInfo> data = response.getData();
            int dataSize = data != null ? data.size() : 0;
            log.info("[ConversionSync] page={} returned {} records, responsePageSize={}",
                    page, dataSize, response.getPageSize());
            if (data != null) {
                processPayments(data, result);
            }
            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) throws Exception {
        List<CompletableFuture<Void>> futures = new ArrayList<>(payments.size());
        for (PaymentInfo payment : payments) {
            futures.add(CompletableFuture.runAsync(() -> processSinglePayment(payment, result), executor));
        }
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    }

    private void processSinglePayment(PaymentInfo payment, SyncResult result) {
        result.fetched.incrementAndGet();

        // 用 qk 查找 bid
        BidRecord bid;
        try {
            bid = hotStore.findBidByQk(payment.getQk());
        } catch (Exception e) {
            log.error("[ConversionSync] findBidByQk error: qk={}, {}", payment.getQk(), e.getMessage());
            result.failed.incrementAndGet();
            return;
        }

        if (bid == null) {
            if (coldStore != null) {
                try { coldStore.saveConversion(conversionRecord(payment, null, "")); } catch (Exception ignored) {}
            }
            result.skipped.incrementAndGet();
            return;
        }


        // bid 不是腾讯，或没有设备标识
        if (!"tencent".equals(bid.getMedia())) {
            if (coldStore != null) {
                try { coldStore.saveConversion(conversionRecord(payment, bid, bid.getMedia())); } catch (Exception ignored) {}
            }
            result.skipped.incrementAndGet();
            return;
        }

        result.matched.incrementAndGet();
        log.info("[ConversionSync] 匹配到腾讯转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
                payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());
        if (coldStore != null) {
            try { coldStore.saveConversion(conversionRecord(payment, bid, bid.getMedia())); } catch (Exception ignored) {}
        }

        String actionType = null;
        if (tagEventResolver != null) {
            actionType = tagEventResolver.resolveActionType(bid.getTagId(), payment.getAct());
        }
        if (actionType == null || actionType.isBlank()) {
            actionType = tencentClient.conversionActionType(payment.getAct());
            log.warn("[ConversionSync] 广告位回传方式未配置, 回退act-map | qk={} | tagId={} | act={} | actionType={}",
                    payment.getQk(), bid.getTagId(), payment.getAct(), actionType);
        }

        // 幂等检查
        String callbackKey = callbackDedupeKey(payment, bid, actionType);
        if (coldStore != null) {
            try {
                String legacyCallbackKey = legacyCallbackDedupeKey(payment, bid.getMedia());
                boolean exists = coldStore.successfulCallbackExistsForEvent(
                        callbackKey, legacyCallbackKey, payment.getAct());
                if (exists) {
                    result.alreadySent.incrementAndGet();
                    return;
                }
            } catch (Exception e) {
                log.error("[ConversionSync] dedup check error: key={}, {}", callbackKey, e.getMessage());
            }
        }

        String platform = resolvePlatform(bid);
        MediaCallbackRecord callbackRecord = tencentClient.sendConversionEvent(bid, payment, platform, actionType);
        callbackRecord.setDedupeKey(callbackKey);

        if (callbackRecord.getStatus() == 0 && callbackRecord.getErrorMessage() != null) {
            // 网络级错误
            result.failed.incrementAndGet();
            if (coldStore != null) {
                try {
                    coldStore.saveMediaCallback(callbackRecord);
                } catch (Exception e) {
                    log.error("[ConversionSync] 保存callback记录失败: dedupeKey={}, error={}",
                            callbackKey, e.getMessage());
                }
            }
            return;
        }

        // HTTP 请求完成
        result.sent.incrementAndGet();
        if (coldStore != null) {
            try {
                coldStore.saveMediaCallback(callbackRecord);
            } catch (Exception e) {
                log.error("[ConversionSync] 保存callback记录失败: dedupeKey={}, error={}",
                        callbackKey, e.getMessage());
            }
        }
    }

    // ─── 辅助方法 ─────────────────────────────────────────────────────────────

    /**
     * 从 BidRecord 中提取平台信息。
     * 优先读取 platform 字段，其次从 mediaParams 中提取，最后根据设备标识推断。
     */
    private static String resolvePlatform(BidRecord bid) {
        // 优先使用 BidRecord 的 platform 字段
        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();
        // 有 idfa 则为 ios
        String idfa = bid.getMediaParams().get("idfa");
        if (idfa != null && !idfa.isBlank()) return "ios";
        return "android";
    }

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

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

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

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

    // ─── Result DTO ─────────────────────────────────────────────────────────

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