ConversionSyncService.java 37 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869
  1. package com.adx.tencent.conversionsync;
  2. import com.adx.tencent.baidu.ConversionClient;
  3. import com.adx.tencent.baidu.model.ConversionQuery;
  4. import com.adx.tencent.baidu.model.ConversionResponse;
  5. import com.adx.tencent.baidu.model.PaymentInfo;
  6. import com.adx.tencent.config.AppProperties;
  7. import com.adx.tencent.honor.client.HonorClient;
  8. import com.adx.tencent.honor.model.HonorBidRecord;
  9. import com.adx.tencent.honor.model.HonorMediaCallbackRecord;
  10. import com.adx.tencent.honor.service.HonorTagEventResolver;
  11. import com.adx.tencent.honor.store.HonorColdStore;
  12. import com.adx.tencent.honor.store.HonorHotStore;
  13. import com.adx.tencent.storage.RedisHotStore;
  14. import com.adx.tencent.storage.TiDBColdStore;
  15. import com.adx.tencent.storage.model.BidRecord;
  16. import com.adx.tencent.storage.model.ConversionRecord;
  17. import com.adx.tencent.storage.model.MediaCallbackRecord;
  18. import com.adx.tencent.tagsync.TagEventResolver;
  19. import com.adx.tencent.tencent.TencentClient;
  20. import org.slf4j.Logger;
  21. import org.slf4j.LoggerFactory;
  22. import java.time.Instant;
  23. import java.util.ArrayList;
  24. import java.util.HashSet;
  25. import java.util.LinkedHashSet;
  26. import java.util.List;
  27. import java.util.Map;
  28. import java.util.Set;
  29. import java.util.concurrent.CompletableFuture;
  30. import java.util.concurrent.ExecutorService;
  31. import java.util.concurrent.Executors;
  32. import java.util.concurrent.TimeUnit;
  33. import java.util.concurrent.atomic.AtomicInteger;
  34. /**
  35. * 统一转化同步:
  36. * 1. 只从百度拉一次转化
  37. * 2. 按 qk 匹配腾讯/荣耀 bid
  38. * 3. 百度转化明细统一落 tencent_baidu_conversions
  39. * 4. 媒体回传结果分别落各自表
  40. */
  41. public class ConversionSyncService {
  42. private static final Logger log = LoggerFactory.getLogger(ConversionSyncService.class);
  43. private static final int REALTIME_CONCURRENCY = 10;
  44. private static final int BACKFILL_CONCURRENCY = 16;
  45. private final ConversionClient baiduClient;
  46. private final RedisHotStore hotStore;
  47. private final HonorHotStore honorHotStore;
  48. private final TencentClient tencentClient;
  49. private final TiDBColdStore coldStore;
  50. private final HonorColdStore honorColdStore;
  51. private final HonorClient honorClient;
  52. private final TagEventResolver tagEventResolver;
  53. private final HonorTagEventResolver honorTagEventResolver;
  54. private final ExecutorService realtimeExecutor;
  55. private final ExecutorService backfillExecutor;
  56. public enum SyncMode {
  57. REALTIME,
  58. BACKFILL
  59. }
  60. public ConversionSyncService(ConversionClient baiduClient,
  61. RedisHotStore hotStore,
  62. TencentClient tencentClient,
  63. TiDBColdStore coldStore,
  64. TagEventResolver tagEventResolver,
  65. HonorTagEventResolver honorTagEventResolver,
  66. HonorHotStore honorHotStore,
  67. HonorColdStore honorColdStore,
  68. HonorClient honorClient,
  69. AppProperties props) {
  70. this.baiduClient = baiduClient;
  71. this.hotStore = hotStore;
  72. this.tencentClient = tencentClient;
  73. this.coldStore = coldStore;
  74. this.tagEventResolver = tagEventResolver;
  75. this.honorTagEventResolver = honorTagEventResolver;
  76. this.honorHotStore = honorHotStore;
  77. this.honorColdStore = honorColdStore;
  78. this.honorClient = honorClient;
  79. this.realtimeExecutor = Executors.newFixedThreadPool(
  80. REALTIME_CONCURRENCY,
  81. r -> {
  82. Thread t = new Thread(r, "conv-sync-worker");
  83. t.setDaemon(true);
  84. return t;
  85. });
  86. this.backfillExecutor = Executors.newFixedThreadPool(
  87. BACKFILL_CONCURRENCY,
  88. r -> {
  89. Thread t = new Thread(r, "conv-backfill-worker");
  90. t.setDaemon(true);
  91. return t;
  92. });
  93. }
  94. public SyncResult syncTencentConversions(ConversionQuery query) throws Exception {
  95. return syncTencentConversions(query, SyncMode.REALTIME);
  96. }
  97. public SyncResult syncTencentConversions(ConversionQuery query, SyncMode mode) throws Exception {
  98. if (query.getPageSize() <= 0) {
  99. query.setPageSize(1);
  100. }
  101. SyncResult result = new SyncResult();
  102. int maxPages = 10000;
  103. for (int page = query.getPageSize(); ; page++) {
  104. query.setPageSize(page);
  105. long pageStartNs = System.nanoTime();
  106. log.info("[ConversionSync] fetching date={}, page={}, mode={}", query.getDate(), page, mode);
  107. ConversionResponse response = baiduClient.queryPayments(query);
  108. List<PaymentInfo> data = response.getData();
  109. int dataSize = data != null ? data.size() : 0;
  110. log.info("[ConversionSync] date={}, page={} returned {} records, responsePageSize={}, mode={}",
  111. query.getDate(), page, dataSize, response.getPageSize(), mode);
  112. if (data != null) {
  113. processPayments(data, result, mode);
  114. }
  115. long pageCostMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - pageStartNs);
  116. log.info("[ConversionSync] date={}, page={} processed, cost={}ms, totalFetched={}, matched={}, sent={}, failed={}, skipped={}, alreadySent={}, mode={}",
  117. query.getDate(), page, pageCostMs,
  118. result.fetched.get(), result.matched.get(), result.sent.get(), result.failed.get(),
  119. result.skipped.get(), result.alreadySent.get(), mode);
  120. if (response.getPageSize() <= page) {
  121. break;
  122. }
  123. if (page >= maxPages) {
  124. log.warn("[ConversionSync] reached max page limit ({}), stopping", maxPages);
  125. break;
  126. }
  127. }
  128. return result;
  129. }
  130. private void processPayments(List<PaymentInfo> payments, SyncResult result, SyncMode mode) {
  131. if (payments == null || payments.isEmpty()) {
  132. return;
  133. }
  134. ExecutorService executor = mode == SyncMode.BACKFILL ? backfillExecutor : realtimeExecutor;
  135. // 1. 批量查这一页转化对应的腾讯/荣耀 bid;Redis miss 时会自动回落到冷库。
  136. BidLookup lookup = loadBidLookup(payments, result);
  137. // 2. 统一整理要落库的百度转化明细,以及要回传给媒体的任务。
  138. List<ConversionRecord> conversionRecords = new ArrayList<>();
  139. List<CallbackTask> callbackTasks = new ArrayList<>();
  140. buildRecordsAndTasks(payments, lookup, conversionRecords, callbackTasks, result);
  141. // 3. 百度转化明细统一落 tencent_baidu_conversions。
  142. saveConversions(conversionRecords);
  143. if (callbackTasks.isEmpty()) {
  144. return;
  145. }
  146. // 4. 回传前先做去重,只发送未成功回传过的任务。
  147. List<CallbackTask> pendingTasks = filterPendingCallbacks(callbackTasks, result);
  148. if (pendingTasks.isEmpty()) {
  149. return;
  150. }
  151. // 5. 并发回传媒体,并把回传结果分别落到腾讯/荣耀回调表。
  152. List<CallbackDispatchResult> dispatchResults = dispatchCallbacks(pendingTasks, executor, result);
  153. saveCallbackResults(dispatchResults);
  154. }
  155. // ─── Step 1: 加载 bid ───────────────────────────────────────────────────
  156. private BidLookup loadBidLookup(List<PaymentInfo> payments, SyncResult result) {
  157. Set<String> qks = new LinkedHashSet<>();
  158. for (PaymentInfo payment : payments) {
  159. result.fetched.incrementAndGet();
  160. if (payment.getQk() != null && !payment.getQk().isBlank()) {
  161. qks.add(payment.getQk());
  162. }
  163. }
  164. List<String> qkList = new ArrayList<>(qks);
  165. Map<String, BidRecord> tencentBids = hotStore.findBidsByQks(qkList);
  166. backfillTencentAccountIds(qkList, tencentBids);
  167. Map<String, HonorBidRecord> honorBids = honorHotStore != null
  168. ? honorHotStore.findBidsByQks(qkList)
  169. : Map.of();
  170. return new BidLookup(tencentBids, honorBids);
  171. }
  172. private void backfillTencentAccountIds(List<String> qkList, Map<String, BidRecord> tencentBids) {
  173. if (coldStore == null || qkList == null || qkList.isEmpty() || tencentBids == null || tencentBids.isEmpty()) {
  174. return;
  175. }
  176. List<String> missingAccountQks = new ArrayList<>();
  177. for (String qk : qkList) {
  178. BidRecord bid = tencentBids.get(qk);
  179. if (bid != null
  180. && "tencent".equals(bid.getMedia())
  181. && (bid.getAccountId() == null || bid.getAccountId().isBlank())) {
  182. missingAccountQks.add(qk);
  183. }
  184. }
  185. if (missingAccountQks.isEmpty()) {
  186. return;
  187. }
  188. try {
  189. for (BidRecord coldBid : coldStore.getBidsByQks(missingAccountQks)) {
  190. if (coldBid == null || coldBid.getQk() == null) {
  191. continue;
  192. }
  193. BidRecord hotBid = tencentBids.get(coldBid.getQk());
  194. if (hotBid == null || hotBid.getAccountId() != null && !hotBid.getAccountId().isBlank()) {
  195. continue;
  196. }
  197. if (coldBid.getAccountId() == null || coldBid.getAccountId().isBlank()) {
  198. continue;
  199. }
  200. hotBid.setAccountId(coldBid.getAccountId());
  201. log.info("[ConversionSync] 腾讯 bid accountId 回落冷库补齐 | qk={} | accountId={}",
  202. coldBid.getQk(), coldBid.getAccountId());
  203. }
  204. } catch (Exception e) {
  205. log.warn("[ConversionSync] 腾讯 bid accountId 回落冷库失败 | size={} | error={}",
  206. missingAccountQks.size(), e.getMessage(), e);
  207. }
  208. }
  209. // ─── Step 2: 组装转化记录与回传任务 ───────────────────────────────────────
  210. private void buildRecordsAndTasks(List<PaymentInfo> payments,
  211. BidLookup lookup,
  212. List<ConversionRecord> conversionRecords,
  213. List<CallbackTask> callbackTasks,
  214. SyncResult result) {
  215. for (PaymentInfo payment : payments) {
  216. BidRecord tencentBid = lookup.tencentBids.get(payment.getQk());
  217. HonorBidRecord honorBid = lookup.honorBids.get(payment.getQk());
  218. if (tencentBid == null && honorBid == null) {
  219. conversionRecords.add(toConversionRecord(payment, null, ""));
  220. result.skipped.incrementAndGet();
  221. continue;
  222. }
  223. boolean hasTencentBid = tencentBid != null;
  224. boolean hasHonorBid = honorBid != null;
  225. boolean validTencent = hasTencentBid && "tencent".equals(tencentBid.getMedia());
  226. boolean validHonor = hasHonorBid && "honor".equals(honorBid.getMedia());
  227. if (hasTencentBid && !validTencent) {
  228. log.warn("[ConversionSync] 腾讯 bid media 非法 | qk={} | media={}", payment.getQk(), tencentBid.getMedia());
  229. }
  230. if (hasHonorBid && !validHonor) {
  231. log.warn("[ConversionSync] 荣耀 bid media 非法 | qk={} | media={}", payment.getQk(), honorBid.getMedia());
  232. }
  233. if (validTencent && validHonor) {
  234. log.warn("[ConversionSync] qk 同时命中腾讯和荣耀,跳过分发 | qk={} | act={} | date={}",
  235. payment.getQk(), payment.getAct(), payment.getDate());
  236. result.skipped.incrementAndGet();
  237. continue;
  238. }
  239. if (validTencent) {
  240. conversionRecords.add(toConversionRecord(payment, tencentBid, "tencent"));
  241. callbackTasks.add(buildTencentCallbackTask(payment, tencentBid));
  242. result.matched.incrementAndGet();
  243. log.info("[ConversionSync] 匹配到腾讯转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
  244. payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());
  245. continue;
  246. }
  247. if (!validHonor) {
  248. conversionRecords.add(toConversionRecord(payment, hasTencentBid ? tencentBid : null, hasTencentBid ? tencentBid.getMedia() : ""));
  249. result.skipped.incrementAndGet();
  250. continue;
  251. }
  252. conversionRecords.add(toHonorConversionRecord(payment, honorBid));
  253. result.matched.incrementAndGet();
  254. log.info("[ConversionSync] 匹配到荣耀转化 | qk={} | act={} | deviceId={} | date={} | payment={}",
  255. payment.getQk(), payment.getAct(), payment.getDeviceId(), payment.getDate(), payment.getPayment());
  256. CallbackTask honorTask = buildHonorCallbackTask(payment, honorBid);
  257. if (honorTask != null) {
  258. callbackTasks.add(honorTask);
  259. }
  260. }
  261. }
  262. private CallbackTask buildTencentCallbackTask(PaymentInfo payment, BidRecord bid) {
  263. String actionType = null;
  264. if (tagEventResolver != null) {
  265. actionType = tagEventResolver.resolveActionType(bid.getTagId(), payment.getAct());
  266. }
  267. if (actionType == null || actionType.isBlank()) {
  268. actionType = tencentClient.conversionActionType(payment.getAct());
  269. log.warn("[ConversionSync] 广告位回传方式未配置, 回退act-map | qk={} | tagId={} | act={} | actionType={}",
  270. payment.getQk(), bid.getTagId(), payment.getAct(), actionType);
  271. }
  272. return CallbackTask.tencent(
  273. payment,
  274. bid,
  275. callbackDedupeKey(payment, bid, actionType),
  276. actionType
  277. );
  278. }
  279. private CallbackTask buildHonorCallbackTask(PaymentInfo payment, HonorBidRecord bid) {
  280. if (honorColdStore == null && honorTagEventResolver == null) {
  281. return null;
  282. }
  283. Integer conversionId = honorTagEventResolver != null
  284. ? honorTagEventResolver.resolveConversionId(bid.getTagId(), payment.getAct())
  285. : resolveHonorConversionIdFromDb(bid.getTagId(), payment.getAct());
  286. if (conversionId == null) {
  287. log.warn("[ConversionSync] 荣耀 conversionId 未配置 | qk={} | tagId={} | act={}",
  288. payment.getQk(), bid.getTagId(), payment.getAct());
  289. return null;
  290. }
  291. return CallbackTask.honor(
  292. payment,
  293. bid,
  294. honorCallbackDedupeKey(payment, bid, conversionId),
  295. conversionId
  296. );
  297. }
  298. private Integer resolveHonorConversionIdFromDb(String tagId, int baiduAct) {
  299. if (honorColdStore == null) {
  300. return null;
  301. }
  302. com.adx.tencent.honor.model.HonorTagEventRecord exact = honorColdStore.getTagEvent(tagId, baiduAct);
  303. if (exact != null) {
  304. return exact.getConversionId();
  305. }
  306. com.adx.tencent.honor.model.HonorTagEventRecord fallback = honorColdStore.getTagEvent(tagId, 0);
  307. return fallback != null ? fallback.getConversionId() : null;
  308. }
  309. // ─── Step 3: 统一落百度转化明细 ───────────────────────────────────────────
  310. private void saveConversions(List<ConversionRecord> records) {
  311. if (coldStore == null || records.isEmpty()) {
  312. return;
  313. }
  314. try {
  315. coldStore.saveConversions(records);
  316. } catch (Exception e) {
  317. log.error("[ConversionSync] batch save conversions failed: size={}, {}", records.size(), e.getMessage(), e);
  318. }
  319. }
  320. // ─── Step 4: 回传前去重 ─────────────────────────────────────────────────
  321. private List<CallbackTask> filterPendingCallbacks(List<CallbackTask> callbackTasks, SyncResult result) {
  322. Set<String> existing = new HashSet<>();
  323. existing.addAll(collectTencentSuccessfulCallbackKeys(callbackTasks));
  324. existing.addAll(collectHonorSuccessfulCallbackKeys(callbackTasks));
  325. List<CallbackTask> pending = new ArrayList<>(callbackTasks.size());
  326. for (CallbackTask task : callbackTasks) {
  327. if (existing.contains(task.callbackKey)) {
  328. result.alreadySent.incrementAndGet();
  329. } else {
  330. pending.add(task);
  331. }
  332. }
  333. return pending;
  334. }
  335. private Set<String> collectTencentSuccessfulCallbackKeys(List<CallbackTask> callbackTasks) {
  336. if (coldStore == null) {
  337. return Set.of();
  338. }
  339. List<String> callbackKeys = new ArrayList<>();
  340. List<String> legacyKeys = new ArrayList<>();
  341. List<Integer> eventTypes = new ArrayList<>();
  342. for (CallbackTask task : callbackTasks) {
  343. if (!task.isTencent()) {
  344. continue;
  345. }
  346. callbackKeys.add(task.callbackKey);
  347. legacyKeys.add(legacyCallbackDedupeKey(task.payment, "tencent"));
  348. eventTypes.add(task.payment.getAct());
  349. }
  350. if (callbackKeys.isEmpty()) {
  351. return Set.of();
  352. }
  353. try {
  354. return new HashSet<>(coldStore.successfulCallbackKeysWithLegacy(callbackKeys, legacyKeys, eventTypes));
  355. } catch (Exception e) {
  356. log.error("[ConversionSync] batch dedup check error: size={}, {}", callbackKeys.size(), e.getMessage(), e);
  357. return Set.of();
  358. }
  359. }
  360. private Set<String> collectHonorSuccessfulCallbackKeys(List<CallbackTask> callbackTasks) {
  361. if (honorColdStore == null) {
  362. return Set.of();
  363. }
  364. Set<String> keys = new HashSet<>();
  365. for (CallbackTask task : callbackTasks) {
  366. if (task.isHonor()) {
  367. keys.add(task.callbackKey);
  368. }
  369. }
  370. if (keys.isEmpty()) {
  371. return Set.of();
  372. }
  373. try {
  374. return honorColdStore.successfulCallbackKeys(keys);
  375. } catch (Exception e) {
  376. log.error("[ConversionSync] honor dedup check error: size={}, {}", keys.size(), e.getMessage(), e);
  377. return Set.of();
  378. }
  379. }
  380. // ─── Step 5: 并发回传媒体 ───────────────────────────────────────────────
  381. private List<CallbackDispatchResult> dispatchCallbacks(List<CallbackTask> tasks,
  382. ExecutorService executor,
  383. SyncResult result) {
  384. List<CompletableFuture<CallbackDispatchResult>> futures = new ArrayList<>(tasks.size());
  385. for (CallbackTask task : tasks) {
  386. futures.add(CompletableFuture.supplyAsync(() -> sendCallback(task), executor));
  387. }
  388. List<CallbackDispatchResult> results = new ArrayList<>(tasks.size());
  389. for (CompletableFuture<CallbackDispatchResult> future : futures) {
  390. CallbackDispatchResult dispatchResult = future.join();
  391. if (dispatchResult == null || dispatchResult.record == null) {
  392. continue;
  393. }
  394. if (dispatchResult.record.isOk()) {
  395. result.sent.incrementAndGet();
  396. } else {
  397. result.failed.incrementAndGet();
  398. }
  399. results.add(dispatchResult);
  400. }
  401. return results;
  402. }
  403. private CallbackDispatchResult sendCallback(CallbackTask task) {
  404. if (task.isTencent()) {
  405. DeductionHandling handling = handleTencentDeduction(task);
  406. if (handling == DeductionHandling.SKIP_DUPLICATE) {
  407. return null;
  408. }
  409. if (handling == DeductionHandling.DEDUCT) {
  410. MediaCallbackRecord deductedRecord = buildDeductedCallbackRecord(task);
  411. return CallbackDispatchResult.tencent(task, deductedRecord);
  412. }
  413. String platform = resolvePlatform(task.tencentBid);
  414. MediaCallbackRecord callbackRecord = tencentClient.sendConversionEvent(
  415. task.tencentBid, task.payment, platform, task.tencentActionType);
  416. callbackRecord.setDedupeKey(task.callbackKey);
  417. return CallbackDispatchResult.tencent(task, callbackRecord);
  418. }
  419. if (task.isHonor()) {
  420. HonorMediaCallbackRecord honorRecord = honorClient.sendConversion(
  421. task.honorBid, task.payment, task.honorConversionId);
  422. return CallbackDispatchResult.honor(task, toCallbackRecord(task, honorRecord));
  423. }
  424. return null;
  425. }
  426. // ─── Step 6: 落媒体回传结果 ─────────────────────────────────────────────
  427. private void saveCallbackResults(List<CallbackDispatchResult> results) {
  428. if (results.isEmpty()) {
  429. return;
  430. }
  431. List<MediaCallbackRecord> tencentRecords = new ArrayList<>();
  432. List<HonorMediaCallbackRecord> honorRecords = new ArrayList<>();
  433. for (CallbackDispatchResult result : results) {
  434. if (result.isHonor()) {
  435. honorRecords.add(toHonorCallbackRecord(result.task, result.record));
  436. } else {
  437. tencentRecords.add(result.record);
  438. }
  439. }
  440. if (coldStore != null && !tencentRecords.isEmpty()) {
  441. try {
  442. coldStore.saveMediaCallbacks(tencentRecords);
  443. } catch (Exception e) {
  444. log.error("[ConversionSync] batch save callbacks failed: size={}, {}", tencentRecords.size(), e.getMessage(), e);
  445. }
  446. }
  447. if (honorColdStore != null && !honorRecords.isEmpty()) {
  448. try {
  449. honorColdStore.saveMediaCallbacks(honorRecords);
  450. } catch (Exception e) {
  451. log.error("[ConversionSync] batch save honor callbacks failed: size={}, {}", honorRecords.size(), e.getMessage(), e);
  452. }
  453. }
  454. }
  455. // ─── 工具方法 ────────────────────────────────────────────────────────────
  456. private static String resolvePlatform(BidRecord bid) {
  457. if (bid.getPlatform() != null && !bid.getPlatform().isBlank()) {
  458. return bid.getPlatform().toLowerCase();
  459. }
  460. if (bid.getMediaParams() == null) {
  461. return "android";
  462. }
  463. String platform = bid.getMediaParams().get("baidu_platform");
  464. if (platform != null && !platform.isBlank()) {
  465. return platform.toLowerCase();
  466. }
  467. String idfa = bid.getMediaParams().get("idfa");
  468. if (idfa != null && !idfa.isBlank()) {
  469. return "ios";
  470. }
  471. return "android";
  472. }
  473. private static ConversionRecord toConversionRecord(PaymentInfo payment, BidRecord bid, String media) {
  474. ConversionRecord record = new ConversionRecord();
  475. record.setDedupeKey(conversionDedupeKey(payment));
  476. record.setQk(payment.getQk());
  477. record.setMedia(media);
  478. record.setTagId(bid != null ? bid.getTagId() : null);
  479. record.setDate(payment.getDate());
  480. record.setAppSid(payment.getAppSid());
  481. record.setCustomerName(payment.getCustomerName());
  482. record.setDeviceId(payment.getDeviceId());
  483. record.setConv(payment.getConv());
  484. record.setPayment(payment.getPayment());
  485. record.setGmv(payment.getGmv());
  486. record.setAct(payment.getAct());
  487. record.setTu(payment.getTu());
  488. record.setClkTime(payment.getClkTime());
  489. record.setCreatedAt(Instant.now());
  490. return record;
  491. }
  492. private static ConversionRecord toHonorConversionRecord(PaymentInfo payment, HonorBidRecord bid) {
  493. ConversionRecord record = new ConversionRecord();
  494. record.setDedupeKey(conversionDedupeKey(payment));
  495. record.setQk(payment.getQk());
  496. record.setMedia("honor");
  497. record.setTagId(bid != null ? bid.getTagId() : null);
  498. record.setDate(payment.getDate());
  499. record.setAppSid(payment.getAppSid());
  500. record.setCustomerName(payment.getCustomerName());
  501. record.setDeviceId(payment.getDeviceId());
  502. record.setConv(payment.getConv());
  503. record.setPayment(payment.getPayment());
  504. record.setGmv(payment.getGmv());
  505. record.setAct(payment.getAct());
  506. record.setTu(payment.getTu());
  507. record.setClkTime(payment.getClkTime());
  508. record.setCreatedAt(Instant.now());
  509. return record;
  510. }
  511. static String conversionDedupeKey(PaymentInfo payment) {
  512. return String.format("%s:%s:%s:%d",
  513. payment.getQk(),
  514. payment.getDate(),
  515. payment.getDeviceId(),
  516. payment.getAct());
  517. }
  518. static String callbackDedupeKey(PaymentInfo payment, BidRecord bid, String actionType) {
  519. return String.format("%s:%s:%s:%s:%s:%d:%s",
  520. bid.getMedia(),
  521. payment.getQk(),
  522. payment.getDate(),
  523. payment.getDeviceId(),
  524. bid.getTagId(),
  525. payment.getAct(),
  526. actionType);
  527. }
  528. static String legacyCallbackDedupeKey(PaymentInfo payment, String media) {
  529. return String.format("%s:%s:%s:%s:%d:%d",
  530. media,
  531. payment.getQk(),
  532. payment.getDate(),
  533. payment.getDeviceId(),
  534. payment.getAct(),
  535. payment.getAct());
  536. }
  537. static String honorCallbackDedupeKey(PaymentInfo payment, HonorBidRecord bid, int conversionId) {
  538. return String.format("honor:%s:%s:%s:%s:%d:%d",
  539. payment.getQk(),
  540. payment.getDate(),
  541. payment.getDeviceId(),
  542. bid.getTagId(),
  543. payment.getAct(),
  544. conversionId);
  545. }
  546. private DeductionHandling handleTencentDeduction(CallbackTask task) {
  547. if (!task.isTencent() || task.tencentBid == null) {
  548. return DeductionHandling.SEND;
  549. }
  550. int deductionRate = resolveDeductionRate(task.tencentBid);
  551. if (task.payment.getAct() == 2001 || deductionRate <= 0) {
  552. return DeductionHandling.SEND;
  553. }
  554. if (!isNewTrackingLink(task.tencentBid)) {
  555. return DeductionHandling.SEND;
  556. }
  557. if (!hotStore.markConversionSeen(task.callbackKey)) {
  558. log.info("[ConversionSync] 腾讯转化已见过,跳过后续发送 | callbackKey={} | qk={} | accountId={}",
  559. task.callbackKey, task.payment.getQk(), safe(task.tencentBid.getAccountId()));
  560. return DeductionHandling.SKIP_DUPLICATE;
  561. }
  562. RedisHotStore.DeductionDecision decision = hotStore.evaluateDeduction(
  563. task.tencentBid.getAccountId(),
  564. task.payment.getAct(),
  565. resolveTrackingVersion(task.tencentBid),
  566. task.payment.getDate(),
  567. deductionRate
  568. );
  569. boolean deduct = decision.isShouldDeduct();
  570. log.info("[ConversionSync] 腾讯转化扣量判定 | qk={} | act={} | accountId={} | version={} | rate={} | totalSeen={} | totalDeducted={} | callbackKey={} | deduct={}",
  571. task.payment.getQk(), task.payment.getAct(), safe(task.tencentBid.getAccountId()),
  572. resolveTrackingVersion(task.tencentBid), deductionRate,
  573. decision.getTotalSeen(), decision.getTotalDeducted(), task.callbackKey, deduct);
  574. return deduct ? DeductionHandling.DEDUCT : DeductionHandling.SEND;
  575. }
  576. private static boolean isNewTrackingLink(BidRecord bid) {
  577. if (bid == null || bid.getMediaParams() == null) {
  578. return false;
  579. }
  580. String flag = bid.getMediaParams().get(BidRecord.NEW_LINK_FLAG_KEY);
  581. if (flag != null && Boolean.parseBoolean(flag)) {
  582. return true;
  583. }
  584. String version = bid.getMediaParams().get(BidRecord.TRACKING_VERSION_KEY);
  585. return "v2".equalsIgnoreCase(version)
  586. || "v3".equalsIgnoreCase(version)
  587. || "v4".equalsIgnoreCase(version);
  588. }
  589. private MediaCallbackRecord buildDeductedCallbackRecord(CallbackTask task) {
  590. MediaCallbackRecord record = new MediaCallbackRecord();
  591. Instant now = Instant.now();
  592. record.setDedupeKey(task.callbackKey);
  593. record.setMedia("tencent");
  594. record.setQk(task.payment.getQk());
  595. String callbackUrl = task.tencentBid.getMediaParams() != null
  596. ? task.tencentBid.getMediaParams().get("callback")
  597. : null;
  598. record.setCallbackUrl(callbackUrl != null ? callbackUrl : "");
  599. record.setEventType(task.payment.getAct());
  600. record.setEventTimeMs(now.toEpochMilli());
  601. record.setPurchase(task.payment.getPayment());
  602. record.setStatus(0);
  603. record.setOk(false);
  604. record.setAttempt(1);
  605. record.setErrorMessage("deducted_by_new_link_rate");
  606. record.setDispatchStatus(MediaCallbackRecord.DISPATCH_STATUS_DEDUCTED);
  607. record.setTrackingVersion(resolveTrackingVersion(task.tencentBid));
  608. record.setCreatedAt(now);
  609. int deductionRate = resolveDeductionRate(task.tencentBid);
  610. log.info("[ConversionSync] 腾讯转化扣量 | qk={} | act={} | rate={} | tagId={} | accountId={} | callbackKey={}",
  611. task.payment.getQk(), task.payment.getAct(), deductionRate,
  612. task.tencentBid.getTagId(), safe(task.tencentBid.getAccountId()), task.callbackKey);
  613. return record;
  614. }
  615. private static String resolveTrackingVersion(BidRecord bid) {
  616. if (bid == null || bid.getMediaParams() == null) {
  617. return "v1";
  618. }
  619. String version = bid.getMediaParams().get(BidRecord.TRACKING_VERSION_KEY);
  620. return (version == null || version.isBlank()) ? "v1" : version;
  621. }
  622. private int resolveDeductionRate(BidRecord bid) {
  623. if (bid == null || bid.getMediaParams() == null) {
  624. return 0;
  625. }
  626. String rawRate = bid.getMediaParams().get(BidRecord.DEDUCTION_RATE_KEY);
  627. if (rawRate != null && !rawRate.isBlank()) {
  628. try {
  629. return sanitizeDeductionRate(Integer.parseInt(rawRate.trim()));
  630. } catch (NumberFormatException ignored) {
  631. }
  632. }
  633. return 0;
  634. }
  635. private static String safe(String value) {
  636. return value == null ? "" : value;
  637. }
  638. private static int sanitizeDeductionRate(int rate) {
  639. if (rate < 0) {
  640. return 0;
  641. }
  642. return Math.min(rate, 100);
  643. }
  644. private static HonorMediaCallbackRecord toHonorCallbackRecord(CallbackTask task, MediaCallbackRecord record) {
  645. HonorMediaCallbackRecord callback = new HonorMediaCallbackRecord();
  646. callback.setDedupeKey(record.getDedupeKey());
  647. callback.setQk(record.getQk());
  648. callback.setTagId(task.tagId);
  649. callback.setCallbackUrl(record.getCallbackUrl());
  650. callback.setEventType(record.getEventType());
  651. callback.setEventTimeMs(record.getEventTimeMs());
  652. callback.setPurchase(record.getPurchase());
  653. callback.setStatus(record.getStatus());
  654. callback.setOk(record.isOk());
  655. callback.setAttempt(record.getAttempt());
  656. callback.setResponseBody(record.getResponseBody());
  657. callback.setErrorMessage(record.getErrorMessage());
  658. callback.setRequestBody(record.getRequestBody());
  659. callback.setCreatedAt(record.getCreatedAt());
  660. return callback;
  661. }
  662. private static MediaCallbackRecord toCallbackRecord(CallbackTask task, HonorMediaCallbackRecord honorRecord) {
  663. MediaCallbackRecord callbackRecord = new MediaCallbackRecord();
  664. callbackRecord.setDedupeKey(task.callbackKey);
  665. callbackRecord.setMedia("honor");
  666. callbackRecord.setQk(honorRecord.getQk());
  667. callbackRecord.setCallbackUrl(honorRecord.getCallbackUrl());
  668. callbackRecord.setEventType(honorRecord.getEventType());
  669. callbackRecord.setEventTimeMs(honorRecord.getEventTimeMs());
  670. callbackRecord.setPurchase(honorRecord.getPurchase());
  671. callbackRecord.setStatus(honorRecord.getStatus());
  672. callbackRecord.setOk(honorRecord.isOk());
  673. callbackRecord.setAttempt(honorRecord.getAttempt());
  674. callbackRecord.setResponseBody(honorRecord.getResponseBody());
  675. callbackRecord.setErrorMessage(honorRecord.getErrorMessage());
  676. callbackRecord.setRequestBody(honorRecord.getRequestBody());
  677. callbackRecord.setTrackingVersion("v1");
  678. callbackRecord.setCreatedAt(honorRecord.getCreatedAt());
  679. return callbackRecord;
  680. }
  681. // ─── 内部数据结构 ────────────────────────────────────────────────────────
  682. private static class BidLookup {
  683. private final Map<String, BidRecord> tencentBids;
  684. private final Map<String, HonorBidRecord> honorBids;
  685. private BidLookup(Map<String, BidRecord> tencentBids, Map<String, HonorBidRecord> honorBids) {
  686. this.tencentBids = tencentBids;
  687. this.honorBids = honorBids;
  688. }
  689. }
  690. private static class CallbackTask {
  691. private final String media;
  692. private final PaymentInfo payment;
  693. private final String tagId;
  694. private final String callbackKey;
  695. private final BidRecord tencentBid;
  696. private final String tencentActionType;
  697. private final HonorBidRecord honorBid;
  698. private final Integer honorConversionId;
  699. private CallbackTask(String media,
  700. PaymentInfo payment,
  701. String tagId,
  702. String callbackKey,
  703. BidRecord tencentBid,
  704. String tencentActionType,
  705. HonorBidRecord honorBid,
  706. Integer honorConversionId) {
  707. this.media = media;
  708. this.payment = payment;
  709. this.tagId = tagId;
  710. this.callbackKey = callbackKey;
  711. this.tencentBid = tencentBid;
  712. this.tencentActionType = tencentActionType;
  713. this.honorBid = honorBid;
  714. this.honorConversionId = honorConversionId;
  715. }
  716. private static CallbackTask tencent(PaymentInfo payment,
  717. BidRecord bid,
  718. String callbackKey,
  719. String actionType) {
  720. return new CallbackTask("tencent", payment, bid.getTagId(), callbackKey, bid, actionType, null, null);
  721. }
  722. private static CallbackTask honor(PaymentInfo payment,
  723. HonorBidRecord bid,
  724. String callbackKey,
  725. Integer conversionId) {
  726. return new CallbackTask("honor", payment, bid.getTagId(), callbackKey, null, null, bid, conversionId);
  727. }
  728. private boolean isTencent() {
  729. return "tencent".equals(media);
  730. }
  731. private boolean isHonor() {
  732. return "honor".equals(media);
  733. }
  734. }
  735. private static class CallbackDispatchResult {
  736. private final String media;
  737. private final CallbackTask task;
  738. private final MediaCallbackRecord record;
  739. private CallbackDispatchResult(String media, CallbackTask task, MediaCallbackRecord record) {
  740. this.media = media;
  741. this.task = task;
  742. this.record = record;
  743. }
  744. private static CallbackDispatchResult tencent(CallbackTask task, MediaCallbackRecord record) {
  745. return new CallbackDispatchResult("tencent", task, record);
  746. }
  747. private static CallbackDispatchResult honor(CallbackTask task, MediaCallbackRecord record) {
  748. return new CallbackDispatchResult("honor", task, record);
  749. }
  750. private boolean isHonor() {
  751. return "honor".equals(media);
  752. }
  753. }
  754. private enum DeductionHandling {
  755. SEND,
  756. DEDUCT,
  757. SKIP_DUPLICATE
  758. }
  759. public static class SyncResult {
  760. public final AtomicInteger fetched = new AtomicInteger();
  761. public final AtomicInteger matched = new AtomicInteger();
  762. public final AtomicInteger sent = new AtomicInteger();
  763. public final AtomicInteger skipped = new AtomicInteger();
  764. public final AtomicInteger failed = new AtomicInteger();
  765. public final AtomicInteger alreadySent = new AtomicInteger();
  766. }
  767. }